refactor(sqlite): centralize verified snapshot publication (#105412)

Co-authored-by: Gio Della-Libera <giodl73@gmail.com>

src/commands/doctor-state-sqlite-compact.ts

@@ -1,14 +1,9 @@
 /** Explicit doctor maintenance for the canonical shared state SQLite database. */
 import fs from "node:fs";
-import type { DatabaseSync } from "node:sqlite";
-import {
-  createNewerSqliteSchemaVersionError,
-  readSqliteUserVersion,
-} from "../infra/sqlite-user-version.js";
 import {
+  assertOpenClawStateDatabaseForMaintenance,
   ensureOpenClawStatePermissions,
   isOpenClawStateDatabaseOpen,
-  OPENCLAW_STATE_SCHEMA_VERSION,
 } from "../state/openclaw-state-db.js";
 import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
 import {
@@ -65,7 +60,8 @@ export function runDoctorStateSqliteCompact(
   const compact = compactDoctorSqliteFile({
     afterMutation: () => ensureOpenClawStatePermissions(sqlitePath, env),
     sqlitePath,
-    validateBeforeMutation: (database) => validateCanonicalStateDatabase(database, sqlitePath),
+    validateBeforeMutation: (database) =>
+      assertOpenClawStateDatabaseForMaintenance(database, { pathname: sqlitePath }),
   });
   return {
     ...compact,
@@ -85,37 +81,3 @@ function readCanonicalStateDatabaseStat(sqlitePath: string): fs.Stats | undefine
     throw error;
   }
 }
-
-function validateCanonicalStateDatabase(database: DatabaseSync, sqlitePath: string): void {
-  const userVersion = readSqliteUserVersion(database);
-  if (userVersion > OPENCLAW_STATE_SCHEMA_VERSION) {
-    throw createNewerSqliteSchemaVersionError(
-      "OpenClaw state database",
-      sqlitePath,
-      userVersion,
-      OPENCLAW_STATE_SCHEMA_VERSION,
-    );
-  }
-  if (userVersion !== OPENCLAW_STATE_SCHEMA_VERSION) {
-    throw new Error(
-      `OpenClaw state database ${sqlitePath} uses schema version ${userVersion}; run openclaw doctor --fix before compacting it.`,
-    );
-  }
-
-  const metadata = database
-    .prepare("SELECT role, schema_version FROM schema_meta WHERE meta_key = 'primary' LIMIT 1")
-    .get() as { role?: unknown; schema_version?: unknown } | undefined;
-  if (metadata?.role !== "global") {
-    const role = typeof metadata?.role === "string" ? metadata.role : "missing";
-    throw new Error(
-      `OpenClaw state database ${sqlitePath} has schema role ${role}; expected global.`,
-    );
-  }
-  if (metadata.schema_version !== OPENCLAW_STATE_SCHEMA_VERSION) {
-    const schemaVersion =
-      typeof metadata.schema_version === "number" ? metadata.schema_version : "invalid";
-    throw new Error(
-      `OpenClaw state database ${sqlitePath} metadata schema version ${schemaVersion} does not match ${OPENCLAW_STATE_SCHEMA_VERSION}; run openclaw doctor --fix before compacting it.`,
-    );
-  }
-}

src/infra/backup-create.ts

@@ -7,7 +7,6 @@ import path from "node:path";
 import type { DatabaseSync } from "node:sqlite";
 import { pipeline } from "node:stream/promises";
 import { resolveDateTimestampMs } from "@openclaw/normalization-core/number-coercion";
-import { loadSqliteVecExtension } from "../../packages/memory-host-sdk/src/engine-storage.js";
 import {
   buildBackupArchiveBasename,
   buildBackupArchivePath,
@@ -24,8 +23,7 @@ import { resolveRuntimeServiceVersion } from "../version.js";
 import { isVolatileBackupPath } from "./backup-volatile-filter.js";
 import { formatErrorMessage } from "./errors.js";
 import { writeJson } from "./json-files.js";
-import { requireNodeSqlite } from "./node-sqlite.js";
-import { assertSqliteIntegrity } from "./sqlite-integrity.js";
+import { createVerifiedSqliteSnapshot } from "./sqlite-snapshot.js";
 
 const loadTarRuntime = createLazyRuntimeModule(() => import("tar"));
 
@@ -631,7 +629,6 @@ function tableExistsSql(db: DatabaseSync, tableName: string): boolean {
 function sanitizeGlobalStateSqliteSnapshot(db: DatabaseSync): void {
   if (tableExistsSql(db, "delivery_queue_entries")) {
     db.prepare("DELETE FROM delivery_queue_entries").run();
-    db.exec("VACUUM;");
   }
 }
 
@@ -738,7 +735,6 @@ async function createStateSqliteBackupPlan(params: {
     stateDir: params.stateDir,
     globalStateSqlitePath,
   });
-  const sqlite = requireNodeSqlite();
   const snapshots: SqliteBackupAsset[] = [];
   for (const archiveSourcePath of discovery.snapshotPaths) {
     // A discovered *.sqlite file that SQLite cannot snapshot aborts backup.
@@ -749,44 +745,21 @@ async function createStateSqliteBackupPlan(params: {
       path.resolve(archiveSourcePath) === globalStateSqlitePath
         ? await fs.realpath(archiveSourcePath)
         : archiveSourcePath;
-    const source = new sqlite.DatabaseSync(sourceDatabasePath, {
-      allowExtension: true,
-      readOnly: true,
-    });
     const sourcePath = path.join(params.tempDir, `openclaw-state-db-${snapshots.length}.sqlite`);
     try {
-      source.exec("PRAGMA busy_timeout = 30000;");
-      try {
-        // VACUUM INTO removes deleted-page remnants before the snapshot enters
-        // the archive. Load known bundled extensions, but fail closed when an
-        // owner schema needs capabilities core cannot safely reproduce.
-        await loadSqliteVecExtension({ db: source });
-        assertSqliteIntegrity(source, archiveSourcePath);
-        source.prepare("VACUUM INTO ?").run(sourcePath);
-      } catch (err) {
-        throw new Error(
-          `SQLite database cannot be compacted safely for backup: ${archiveSourcePath}. ${formatErrorMessage(err)}. The source must pass full integrity checks and VACUUM INTO with its required SQLite capabilities; raw page backup was refused because it can retain deleted data.`,
-          { cause: err },
-        );
-      }
-    } finally {
-      source.close();
-    }
-    await fs.chmod(sourcePath, 0o600);
-    const snapshot = new sqlite.DatabaseSync(sourcePath, { allowExtension: true });
-    try {
-      await loadSqliteVecExtension({ db: snapshot });
-      if (path.resolve(archiveSourcePath) === globalStateSqlitePath) {
-        sanitizeGlobalStateSqliteSnapshot(snapshot);
-      }
-      assertSqliteIntegrity(snapshot, sourcePath);
+      await createVerifiedSqliteSnapshot({
+        sourcePath: sourceDatabasePath,
+        targetPath: sourcePath,
+        transform:
+          path.resolve(archiveSourcePath) === globalStateSqlitePath
+            ? sanitizeGlobalStateSqliteSnapshot
+            : undefined,
+      });
     } catch (err) {
       throw new Error(
-        `SQLite backup snapshot failed verification for ${archiveSourcePath}: ${formatErrorMessage(err)}`,
+        `SQLite database cannot be compacted safely for backup: ${archiveSourcePath}. ${formatErrorMessage(err)}. The source must pass full integrity checks and VACUUM INTO with its required SQLite capabilities; raw page backup was refused because it can retain deleted data.`,
         { cause: err },
       );
-    } finally {
-      snapshot.close();
     }
     snapshots.push({
       sourcePath,

src/infra/sqlite-snapshot.test.ts

@@ -0,0 +1,271 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { requireNodeSqlite } from "./node-sqlite.js";
+import { createVerifiedSqliteSnapshot } from "./sqlite-snapshot.js";
+
+const tempDirs: string[] = [];
+
+async function createTempDir(): Promise<string> {
+  const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-sqlite-snapshot-"));
+  tempDirs.push(tempDir);
+  return tempDir;
+}
+
+afterEach(async () => {
+  await Promise.all(tempDirs.splice(0).map((tempDir) => fs.rm(tempDir, { recursive: true })));
+});
+
+function createUnsafeIndexDrift(sqlitePath: string): void {
+  const sqlite = requireNodeSqlite();
+  const database = new sqlite.DatabaseSync(sqlitePath);
+  try {
+    database.exec(`
+      CREATE TABLE records (
+        id INTEGER PRIMARY KEY,
+        indexed_value TEXT NOT NULL,
+        alternate_value TEXT NOT NULL
+      );
+      CREATE INDEX records_value ON records(indexed_value);
+      INSERT INTO records (indexed_value, alternate_value)
+      VALUES ('alpha', 'zeta'), ('beta', 'eta'), ('gamma', 'theta');
+    `);
+    database.enableDefensive?.(false);
+    database.exec("PRAGMA writable_schema = ON;");
+    database
+      .prepare(
+        "UPDATE sqlite_schema SET sql = 'CREATE INDEX records_value ON records(alternate_value)' WHERE name = 'records_value'",
+      )
+      .run();
+    const schemaVersion = Number(
+      Object.values(database.prepare("PRAGMA schema_version;").get() as Record<string, unknown>)[0],
+    );
+    database.exec(`PRAGMA writable_schema = OFF; PRAGMA schema_version = ${schemaVersion + 1};`);
+  } finally {
+    database.close();
+  }
+}
+
+describe("createVerifiedSqliteSnapshot", () => {
+  it("captures committed WAL state and removes deleted page contents", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const deletedValue = `deleted-secret-${"x".repeat(256)}`;
+    const sqlite = requireNodeSqlite();
+    const source = new sqlite.DatabaseSync(sourcePath);
+    try {
+      source.exec(`
+        PRAGMA journal_mode = WAL;
+        PRAGMA wal_autocheckpoint = 0;
+        PRAGMA secure_delete = OFF;
+        CREATE TABLE records (id INTEGER PRIMARY KEY, value TEXT NOT NULL);
+        PRAGMA wal_checkpoint(TRUNCATE);
+      `);
+      source.prepare("INSERT INTO records (value) VALUES (?)").run("survivor");
+      source.prepare("INSERT INTO records (value) VALUES (?)").run(deletedValue);
+      source.prepare("DELETE FROM records WHERE value = ?").run(deletedValue);
+
+      const result = await createVerifiedSqliteSnapshot({ sourcePath, targetPath });
+      expect(result).toEqual({ path: targetPath, userVersion: 0 });
+      expect((await fs.readFile(targetPath)).includes(deletedValue)).toBe(false);
+
+      const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true });
+      try {
+        expect(snapshot.prepare("SELECT value FROM records").all()).toEqual([
+          { value: "survivor" },
+        ]);
+      } finally {
+        snapshot.close();
+      }
+    } finally {
+      source.close();
+    }
+  });
+
+  it("rejects unsafe index drift and removes the failed target", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    createUnsafeIndexDrift(sourcePath);
+
+    await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow(
+      /integrity_check failed|malformed database schema/iu,
+    );
+    await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" });
+  });
+
+  it("rejects an existing target without modifying it", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const sqlite = requireNodeSqlite();
+    new sqlite.DatabaseSync(sourcePath).close();
+    await fs.writeFile(targetPath, "keep");
+
+    await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow(
+      /target already exists/u,
+    );
+    await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("keep");
+  });
+
+  it("preserves a target created while the snapshot is being prepared", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const sqlite = requireNodeSqlite();
+    new sqlite.DatabaseSync(sourcePath).close();
+
+    await expect(
+      createVerifiedSqliteSnapshot({
+        sourcePath,
+        targetPath,
+        transform: async () => {
+          await fs.writeFile(targetPath, "racer");
+        },
+      }),
+    ).rejects.toThrow(/EEXIST|already exists/iu);
+    await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("racer");
+  });
+
+  it("preserves a target replaced after hard-link publication", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const sqlite = requireNodeSqlite();
+    new sqlite.DatabaseSync(sourcePath).close();
+    const originalLink = fs.link.bind(fs);
+    const linkSpy = vi.spyOn(fs, "link").mockImplementation(async (source, target) => {
+      await originalLink(source, target);
+      await fs.unlink(target);
+      await fs.writeFile(target, "racer");
+    });
+
+    try {
+      await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow(
+        /target changed during publication/u,
+      );
+      await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("racer");
+    } finally {
+      linkSpy.mockRestore();
+    }
+  });
+
+  it("rejects a target replaced after exclusive-copy publication", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const sqlite = requireNodeSqlite();
+    new sqlite.DatabaseSync(sourcePath).close();
+    const originalOpen = fs.open.bind(fs);
+    const linkSpy = vi.spyOn(fs, "link").mockRejectedValue(
+      Object.assign(new Error("hard links unsupported"), {
+        code: "EPERM",
+      }),
+    );
+    const openSpy = vi.spyOn(fs, "open").mockImplementation(async (filePath, flags, mode) => {
+      const handle = await originalOpen(filePath, flags, mode);
+      if (path.resolve(String(filePath)) === targetPath && flags === "wx+") {
+        const originalSync = handle.sync.bind(handle);
+        vi.spyOn(handle, "sync").mockImplementationOnce(async () => {
+          await originalSync();
+          await fs.unlink(targetPath);
+          await fs.writeFile(targetPath, "racer");
+        });
+      }
+      return handle;
+    });
+
+    try {
+      await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow(
+        /target changed during publication/u,
+      );
+      await expect(fs.readFile(targetPath, "utf8")).resolves.toBe("racer");
+    } finally {
+      openSpy.mockRestore();
+      linkSpy.mockRestore();
+    }
+  });
+
+  it("syncs fallback copies before reporting success", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const sqlite = requireNodeSqlite();
+    new sqlite.DatabaseSync(sourcePath).close();
+    const originalOpen = fs.open.bind(fs);
+    const openSpy = vi.spyOn(fs, "open").mockImplementation(originalOpen);
+    const linkSpy = vi.spyOn(fs, "link").mockRejectedValue(
+      Object.assign(new Error("hard links unsupported"), {
+        code: "EPERM",
+      }),
+    );
+
+    try {
+      await createVerifiedSqliteSnapshot({ sourcePath, targetPath });
+      expect(
+        openSpy.mock.calls.some(([filePath]) => path.resolve(String(filePath)) === targetPath),
+      ).toBe(true);
+    } finally {
+      linkSpy.mockRestore();
+      openSpy.mockRestore();
+    }
+  });
+
+  it("removes its published target when final directory sync fails", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const sqlite = requireNodeSqlite();
+    new sqlite.DatabaseSync(sourcePath).close();
+    const originalOpen = fs.open.bind(fs);
+    const openSpy = vi.spyOn(fs, "open").mockImplementation(async (filePath, flags, mode) => {
+      if (path.resolve(String(filePath)) === tempDir) {
+        throw Object.assign(new Error("directory sync failed"), { code: "EIO" });
+      }
+      return await originalOpen(filePath, flags, mode);
+    });
+
+    try {
+      await expect(createVerifiedSqliteSnapshot({ sourcePath, targetPath })).rejects.toThrow(
+        /directory sync failed/u,
+      );
+      await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" });
+    } finally {
+      openSpy.mockRestore();
+    }
+  });
+
+  it("validates both the source and transformed snapshot", async () => {
+    const tempDir = await createTempDir();
+    const sourcePath = path.join(tempDir, "source.sqlite");
+    const targetPath = path.join(tempDir, "snapshot.sqlite");
+    const removedValue = `removed-secret-${"x".repeat(256)}`;
+    const sqlite = requireNodeSqlite();
+    const source = new sqlite.DatabaseSync(sourcePath);
+    source.exec("PRAGMA secure_delete = OFF; CREATE TABLE records (value TEXT NOT NULL);");
+    source.prepare("INSERT INTO records VALUES (?)").run(removedValue);
+    source.close();
+    const labels: string[] = [];
+
+    await createVerifiedSqliteSnapshot({
+      sourcePath,
+      targetPath,
+      transform: (database) => {
+        database.exec("DELETE FROM records;");
+        database.prepare("INSERT INTO records VALUES (?)").run("new");
+      },
+      validate: (_database, label) => labels.push(label),
+    });
+
+    expect(labels).toEqual([sourcePath, targetPath]);
+    expect((await fs.readFile(targetPath)).includes(removedValue)).toBe(false);
+    const snapshot = new sqlite.DatabaseSync(targetPath, { readOnly: true });
+    try {
+      expect(snapshot.prepare("SELECT value FROM records").get()).toEqual({ value: "new" });
+    } finally {
+      snapshot.close();
+    }
+  });
+});

src/infra/sqlite-snapshot.ts

@@ -0,0 +1,268 @@
+// Creates compact SQLite snapshots only after verifying both source and output.
+import type { Stats } from "node:fs";
+import fs from "node:fs/promises";
+import path from "node:path";
+import type { DatabaseSync } from "node:sqlite";
+import { loadSqliteVecExtension } from "../../packages/memory-host-sdk/src/engine-storage.js";
+import { formatErrorMessage } from "./errors.js";
+import { sameFileIdentity } from "./fs-safe-advanced.js";
+import { requireNodeSqlite } from "./node-sqlite.js";
+import { assertSqliteIntegrity } from "./sqlite-integrity.js";
+import { readSqliteUserVersion } from "./sqlite-user-version.js";
+
+export type SqliteSnapshotValidator = (database: DatabaseSync, databaseLabel: string) => void;
+
+export type CreateVerifiedSqliteSnapshotOptions = {
+  sourcePath: string;
+  targetPath: string;
+  transform?: (database: DatabaseSync) => void | Promise<void>;
+  validate?: SqliteSnapshotValidator;
+};
+
+export type VerifiedSqliteSnapshot = {
+  path: string;
+  userVersion: number;
+};
+
+async function assertRegularSourceFile(sourcePath: string): Promise<void> {
+  const stat = await fs.lstat(sourcePath);
+  if (!stat.isFile()) {
+    throw new Error(`SQLite snapshot source must be a regular file: ${sourcePath}`);
+  }
+}
+
+async function assertTargetAbsent(targetPath: string): Promise<void> {
+  try {
+    await fs.lstat(targetPath);
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") {
+      return;
+    }
+    throw error;
+  }
+  throw new Error(`SQLite snapshot target already exists: ${targetPath}`);
+}
+
+function isLinkFallbackError(error: unknown): boolean {
+  const code = (error as NodeJS.ErrnoException).code;
+  return (
+    code === "EPERM" ||
+    code === "EXDEV" ||
+    code === "ENOTSUP" ||
+    code === "EOPNOTSUPP" ||
+    code === "ENOSYS"
+  );
+}
+
+async function publishSnapshotNoOverwrite(
+  stagedPath: string,
+  targetPath: string,
+  stagedIdentity: Stats,
+): Promise<Stats> {
+  try {
+    await fs.link(stagedPath, targetPath);
+    return stagedIdentity;
+  } catch (error) {
+    if (!isLinkFallbackError(error)) {
+      throw error;
+    }
+    return await copyFileExclusive(stagedPath, targetPath);
+  }
+}
+
+async function copyFileExclusive(sourcePath: string, targetPath: string): Promise<Stats> {
+  const source = await fs.open(sourcePath, "r");
+  let target: Awaited<ReturnType<typeof fs.open>> | undefined;
+  let targetIdentity: Stats | undefined;
+  try {
+    target = await fs.open(targetPath, "wx+", 0o600);
+    targetIdentity = await target.stat();
+    const buffer = Buffer.allocUnsafe(1024 * 1024);
+    let offset = 0;
+    while (true) {
+      const { bytesRead } = await source.read(buffer, 0, buffer.length, offset);
+      if (bytesRead === 0) {
+        break;
+      }
+      let bytesWritten = 0;
+      while (bytesWritten < bytesRead) {
+        const result = await target.write(
+          buffer,
+          bytesWritten,
+          bytesRead - bytesWritten,
+          offset + bytesWritten,
+        );
+        if (result.bytesWritten === 0) {
+          throw new Error(`SQLite snapshot copy made no progress: ${targetPath}`);
+        }
+        bytesWritten += result.bytesWritten;
+      }
+      offset += bytesRead;
+    }
+    await target.sync();
+    return targetIdentity;
+  } catch (error) {
+    if (targetIdentity) {
+      await target?.close().catch(() => undefined);
+      target = undefined;
+      await removePublishedTargetIfOwned(targetPath, targetIdentity);
+    }
+    throw error;
+  } finally {
+    await target?.close().catch(() => undefined);
+    await source.close().catch(() => undefined);
+  }
+}
+
+async function syncFile(filePath: string): Promise<void> {
+  const handle = await fs.open(filePath, "r+");
+  try {
+    await handle.sync();
+  } finally {
+    await handle.close();
+  }
+}
+
+async function syncPublishedFile(filePath: string, expectedIdentity: Stats): Promise<void> {
+  const handle = await fs.open(filePath, "r+");
+  try {
+    const openedIdentity = await handle.stat();
+    if (!sameFileIdentity(expectedIdentity, openedIdentity)) {
+      throw new Error(`SQLite snapshot target changed before sync: ${filePath}`);
+    }
+    await handle.sync();
+    const currentIdentity = await fs.lstat(filePath);
+    if (!sameFileIdentity(expectedIdentity, currentIdentity)) {
+      throw new Error(`SQLite snapshot target changed during sync: ${filePath}`);
+    }
+  } finally {
+    await handle.close();
+  }
+}
+
+async function removePublishedTargetIfOwned(
+  filePath: string,
+  expectedIdentity: Stats,
+): Promise<void> {
+  const currentIdentity = await fs.lstat(filePath).catch(() => undefined);
+  if (currentIdentity && sameFileIdentity(expectedIdentity, currentIdentity)) {
+    await fs.unlink(filePath).catch(() => undefined);
+  }
+}
+
+function isUnsupportedDirectorySyncError(error: unknown): boolean {
+  const code = (error as NodeJS.ErrnoException).code;
+  return (
+    code === "EINVAL" ||
+    code === "ENOTSUP" ||
+    code === "ENOSYS" ||
+    (process.platform === "win32" && (code === "EISDIR" || code === "EPERM" || code === "EACCES"))
+  );
+}
+
+async function syncDirectoryBestEffort(directoryPath: string): Promise<void> {
+  const handle = await fs.open(directoryPath, "r").catch((error: unknown) => {
+    if (isUnsupportedDirectorySyncError(error)) {
+      return undefined;
+    }
+    throw error;
+  });
+  if (!handle) {
+    return;
+  }
+  try {
+    await handle.sync();
+  } catch (error) {
+    if (!isUnsupportedDirectorySyncError(error)) {
+      throw error;
+    }
+  } finally {
+    await handle.close();
+  }
+}
+
+/**
+ * Compact one SQLite database into a fresh private file and verify the result.
+ *
+ * The source and output both receive full structural, index, and foreign-key
+ * checks. Only a fully verified, synced snapshot is published to the target.
+ */
+export async function createVerifiedSqliteSnapshot(
+  options: CreateVerifiedSqliteSnapshotOptions,
+): Promise<VerifiedSqliteSnapshot> {
+  await assertRegularSourceFile(options.sourcePath);
+  await assertTargetAbsent(options.targetPath);
+
+  const stagingDir = await fs.mkdtemp(
+    path.join(path.dirname(options.targetPath), ".sqlite-snapshot-"),
+  );
+  await fs.chmod(stagingDir, 0o700);
+  const stagedPath = path.join(stagingDir, "database.sqlite");
+  const sqlite = requireNodeSqlite();
+  let stagedIdentity: Stats | undefined;
+  let publishedIdentity: Stats | undefined;
+  try {
+    const source = new sqlite.DatabaseSync(options.sourcePath, {
+      allowExtension: true,
+      readOnly: true,
+    });
+    try {
+      source.exec("PRAGMA busy_timeout = 30000; PRAGMA trusted_schema = OFF;");
+      await loadSqliteVecExtension({ db: source });
+      assertSqliteIntegrity(source, options.sourcePath);
+      options.validate?.(source, options.sourcePath);
+      source.prepare("VACUUM INTO ?").run(stagedPath);
+    } finally {
+      source.close();
+    }
+
+    await fs.chmod(stagedPath, 0o600);
+    const snapshot = new sqlite.DatabaseSync(stagedPath, { allowExtension: true });
+    try {
+      snapshot.exec("PRAGMA busy_timeout = 30000; PRAGMA trusted_schema = OFF;");
+      await loadSqliteVecExtension({ db: snapshot });
+      if (options.transform) {
+        await options.transform(snapshot);
+        // A transform may delete sensitive rows. Compact again so the
+        // published artifact cannot retain their bytes in free pages.
+        snapshot.exec("VACUUM;");
+      }
+      assertSqliteIntegrity(snapshot, options.targetPath);
+      options.validate?.(snapshot, options.targetPath);
+      const userVersion = readSqliteUserVersion(snapshot);
+      snapshot.close();
+      await syncFile(stagedPath);
+      stagedIdentity = await fs.lstat(stagedPath);
+      publishedIdentity = await publishSnapshotNoOverwrite(
+        stagedPath,
+        options.targetPath,
+        stagedIdentity,
+      );
+      const currentIdentity = await fs.lstat(options.targetPath);
+      if (!currentIdentity.isFile()) {
+        throw new Error(`SQLite snapshot target is not a regular file: ${options.targetPath}`);
+      }
+      if (!sameFileIdentity(publishedIdentity, currentIdentity)) {
+        throw new Error(`SQLite snapshot target changed during publication: ${options.targetPath}`);
+      }
+      await syncPublishedFile(options.targetPath, publishedIdentity);
+      await syncDirectoryBestEffort(path.dirname(options.targetPath));
+      return { path: options.targetPath, userVersion };
+    } finally {
+      if (snapshot.isOpen) {
+        snapshot.close();
+      }
+    }
+  } catch (error) {
+    const ownedIdentity = publishedIdentity ?? stagedIdentity;
+    if (ownedIdentity) {
+      await removePublishedTargetIfOwned(options.targetPath, ownedIdentity);
+    }
+    throw new Error(
+      `SQLite database cannot be snapshotted safely: ${options.sourcePath}. ${formatErrorMessage(error)}`,
+      { cause: error },
+    );
+  } finally {
+    await fs.rm(stagingDir, { force: true, recursive: true }).catch(() => undefined);
+  }
+}

src/state/openclaw-state-db.ts

@@ -99,6 +99,44 @@ function assertSupportedSchemaVersion(db: DatabaseSync, pathname: string): void
   }
 }
 
+/** Require the canonical shared-state owner and schema before offline file maintenance. */
+export function assertOpenClawStateDatabaseForMaintenance(
+  database: DatabaseSync,
+  options: { pathname: string },
+): void {
+  const userVersion = readSqliteUserVersion(database);
+  if (userVersion > OPENCLAW_STATE_SCHEMA_VERSION) {
+    throw createNewerSqliteSchemaVersionError(
+      "OpenClaw state database",
+      options.pathname,
+      userVersion,
+      OPENCLAW_STATE_SCHEMA_VERSION,
+    );
+  }
+  if (userVersion !== OPENCLAW_STATE_SCHEMA_VERSION) {
+    throw new Error(
+      `OpenClaw state database ${options.pathname} uses schema version ${userVersion}; run openclaw doctor --fix before compacting it.`,
+    );
+  }
+
+  const metadata = database
+    .prepare("SELECT role, schema_version FROM schema_meta WHERE meta_key = 'primary' LIMIT 1")
+    .get() as { role?: unknown; schema_version?: unknown } | undefined;
+  if (metadata?.role !== "global") {
+    const role = typeof metadata?.role === "string" ? metadata.role : "missing";
+    throw new Error(
+      `OpenClaw state database ${options.pathname} has schema role ${role}; expected global.`,
+    );
+  }
+  if (metadata.schema_version !== OPENCLAW_STATE_SCHEMA_VERSION) {
+    const schemaVersion =
+      typeof metadata.schema_version === "number" ? metadata.schema_version : "invalid";
+    throw new Error(
+      `OpenClaw state database ${options.pathname} metadata schema version ${schemaVersion} does not match ${OPENCLAW_STATE_SCHEMA_VERSION}; run openclaw doctor --fix before compacting it.`,
+    );
+  }
+}
+
 const stateDbLog = createSubsystemLogger("state/db");
 
 /** Targets already warned about, so chmod-less filesystems warn once per path. */