fix(tasks): make registry reads index-safe (#104990)

src/infra/sqlite-integrity.ts

@@ -18,12 +18,25 @@ export function assertSqliteIntegrity(
   };
 }
 
+/** Require table and associated index consistency before trusting indexed reads. */
+export function assertSqliteTableIntegrity(
+  database: DatabaseSync,
+  databaseLabel: string,
+  tableName: string,
+): void {
+  runSqliteCheck(database, `${databaseLabel} table ${tableName}`, "integrity_check", tableName);
+}
+
 function runSqliteCheck(
   database: DatabaseSync,
   databaseLabel: string,
   pragma: SqliteCheckPragma,
+  tableName?: string,
 ): "ok" {
-  const rows = database.prepare(`PRAGMA ${pragma};`).all() as Array<Record<string, unknown>>;
+  const argument = tableName ? `('${tableName.replaceAll("'", "''")}')` : "";
+  const rows = database.prepare(`PRAGMA ${pragma}${argument};`).all() as Array<
+    Record<string, unknown>
+  >;
   const results = rows.map((row) => row[pragma] ?? Object.values(row)[0]);
   if (results.length === 1 && results[0] === "ok") {
     return "ok";

src/infra/sqlite-transaction.test.ts

@@ -1,18 +1,20 @@
 // Covers synchronous SQLite transaction helpers.
 import { spawn, type ChildProcessByStdio } from "node:child_process";
-import fs from "node:fs";
-import os from "node:os";
 import path from "node:path";
 import type { Readable } from "node:stream";
 import { afterEach, describe, expect, it, vi } from "vitest";
+import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
 import { requireNodeSqlite } from "./node-sqlite.js";
-import { runSqliteImmediateTransactionSync } from "./sqlite-transaction.js";
+import {
+  runSqliteDeferredTransactionSync,
+  runSqliteImmediateTransactionSync,
+} from "./sqlite-transaction.js";
 
 const openDatabases: Array<import("node:sqlite").DatabaseSync> = [];
 type WriterLockChild = ChildProcessByStdio<null, Readable, Readable>;
 
 const openChildren: WriterLockChild[] = [];
-const tempDirs: string[] = [];
+const tempDirs = useAutoCleanupTempDirTracker(afterEach);
 
 function createDatabase(): import("node:sqlite").DatabaseSync {
   const { DatabaseSync } = requireNodeSqlite();
@@ -92,12 +94,38 @@ afterEach(() => {
       db.close();
     }
   }
-  for (const tempDir of tempDirs.splice(0)) {
-    fs.rmSync(tempDir, { recursive: true, force: true });
-  }
   vi.restoreAllMocks();
 });
 
+describe("runSqliteDeferredTransactionSync", () => {
+  it("keeps multiple reads on one snapshot while another connection commits", () => {
+    const tempDir = tempDirs.make("openclaw-sqlite-read-snapshot-");
+    const databasePath = path.join(tempDir, "snapshot.sqlite");
+    const { DatabaseSync } = requireNodeSqlite();
+    const reader = new DatabaseSync(databasePath);
+    const writer = new DatabaseSync(databasePath);
+    openDatabases.push(reader, writer);
+    reader.exec(
+      "PRAGMA journal_mode = WAL; CREATE TABLE entries (id TEXT PRIMARY KEY); INSERT INTO entries VALUES ('first');",
+    );
+    writer.exec("PRAGMA busy_timeout = 1000;");
+
+    const counts = runSqliteDeferredTransactionSync(reader, () => {
+      const before = reader.prepare("SELECT COUNT(*) AS count FROM entries").get() as {
+        count: number;
+      };
+      writer.prepare("INSERT INTO entries(id) VALUES (?)").run("second");
+      const after = reader.prepare("SELECT COUNT(*) AS count FROM entries").get() as {
+        count: number;
+      };
+      return [before.count, after.count];
+    });
+
+    expect(counts).toEqual([1, 1]);
+    expect(writer.prepare("SELECT COUNT(*) AS count FROM entries").get()).toEqual({ count: 2 });
+  });
+});
+
 describe("runSqliteImmediateTransactionSync", () => {
   it("keeps outer writes when a nested savepoint rolls back", () => {
     const db = createDatabase();
@@ -257,8 +285,7 @@ describe("runSqliteImmediateTransactionSync", () => {
 
   it("waits for a separate writer and exposes the synchronous event-loop cost", async () => {
     const holdMs = 200;
-    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-contention-"));
-    tempDirs.push(tempDir);
+    const tempDir = tempDirs.make("openclaw-sqlite-contention-");
     const databasePath = path.join(tempDir, "contention.sqlite");
     const { DatabaseSync } = requireNodeSqlite();
     const db = new DatabaseSync(databasePath);
@@ -293,8 +320,7 @@ describe("runSqliteImmediateTransactionSync", () => {
   });
 
   it("fails a real separate-writer wait after the single SQLite busy timeout", async () => {
-    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-timeout-"));
-    tempDirs.push(tempDir);
+    const tempDir = tempDirs.make("openclaw-sqlite-timeout-");
     const databasePath = path.join(tempDir, "contention.sqlite");
     const { DatabaseSync } = requireNodeSqlite();
     const db = new DatabaseSync(databasePath);

src/infra/sqlite-transaction.ts

@@ -25,6 +25,7 @@ export type SqliteTransactionOptions = {
 };
 
 type SqliteTransactionStep = "begin" | "commit";
+type SqliteTransactionMode = "deferred" | "immediate";
 
 function nextSavepointName(): string {
   nextSavepointId += 1;
@@ -164,14 +165,15 @@ function execTimedTransactionStep(params: {
   }
 }
 
-function beginImmediateTransaction(
+function beginTransaction(
   db: DatabaseSync,
   options: SqliteTransactionOptions | undefined,
+  mode: SqliteTransactionMode,
 ): void {
   execTimedTransactionStep({
     db,
     options,
-    sql: "BEGIN IMMEDIATE",
+    sql: mode === "immediate" ? "BEGIN IMMEDIATE" : "BEGIN",
     step: "begin",
   });
 }
@@ -214,9 +216,10 @@ function setTransactionDepth(db: DatabaseSync, depth: number): void {
   transactionDepthByDatabase.set(db, depth);
 }
 
-export function runSqliteImmediateTransactionSync<T>(
+function runSqliteTransactionSync<T>(
   db: DatabaseSync,
   operation: () => T,
+  mode: SqliteTransactionMode,
   options?: SqliteTransactionOptions,
 ): T {
   const depth = getTransactionDepth(db);
@@ -241,7 +244,7 @@ export function runSqliteImmediateTransactionSync<T>(
     }
   }
 
-  beginImmediateTransaction(db, options);
+  beginTransaction(db, options, mode);
   setTransactionDepth(db, 1);
   let transactionStillActive = true;
   let result: T;
@@ -285,3 +288,20 @@ export function runSqliteImmediateTransactionSync<T>(
     }
   }
 }
+
+/** Run synchronous reads against one deferred SQLite snapshot. */
+export function runSqliteDeferredTransactionSync<T>(
+  db: DatabaseSync,
+  operation: () => T,
+  options?: SqliteTransactionOptions,
+): T {
+  return runSqliteTransactionSync(db, operation, "deferred", options);
+}
+
+export function runSqliteImmediateTransactionSync<T>(
+  db: DatabaseSync,
+  operation: () => T,
+  options?: SqliteTransactionOptions,
+): T {
+  return runSqliteTransactionSync(db, operation, "immediate", options);
+}

src/tasks/task-registry.store.sqlite.ts

@@ -2,7 +2,9 @@
 import type { DatabaseSync } from "node:sqlite";
 import type { Insertable, Selectable } from "kysely";
 import { executeSqliteQuerySync, getNodeSqliteKysely } from "../infra/kysely-sync.js";
+import { assertSqliteTableIntegrity } from "../infra/sqlite-integrity.js";
 import { normalizeSqliteNumber } from "../infra/sqlite-number.js";
+import { runSqliteDeferredTransactionSync } from "../infra/sqlite-transaction.js";
 import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
 import {
   closeOpenClawStateDatabase,
@@ -216,6 +218,20 @@ function selectTaskRows(db: DatabaseSync): TaskRegistryRow[] {
   return executeSqliteQuerySync(db, query).rows;
 }
 
+function selectTaskRowsByOwnerKey(db: DatabaseSync, ownerKey: string): TaskRegistryRow[] {
+  const selectColumns = TASK_RUN_SELECT_COLUMNS.join(", ");
+  // This lookup gates duplicate media tasks. A table scan is intentional so a
+  // stale secondary index cannot hide an existing task between integrity checks.
+  return db
+    .prepare(
+      `SELECT ${selectColumns}
+       FROM task_runs NOT INDEXED
+       WHERE owner_key = ?
+       ORDER BY created_at ASC, task_id ASC`,
+    )
+    .all(ownerKey) as TaskRegistryRow[];
+}
+
 function selectTaskDeliveryStateRows(db: DatabaseSync): TaskDeliveryStateRow[] {
   const query = getTaskRegistryKysely(db)
     .selectFrom("task_delivery_state")
@@ -316,13 +332,19 @@ function withWriteTransaction(write: (database: TaskRegistryDatabase) => void) {
 }
 
 export function loadTaskRegistryStateFromSqlite(): TaskRegistryStoreSnapshot {
-  const { db } = openTaskRegistryDatabase();
-  const taskRows = selectTaskRows(db);
-  const deliveryRows = selectTaskDeliveryStateRows(db);
-  return {
-    tasks: new Map(taskRows.map((row) => [row.task_id, rowToTaskRecord(row)])),
-    deliveryStates: new Map(deliveryRows.map((row) => [row.task_id, rowToTaskDeliveryState(row)])),
-  };
+  const { db, path } = openTaskRegistryDatabase();
+  return runSqliteDeferredTransactionSync(db, () => {
+    assertSqliteTableIntegrity(db, path, "task_runs");
+    assertSqliteTableIntegrity(db, path, "task_delivery_state");
+    const taskRows = selectTaskRows(db);
+    const deliveryRows = selectTaskDeliveryStateRows(db);
+    return {
+      tasks: new Map(taskRows.map((row) => [row.task_id, rowToTaskRecord(row)])),
+      deliveryStates: new Map(
+        deliveryRows.map((row) => [row.task_id, rowToTaskDeliveryState(row)]),
+      ),
+    };
+  });
 }
 
 export function listTaskRegistryRecordsByOwnerKeyFromSqlite(ownerKey: string): TaskRecord[] {
@@ -331,14 +353,7 @@ export function listTaskRegistryRecordsByOwnerKeyFromSqlite(ownerKey: string): T
     return [];
   }
   const { db } = openTaskRegistryDatabase();
-  const query = getTaskRegistryKysely(db)
-    .selectFrom("task_runs")
-    .select(TASK_RUN_SELECT_COLUMNS)
-    .where("owner_key", "=", key)
-    .orderBy("created_at", "asc")
-    .orderBy("task_id", "asc");
-  const rows = executeSqliteQuerySync(db, query).rows;
-  return rows.map(rowToTaskRecord);
+  return selectTaskRowsByOwnerKey(db, key).map(rowToTaskRecord);
 }
 
 export function saveTaskRegistryStateToSqlite(snapshot: TaskRegistryStoreSnapshot) {

src/tasks/task-registry.store.test.ts

@@ -1,6 +1,7 @@
 // Covers task registry store persistence, in-memory behavior, and observer notifications.
 import { statSync } from "node:fs";
 import path from "node:path";
+import type { DatabaseSync } from "node:sqlite";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { emitAgentEvent } from "../infra/agent-events.js";
 import {
@@ -8,6 +9,8 @@ import {
   executeSqliteQueryTakeFirstSync,
   getNodeSqliteKysely,
 } from "../infra/kysely-sync.js";
+import { requireNodeSqlite } from "../infra/node-sqlite.js";
+import { readSqliteNumberPragma } from "../infra/sqlite-pragma.test-support.js";
 import { resetLogger, setLoggerOverride } from "../logging/logger.js";
 import { loggingState } from "../logging/state.js";
 import { createWarnLogCapture } from "../logging/test-helpers/warn-log-capture.js";
@@ -111,6 +114,22 @@ function createStoredTask(): TaskRecord {
   };
 }
 
+function createUnsafeTaskOwnerIndex(database: DatabaseSync): void {
+  database.exec(`
+    DROP INDEX idx_task_runs_owner_key;
+    CREATE INDEX idx_task_runs_owner_key ON task_runs(status);
+  `);
+  database.enableDefensive?.(false);
+  database.exec("PRAGMA writable_schema = ON;");
+  database
+    .prepare(
+      "UPDATE sqlite_schema SET sql = 'CREATE INDEX idx_task_runs_owner_key ON task_runs(owner_key)' WHERE name = 'idx_task_runs_owner_key'",
+    )
+    .run();
+  const schemaVersion = readSqliteNumberPragma(database, "schema_version");
+  database.exec(`PRAGMA writable_schema = OFF; PRAGMA schema_version = ${schemaVersion + 1};`);
+}
+
 describe("task-registry store runtime", () => {
   afterEach(() => {
     ORIGINAL_ENV.restore();
@@ -317,6 +336,123 @@ describe("task-registry store runtime", () => {
     );
   });
 
+  it("loads task and delivery rows from one sqlite read snapshot", async () => {
+    await withOpenClawTestState(
+      { layout: "state-only", prefix: "openclaw-task-store-read-snapshot-" },
+      async () => {
+        resetTaskRegistryForTests();
+        const created = createTaskRecord({
+          runtime: "acp",
+          ownerKey: "agent:main:main",
+          scopeKind: "session",
+          childSessionKey: "agent:main:acp:snapshot",
+          runId: "run-read-snapshot",
+          task: "Read one task registry snapshot",
+          status: "running",
+          deliveryStatus: "pending",
+          requesterOrigin: {
+            channel: "test-channel",
+            to: "C1234567890",
+          },
+        });
+        const database = openOpenClawStateDatabase();
+        database.db
+          .prepare("UPDATE task_delivery_state SET last_notified_event_at = ? WHERE task_id = ?")
+          .run(100, created.taskId);
+        const { DatabaseSync } = requireNodeSqlite();
+        const writer = new DatabaseSync(database.path);
+        writer.exec("PRAGMA busy_timeout = 1000;");
+        const originalPrepare = database.db.prepare.bind(database.db);
+        let writerCommitted = false;
+        const prepareSpy = vi.spyOn(database.db, "prepare").mockImplementation((sql: string) => {
+          if (sql.includes('from "task_delivery_state"') && !writerCommitted) {
+            writerCommitted = true;
+            writer
+              .prepare(
+                "UPDATE task_delivery_state SET last_notified_event_at = ? WHERE task_id = ?",
+              )
+              .run(200, created.taskId);
+          }
+          return originalPrepare(sql);
+        });
+
+        try {
+          const restored = loadTaskRegistryStateFromSqlite();
+          expect(restored.deliveryStates.get(created.taskId)?.lastNotifiedEventAt).toBe(100);
+          expect(
+            writer
+              .prepare("SELECT last_notified_event_at FROM task_delivery_state WHERE task_id = ?")
+              .get(created.taskId),
+          ).toEqual({ last_notified_event_at: 200 });
+        } finally {
+          prepareSpy.mockRestore();
+          writer.close();
+        }
+      },
+    );
+  });
+
+  it("bypasses stale owner indexes for complete fresh results", async () => {
+    await withOpenClawTestState(
+      { layout: "state-only", prefix: "openclaw-task-store-owner-index-" },
+      async () => {
+        resetTaskRegistryForTests();
+        const ownerKey = "agent:main:main";
+        const target = createTaskRecord({
+          runtime: "cron",
+          ownerKey,
+          scopeKind: "session",
+          sourceId: "owner-index-target",
+          runId: "run-owner-index-target",
+          task: "Find the target owner task",
+          status: "running",
+          deliveryStatus: "not_applicable",
+          notifyPolicy: "silent",
+        });
+        createTaskRecord({
+          runtime: "cron",
+          ownerKey: "agent:other:main",
+          scopeKind: "session",
+          sourceId: "owner-index-other",
+          runId: "run-owner-index-other",
+          task: "Ignore another owner task",
+          status: "queued",
+          deliveryStatus: "not_applicable",
+          notifyPolicy: "silent",
+        });
+        expect(listFreshTasksForOwnerKey(ownerKey).map((task) => task.taskId)).toContain(
+          target.taskId,
+        );
+
+        const database = openOpenClawStateDatabase();
+        createUnsafeTaskOwnerIndex(database.db);
+        expect(database.db.prepare("PRAGMA quick_check").get()).toEqual({ quick_check: "ok" });
+        expect(database.db.prepare("PRAGMA integrity_check('task_runs')").all()).toEqual(
+          expect.arrayContaining([
+            expect.objectContaining({
+              integrity_check: expect.stringMatching(/idx_task_runs_owner_key/),
+            }),
+          ]),
+        );
+        expect(
+          database.db
+            .prepare(
+              "SELECT task_id FROM task_runs INDEXED BY idx_task_runs_owner_key WHERE owner_key = ?",
+            )
+            .all(ownerKey),
+        ).toEqual([]);
+        expect(() => loadTaskRegistryStateFromSqlite()).toThrow(
+          /integrity_check failed.*idx_task_runs_owner_key/iu,
+        );
+        expect(listFreshTasksForOwnerKey(ownerKey).map((task) => task.taskId)).toContain(
+          target.taskId,
+        );
+
+        resetTaskRegistryForTests({ persist: false });
+      },
+    );
+  });
+
   it("emits incremental observer events for restore, mutation, and delete", () => {
     const events: TaskRegistryObserverEvent[] = [];
     configureTaskRegistryRuntime({