fix(state): serialize audit writes through shared owner

2026-08-27 12:56:01 -06:00 · 2026-08-18 00:47:34 +10:00
parent 793669c8f6
commit 9531f4cf06
62 changed files with 1250 additions and 711 deletions
@@ -153,7 +153,6 @@ const rootEntries = [
  // Docker/manual E2E executables and their nested assertion/probe entrypoints.
  "scripts/e2e/*.{js,mjs,ts}!",
  "scripts/e2e/lib/**/{assertions,probe,mock-server}.{js,mjs,ts}!",
-  "src/audit/audit-event-writer.worker.ts!",
  // Loaded by URL from the SQLite lifecycle archive owner.
  "src/config/sessions/session-accessor.sqlite-archive.worker.ts!",
  "src/state/openclaw-database-verify.worker.ts!",
@@ -54,15 +54,17 @@ their 30-day expiry.
 After session work admission succeeds, OpenClaw validates and freezes
 one bounded identity envelope, immediately offers it to the existing audit
 writer queue, and continues the run without waiting for writer readiness,
-SQLite, or persistence. The worker initializes schema and HMAC-key state,
+SQLite, or persistence. The queue drain initializes schema and HMAC-key state,
 pseudonymizes raw references, constructs the immutable context, validates its
-canonical bytes, and persists it. An accepted envelope can therefore be
-temporarily unavailable to inspection while queued work finishes.
+canonical bytes, and persists it through the process-owned shared-state
+connection. Gateway runtime uses the Gateway process's cached owner; direct-local
+execution uses the direct process's cached owner. An accepted envelope can therefore be temporarily unavailable to
+inspection while queued work finishes.

-Persistence remains best-effort. Queue saturation, worker or storage failure,
-and process crashes can lose evidence; they log only a bounded operational
-warning and never abort the run. Normal Gateway and direct-local CLI shutdown
-flushes accepted work when the writer lifecycle permits, but abrupt termination
+Persistence remains best-effort. Queue saturation, storage failure, shutdown
+timeout, and process crashes can lose evidence; they log only a bounded
+operational warning and never abort the run. Normal Gateway and direct-local CLI
+shutdown flushes accepted work when the writer lifecycle permits, but abrupt termination
 can still lose queued evidence.

 When identity collection is enabled, restart recovery stores only the safe
@@ -297,8 +299,9 @@ Message records intentionally omit both.

 Execution identity contexts use the same installation-local key owner with a
 separate HMAC domain. Raw runtime, invoker, ingress-source, assurance, and grant
-references exist only in a deeply frozen, in-process worker message capped at
-16 KiB and 16 entries in each grant/assurance array. The worker replaces them with keyed
+references exist only in a deeply frozen queue payload capped at 16 KiB and 16
+entries in each grant/assurance array. A structured clone strips prototypes at
+the queue boundary. The queue drain replaces raw references with keyed
 pseudonyms before persistence; they are never stored, exported, inspected, or
 logged. Configured agent ids plus context, execution, and run ids remain
 operator-visible.
@@ -319,8 +322,9 @@ what was recorded, not as proof of what happened:
 - **Absence of a row proves nothing.** Pre-admission inbound drops, sends from
  plugin-local or direct-send paths that bypass shared durable delivery, a
  dropped admission envelope, and crash-lost queued work can leave no record.
- Writes go through a bounded background worker; worker failure or queue
-  saturation drops records and logs one operational warning.
+- Writes go through a bounded asynchronous process-owned queue; queue saturation,
+  storage failure, or a bounded shutdown timeout can drop records and logs one
+  operational warning.
 - Crash-ambiguous outbound sends are recorded as `unknown` rather than
  invented outcomes.

@@ -361,7 +365,9 @@ additive table is created lazily on first use without a schema-version bump.
 Fresh and upgraded installations do not populate identity contexts until an
 operator enables collection.
 First-use schema creation, HMAC-key access, canonical context construction, and
-all SQLite work happen in the audit worker, never in agent admission.
+SQLite persistence happen in the process-owned audit queue drain, never in
+agent admission. Lock attempts fail fast and retry asynchronously with bounded
+backoff so SQLite contention does not synchronously wait on the Gateway thread.
 Contexts are retained for 30 days and capped at 100,000 rows. Exact-execution
 inspection and run discovery never return a context, candidate, or admission
 decision after that context is older than 30 days, even if physical cleanup
@@ -388,7 +394,7 @@ first generic fact write, retains facts for 30 days, caps the table at 250,000
 rows, and prunes at most 1,024 rows per write or maintenance tick. Approval
 paths never write this table. Its facts and approval rows are authoritative for
 their recorded decisions. Delivery to the generic table uses the bounded audit
-worker and remains best-effort until persisted; approval-owner writes do not
+queue and remains best-effort until persisted; approval-owner writes do not
 depend on that queue. The activity ledger cannot recreate either source after
 loss.

@@ -884,9 +884,9 @@ shared durable delivery; run inspection merges those sources. Chunking and
 adapter fan-out are aggregated in terminal `resultCount`. Ambiguous sends reach
 a terminal only after acknowledgement, dead
 letter, or reconciliation. Plugin-local and direct-send paths that bypass those
-shared boundaries are not yet covered. The bounded worker queue is best-effort
-and may drop records on failure or saturation, so this surface is not a
-lossless compliance archive.
+shared boundaries are not yet covered. The bounded process-owned async queue is
+best-effort and may drop records on saturation, terminal persistence failure,
+or shutdown timeout, so this surface is not a lossless compliance archive.

 Recording is on by default and controlled by
 [`logging.audit.enabled`](/gateway/configuration-reference#audit). Message
@@ -113,7 +113,6 @@ const requiredPathGroups = [
  "dist/agents/compaction-planning.worker.js",
  "dist/agents/model-provider-auth.worker.js",
  "dist/agents/prepared-model-catalog.worker.js",
-  "dist/audit/audit-event-writer.worker.js",
  "dist/config/sessions/session-accessor.sqlite-archive.worker.js",
  "dist/config/sessions/session-transcript-reconcile.worker.js",
  "dist/state/openclaw-database-verify.worker.js",
@@ -1,9 +1,9 @@
 import { withTempHome as withBaseTempHome } from "openclaw/plugin-sdk/test-env";
 import { describe, expect, it } from "vitest";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
 } from "../state/openclaw-state-db.js";
 import { operatorMcpOAuthIdentity } from "./mcp-oauth-identity.js";
 import {
@@ -0,0 +1,89 @@
+import type { DatabaseSync } from "node:sqlite";
+import {
+  executeSqliteQuerySync,
+  executeSqliteQueryTakeFirstSync,
+  getNodeSqliteKysely,
+} from "../infra/kysely-sync.js";
+import { normalizeSqliteNumber } from "../infra/sqlite-number.js";
+import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
+
+type AuditDatabase = Pick<OpenClawStateKyselyDatabase, "audit_events">;
+
+export const AUDIT_EVENT_RETENTION_MS = 30 * 24 * 60 * 60_000;
+const AUDIT_EVENT_MAX_ROWS = 100_000;
+const AUDIT_EVENT_PRUNE_BATCH_ROWS = 1_024;
+// The single audit writer owns one DB handle. Invalidate on out-of-band
+// maintenance or rollback so the hot path avoids a 100k-row scan per message.
+const auditEventRowCounts = new WeakMap<DatabaseSync, number>();
+
+function getAuditKysely(db: DatabaseSync) {
+  return getNodeSqliteKysely<AuditDatabase>(db);
+}
+
+function countAuditEvents(db: DatabaseSync): number {
+  const row = executeSqliteQueryTakeFirstSync(
+    db,
+    getAuditKysely(db)
+      .selectFrom("audit_events")
+      .select((expression) => expression.fn.countAll<number>().as("count")),
+  );
+  return normalizeSqliteNumber(row?.count ?? null) ?? 0;
+}
+
+function deleteExpiredAuditEvents(db: DatabaseSync, now: number) {
+  const kysely = getAuditKysely(db);
+  const expiredSequences = kysely
+    .selectFrom("audit_events")
+    .select("sequence")
+    .where("occurred_at", "<", now - AUDIT_EVENT_RETENTION_MS)
+    .orderBy("occurred_at", "asc")
+    .orderBy("sequence", "asc")
+    .limit(AUDIT_EVENT_PRUNE_BATCH_ROWS);
+  return executeSqliteQuerySync(
+    db,
+    kysely.deleteFrom("audit_events").where("sequence", "in", expiredSequences),
+  );
+}
+
+export function pruneAuditEventsAfterInsert(db: DatabaseSync, now: number): void {
+  const kysely = getAuditKysely(db);
+  const expired = deleteExpiredAuditEvents(db, now);
+  const cachedCount = auditEventRowCounts.get(db);
+  let rowCount =
+    cachedCount === undefined
+      ? countAuditEvents(db)
+      : Math.max(0, cachedCount + 1 - Number(expired.numAffectedRows ?? 0n));
+  if (rowCount <= AUDIT_EVENT_MAX_ROWS) {
+    auditEventRowCounts.set(db, rowCount);
+    return;
+  }
+  const retainedRows = Math.max(0, AUDIT_EVENT_MAX_ROWS - AUDIT_EVENT_PRUNE_BATCH_ROWS);
+  const overflowRow = executeSqliteQueryTakeFirstSync(
+    db,
+    kysely
+      .selectFrom("audit_events")
+      .select("sequence")
+      .orderBy("sequence", "desc")
+      .offset(retainedRows)
+      .limit(1),
+  );
+  const sequenceCutoff = overflowRow ? normalizeSqliteNumber(overflowRow.sequence) : undefined;
+  if (sequenceCutoff !== undefined) {
+    const pruned = executeSqliteQuerySync(
+      db,
+      kysely.deleteFrom("audit_events").where("sequence", "<=", sequenceCutoff),
+    );
+    rowCount = Math.max(0, rowCount - Number(pruned.numAffectedRows ?? 0n));
+  }
+  auditEventRowCounts.set(db, rowCount);
+}
+
+export function pruneExpiredAuditEventsBatch(db: DatabaseSync, now: number): number {
+  const deleted = deleteExpiredAuditEvents(db, now);
+  auditEventRowCounts.delete(db);
+  return Number(deleted.numAffectedRows ?? 0n);
+}
+
+export function invalidateAuditEventRetentionCache(db: DatabaseSync): void {
+  auditEventRowCounts.delete(db);
+}
@@ -15,6 +15,12 @@ import {
  runOpenClawStateWriteTransaction,
  type OpenClawStateDatabaseOptions,
 } from "../state/openclaw-state-db.js";
+import {
+  AUDIT_EVENT_RETENTION_MS,
+  invalidateAuditEventRetentionCache,
+  pruneAuditEventsAfterInsert,
+  pruneExpiredAuditEventsBatch,
+} from "./audit-event-retention.js";
 import {
  AUDIT_EVENT_SCHEMA_VERSION,
  AUDIT_INBOUND_MESSAGE_COMPLETED_REASONS,
@@ -46,12 +52,7 @@ type AuditEventsTable = OpenClawStateKyselyDatabase["audit_events"];
 type AuditDatabase = Pick<OpenClawStateKyselyDatabase, "audit_events">;
 type AuditEventRow = Selectable<AuditEventsTable>;

-export const AUDIT_EVENT_RETENTION_MS = 30 * 24 * 60 * 60_000;
-const AUDIT_EVENT_MAX_ROWS = 100_000;
-const AUDIT_EVENT_PRUNE_BATCH_ROWS = 1_024;
-// The single audit writer owns one DB handle. Invalidate on out-of-band
-// maintenance or rollback so the hot path avoids a 100k-row scan per message.
-const auditEventRowCounts = new WeakMap<DatabaseSync, number>();
+export { AUDIT_EVENT_RETENTION_MS } from "./audit-event-retention.js";

 function getAuditKysely(db: DatabaseSync) {
  return getNodeSqliteKysely<AuditDatabase>(db);
@@ -547,60 +548,6 @@ function bindAuditEvent(db: DatabaseSync, input: AuditEventInput): Insertable<Au
  };
 }

-function countAuditEvents(db: DatabaseSync): number {
-  const kysely = getAuditKysely(db);
-  const row = executeSqliteQueryTakeFirstSync(
-    db,
-    kysely
-      .selectFrom("audit_events")
-      .select((expression) => expression.fn.countAll<number>().as("count")),
-  );
-  return normalizeSqliteNumber(row?.count ?? null) ?? 0;
-}
-
-function pruneAuditEventsAfterInsert(
-  db: DatabaseSync,
-  now: number,
-  limits: { maxRows: number; pruneBatchRows: number } = {
-    maxRows: AUDIT_EVENT_MAX_ROWS,
-    pruneBatchRows: AUDIT_EVENT_PRUNE_BATCH_ROWS,
-  },
-): void {
-  const kysely = getAuditKysely(db);
-  const expired = executeSqliteQuerySync(
-    db,
-    kysely.deleteFrom("audit_events").where("occurred_at", "<", now - AUDIT_EVENT_RETENTION_MS),
-  );
-  const cachedCount = auditEventRowCounts.get(db);
-  let rowCount =
-    cachedCount === undefined
-      ? countAuditEvents(db)
-      : Math.max(0, cachedCount + 1 - Number(expired.numAffectedRows ?? 0n));
-  if (rowCount <= limits.maxRows) {
-    auditEventRowCounts.set(db, rowCount);
-    return;
-  }
-  const retainedRows = Math.max(0, limits.maxRows - limits.pruneBatchRows);
-  const overflowRow = executeSqliteQueryTakeFirstSync(
-    db,
-    kysely
-      .selectFrom("audit_events")
-      .select("sequence")
-      .orderBy("sequence", "desc")
-      .offset(retainedRows)
-      .limit(1),
-  );
-  const sequenceCutoff = overflowRow ? normalizeSqliteNumber(overflowRow.sequence) : undefined;
-  if (sequenceCutoff !== undefined) {
-    const pruned = executeSqliteQuerySync(
-      db,
-      kysely.deleteFrom("audit_events").where("sequence", "<=", sequenceCutoff),
-    );
-    rowCount = Math.max(0, rowCount - Number(pruned.numAffectedRows ?? 0n));
-  }
-  auditEventRowCounts.set(db, rowCount);
-}
-
 /** Persist one projected event idempotently and prune fixed retention bounds. */
 export function recordAuditEvent(
  input: AuditEventInput,
@@ -650,7 +597,7 @@ export function recordAuditEvent(
    }, options);
  } catch (error) {
    if (countCacheDatabase) {
-      auditEventRowCounts.delete(countCacheDatabase);
+      invalidateAuditEventRetentionCache(countCacheDatabase);
      clearAuditIdentityKeyCacheForDatabase(countCacheDatabase);
    }
    throw error;
@@ -720,20 +667,14 @@ export function listAuditEvents(params: {
  };
 }

-/** Delete expired metadata during Gateway startup and periodic worker maintenance. */
+/** Delete one bounded batch during Gateway startup and periodic audit maintenance. */
 export function pruneExpiredAuditEvents(
  params: {
    now?: number;
    database?: OpenClawStateDatabaseOptions;
  } = {},
-): void {
-  runOpenClawStateWriteTransaction(({ db }) => {
-    executeSqliteQuerySync(
-      db,
-      getAuditKysely(db)
-        .deleteFrom("audit_events")
-        .where("occurred_at", "<", (params.now ?? Date.now()) - AUDIT_EVENT_RETENTION_MS),
-    );
-    auditEventRowCounts.delete(db);
+): number {
+  return runOpenClawStateWriteTransaction(({ db }) => {
+    return pruneExpiredAuditEventsBatch(db, params.now ?? Date.now());
  }, params.database);
 }
@@ -0,0 +1,86 @@
+import path from "node:path";
+import { afterEach, describe, expect, it } from "vitest";
+import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
+import {
+  initializeCachedClawInstallSchemaVersions,
+  readCachedClawInstallSchemaVersions,
+} from "../claws/provenance-runtime-read.js";
+import { resolveGatewayLockDir } from "../config/paths.js";
+import { resolvePathViaExistingAncestorSync } from "../infra/boundary-path.js";
+import { sha256HexPrefixCore } from "../infra/crypto-digest.js";
+import { tryAcquireExclusiveSqliteCoordinator } from "../infra/node-sqlite.js";
+import {
+  closeOpenClawStateDatabaseForTest,
+  openOpenClawStateDatabase,
+} from "../state/openclaw-state-db.js";
+import { resolveOpenClawStateDirForDatabasePath } from "../state/openclaw-state-db.paths.js";
+import { listAuditEvents } from "./audit-event-store.js";
+import type { AuditEventInput } from "./audit-event-types.js";
+import { createAuditEventWriter } from "./audit-event-writer.js";
+
+const tempDirs = useAutoCleanupTempDirTracker((cleanup) => {
+  afterEach(() => {
+    closeOpenClawStateDatabaseForTest();
+    cleanup();
+  });
+});
+
+function input(): AuditEventInput {
+  return {
+    sourceId: "cold-owner-coordinator",
+    sourceSequence: 1,
+    occurredAt: Date.now(),
+    kind: "agent_run",
+    action: "agent.run.started",
+    status: "started",
+    actorType: "agent",
+    actorId: "main",
+    agentId: "main",
+    runId: "cold-owner-coordinator",
+  };
+}
+
+function resolveOwnershipCoordinatorPath(databasePath: string): string {
+  const canonicalDatabasePath = resolvePathViaExistingAncestorSync(databasePath);
+  const stateDir = resolveOpenClawStateDirForDatabasePath(canonicalDatabasePath);
+  return path.join(
+    resolveGatewayLockDir(stateDir),
+    `state-ownership.${sha256HexPrefixCore(canonicalDatabasePath, 8)}.lock.sqlite`,
+  );
+}
+
+describe("audit event writer cold ownership", () => {
+  it("retries a held coordinator without blocking or poisoning provenance", async () => {
+    const stateDir = tempDirs.make("openclaw-audit-writer-");
+    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    const databasePath = openOpenClawStateDatabase(database).path;
+    closeOpenClawStateDatabaseForTest();
+    initializeCachedClawInstallSchemaVersions(database);
+    expect(readCachedClawInstallSchemaVersions(database)).toMatchObject({ kind: "ready" });
+    const coordinator = tryAcquireExclusiveSqliteCoordinator(
+      resolveOwnershipCoordinatorPath(databasePath),
+      { busyTimeoutMs: 0 },
+    );
+    if (!coordinator) {
+      throw new Error("expected to acquire the state ownership coordinator");
+    }
+    const errors: string[] = [];
+    const startedAt = performance.now();
+    const writer = createAuditEventWriter({ stateDir, onError: (error) => errors.push(error) });
+
+    try {
+      await writer.ready;
+      expect(performance.now() - startedAt).toBeLessThan(250);
+      expect(readCachedClawInstallSchemaVersions(database)).toMatchObject({ kind: "ready" });
+      expect(writer.record(input())).toBe(true);
+    } finally {
+      coordinator.release();
+      await writer.stop();
+    }
+
+    expect(errors).toEqual([]);
+    expect(listAuditEvents({ database, limit: 10 }).events.map((event) => event.runId)).toContain(
+      "cold-owner-coordinator",
+    );
+  });
+});
@@ -1,7 +1,8 @@
-import fs from "node:fs";
-import { afterEach, describe, expect, it } from "vitest";
+import { DatabaseSync } from "node:sqlite";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import type { DecisionReceiptV1 } from "../../packages/gateway-protocol/src/index.js";
 import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
+import { readSqliteBusyTimeout } from "../infra/sqlite-busy-timeout.js";
 import { tableExists } from "../state/openclaw-state-db-schema-helpers.js";
 import {
  closeOpenClawStateDatabaseForTest,
@@ -26,7 +27,7 @@ import {
 import type { TrustedMessageAuditEvent } from "./message-audit-events.js";

 function defineObjectPrototypeProperties(descriptors: PropertyDescriptorMap): void {
-  // oxlint-disable-next-line no-extend-native -- Exercise hostile prototype pollution across the real worker boundary.
+  // oxlint-disable-next-line no-extend-native -- Exercise hostile prototype pollution across the real clone boundary.
  Object.defineProperties(Object.prototype, descriptors);
 }

@@ -144,7 +145,7 @@ afterEach(() => {
 });
 const tempDirs = useAutoCleanupTempDirTracker(afterEach);

-describe("audit event worker", () => {
+describe("audit event writer", () => {
  it("keeps progress absent while disabled and routes enabled progress off audit_events", async () => {
    const stateDir = tempDirs.make("openclaw-audit-writer-");
    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
@@ -194,6 +195,24 @@ describe("audit event worker", () => {
    ).toBe("message.outbound.finished");
  });

+  it("flushes accepted events through the gateway-owned state connection", async () => {
+    const stateDir = tempDirs.make("openclaw-audit-writer-");
+    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    const owner = openOpenClawStateDatabase(database).db;
+    const readDataVersion = () =>
+      (owner.prepare("PRAGMA data_version").get() as { data_version: number }).data_version;
+    const dataVersionBefore = readDataVersion();
+    const writer = createAuditEventWriter({ stateDir });
+
+    await writer.ready;
+    expect(writer.record(input())).toBe(true);
+    await writer.stop();
+
+    expect(readDataVersion()).toBe(dataVersionBefore);
+    expect(owner.prepare("PRAGMA quick_check").get()).toEqual({ quick_check: "ok" });
+    expect(listAuditEvents({ database, limit: 10 }).events).toHaveLength(1);
+  });
+
  it("keeps fresh storage identity-free when recovery evidence is missing", async () => {
    const stateDir = tempDirs.make("openclaw-audit-writer-");
    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
@@ -234,7 +253,7 @@ describe("audit event worker", () => {
    ).toBeUndefined();
  });

-  it("persists a generic decision through the bounded worker queue", async () => {
+  it("persists a generic decision through the bounded queue", async () => {
    const stateDir = tempDirs.make("openclaw-audit-writer-");
    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
    const errors: string[] = [];
@@ -271,6 +290,37 @@ describe("audit event worker", () => {
    ).toEqual([receipt]);
  });

+  it("keeps a cold owner open nonblocking under a held write lock", async () => {
+    const stateDir = tempDirs.make("openclaw-audit-writer-");
+    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    recordAuditEvent(input(), database);
+    const path = openOpenClawStateDatabase(database).path;
+    closeOpenClawStateDatabaseForTest();
+    const contender = new DatabaseSync(path);
+    contender.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE;");
+    const errors: string[] = [];
+    const probeStartedAt = performance.now();
+    const writer = createAuditEventWriter({ stateDir, onError: (error) => errors.push(error) });
+
+    try {
+      const eventLoopDelay = await new Promise<number>((resolve) => {
+        setTimeout(() => resolve(performance.now() - probeStartedAt), 25);
+      });
+      expect(eventLoopDelay).toBeLessThan(250);
+      await writer.ready;
+      expect(writer.record({ ...input(), sourceId: "cold-owner", runId: "cold-owner" })).toBe(true);
+    } finally {
+      contender.exec("ROLLBACK");
+      contender.close();
+      await writer.stop();
+    }
+
+    expect(errors).toEqual([]);
+    expect(listAuditEvents({ database, limit: 10 }).events.map((event) => event.runId)).toContain(
+      "cold-owner",
+    );
+  });
+
  it("keeps the shared queue nonblocking under a held write lock and flushes before stop", async () => {
    const stateDir = tempDirs.make("openclaw-audit-writer-");
    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
@@ -283,14 +333,15 @@ describe("audit event worker", () => {
      onError: (error) => errors.push(error),
    });
    await writer.ready;
-    const { db } = openOpenClawStateDatabase(database);
+    const { db, path } = openOpenClawStateDatabase(database);
    expect(
      db
        .prepare("SELECT name FROM sqlite_schema WHERE type = 'table' AND name = ?")
        .get("execution_identity_contexts"),
    ).toBeUndefined();
    db.exec("DELETE FROM audit_identity_keys;");
-    db.exec("BEGIN IMMEDIATE");
+    const contender = new DatabaseSync(path);
+    contender.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE;");
    const clearSink = configureExecutionIdentityAdmissionSink(writer.recordExecutionIdentity);
    const admittedAt = Date.now();

@@ -329,6 +380,12 @@ describe("audit event worker", () => {
        accepted: true,
      });
      expect(performance.now() - startedAt).toBeLessThan(250);
+      const eventLoopProbeStartedAt = performance.now();
+      const eventLoopDelay = await new Promise<number>((resolve) => {
+        setTimeout(() => resolve(performance.now() - eventLoopProbeStartedAt), 25);
+      });
+      expect(eventLoopDelay).toBeLessThan(250);
+      expect(readSqliteBusyTimeout(db)).toBe(5_000);
      expect(
        writer.recordExecutionIdentity({
          kind: "retry-reference",
@@ -350,7 +407,8 @@ describe("audit event worker", () => {
      });
    } finally {
      try {
-        db.exec("ROLLBACK");
+        contender.exec("ROLLBACK");
+        contender.close();
      } finally {
        clearSink();
        await writer.stop();
@@ -387,42 +445,131 @@ describe("audit event worker", () => {
    }
  });

-  it("stops without resetting the WAL owned by an active Gateway reader", async () => {
+  it("preserves FIFO ordering for dependent identity and decision retries", async () => {
    const stateDir = tempDirs.make("openclaw-audit-writer-");
    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
-    recordAuditEvent(input(), database);
-    closeOpenClawStateDatabaseForTest();
-    const errors: string[] = [];
-    const writer = createAuditEventWriter({ stateDir, onError: (error) => errors.push(error) });
+    const writer = createAuditEventWriter({ stateDir });
    await writer.ready;
-    const gateway = openOpenClawStateDatabase(database);
-    gateway.db.exec("BEGIN;");
-    gateway.db.prepare("SELECT count(*) FROM audit_events").get();
+    const { path } = openOpenClawStateDatabase(database);
+    const contender = new DatabaseSync(path);
+    contender.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE;");
+    const receipt = decisionReceipt();
+    const envelope = captureExecutionIdentityAdmissionEnvelope(
+      {
+        runId: receipt.runId,
+        agentId: "main",
+        ingress: { kind: "local-cli", boundary: "agent-command.local", state: "present" },
+        runtime: { kind: "embedded" },
+      },
+      {
+        contextId: receipt.contextId,
+        executionId: receipt.executionId,
+        runtimeInstanceId: "queue-runtime",
+        now: receipt.occurredAt,
+      },
+    );

    try {
-      expect(
-        writer.record({ ...input(), sourceId: "worker-before-stop", runId: "worker-before-stop" }),
-      ).toBe(true);
-      const stopStartedAt = performance.now();
-      await writer.stop();
-      const stopElapsedMs = performance.now() - stopStartedAt;
-
-      expect(errors).toEqual([]);
-      expect(stopElapsedMs).toBeLessThan(1_000);
-      expect(fs.statSync(`${gateway.path}-wal`).size).toBeGreaterThan(0);
+      expect(writer.recordExecutionIdentity(captureWork(envelope))).toBe(true);
+      expect(writer.recordExecutionDecision(receipt)).toBe(true);
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 50);
+      });
    } finally {
-      gateway.db.exec("ROLLBACK;");
+      contender.exec("ROLLBACK");
+      contender.close();
      await writer.stop();
    }

-    expect(gateway.db.prepare("PRAGMA quick_check").get()).toEqual({ quick_check: "ok" });
-    expect(listAuditEvents({ database, limit: 10 }).events.map((event) => event.runId)).toEqual([
-      "worker-before-stop",
-      "run-1",
-    ]);
+    expect(
+      pageExecutionDecisionFactsForContext({
+        context: receipt,
+        limit: 10,
+        now: receipt.occurredAt,
+        database,
+      }).receipts,
+    ).toEqual([receipt]);
  });

-  it("persists owned unknown and omits inherited evidence through the worker clone boundary", async () => {
+  it("bounds shutdown while a write lock remains held", async () => {
+    const stateDir = tempDirs.make("openclaw-audit-writer-");
+    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    const errors: string[] = [];
+    const writer = createAuditEventWriter({ stateDir, onError: (error) => errors.push(error) });
+    await writer.ready;
+    const { path } = openOpenClawStateDatabase(database);
+    const contender = new DatabaseSync(path);
+    contender.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE;");
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+
+    try {
+      expect(writer.record({ ...input(), sourceId: "shutdown-timeout" })).toBe(true);
+      const stopping = writer.stop();
+      await vi.advanceTimersByTimeAsync(10_000);
+      await stopping;
+      expect(errors).toContain(
+        "audit event writer shutdown timed out; pending metadata may be lost",
+      );
+    } finally {
+      vi.useRealTimers();
+      contender.exec("ROLLBACK");
+      contender.close();
+      await writer.stop();
+    }
+  });
+
+  it("reports sustained lock contention once while backing off retries", async () => {
+    const stateDir = tempDirs.make("openclaw-audit-writer-");
+    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    const contentions: string[] = [];
+    const errors: string[] = [];
+    const writer = createAuditEventWriter({
+      stateDir,
+      onContention: (message) => contentions.push(message),
+      onError: (error) => errors.push(error),
+    });
+    await writer.ready;
+    const { path } = openOpenClawStateDatabase(database);
+    const contender = new DatabaseSync(path);
+    contender.exec("PRAGMA busy_timeout = 0; BEGIN IMMEDIATE;");
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+
+    let contentionsDuringLock: string[] = [];
+    try {
+      expect(
+        writer.record({
+          ...input(),
+          sourceId: "sustained-contention",
+          runId: "sustained-contention",
+        }),
+      ).toBe(true);
+      await new Promise<void>((resolve) => {
+        setImmediate(resolve);
+      });
+      await vi.advanceTimersByTimeAsync(3_000);
+      contentionsDuringLock = [...contentions];
+      contender.exec("ROLLBACK");
+      contender.close();
+      await vi.advanceTimersByTimeAsync(1_000);
+      await writer.stop();
+    } finally {
+      vi.useRealTimers();
+      if (contender.isOpen) {
+        contender.close();
+      }
+      await writer.stop();
+    }
+
+    expect(contentionsDuringLock).toEqual([
+      "audit event persistence delayed by SQLite lock contention",
+    ]);
+    expect(errors).toEqual([]);
+    expect(listAuditEvents({ database, limit: 10 }).events.map((event) => event.runId)).toContain(
+      "sustained-contention",
+    );
+  });
+
+  it("persists owned unknown and omits inherited evidence through the queue clone boundary", async () => {
    const stateDir = tempDirs.make("openclaw-audit-writer-");
    const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
    const errors: string[] = [];
@@ -710,7 +857,7 @@ describe("audit event worker", () => {
    }
  });

-  it("keeps unavailable worker, schema, and insert failures off the admission path", async () => {
+  it("keeps schema and insert failures off the admission path", async () => {
    const envelope = captureExecutionIdentityAdmissionEnvelope(
      {
        runId: "nonblocking-failure-run",
@@ -721,18 +868,6 @@ describe("audit event worker", () => {
      { runtimeInstanceId: "runtime-1" },
    );

-    const unavailableErrors: string[] = [];
-    const unavailableWriter = createAuditEventWriter({
-      workerUrl: new URL("./missing-audit-event-writer.worker.ts", import.meta.url),
-      onError: (error) => unavailableErrors.push(error),
-    });
-    await unavailableWriter.ready;
-    const unavailableStartedAt = performance.now();
-    expect(unavailableWriter.recordExecutionIdentity(captureWork(envelope))).toBe(false);
-    expect(performance.now() - unavailableStartedAt).toBeLessThan(250);
-    await unavailableWriter.stop();
-    expect(unavailableErrors).toContain("audit event writer is unavailable; dropping metadata");
-
    const schemaStateDir = tempDirs.make("openclaw-audit-writer-");
    const schemaDatabase = { env: { OPENCLAW_STATE_DIR: schemaStateDir } };
    openOpenClawStateDatabase(schemaDatabase).db.exec(`
@@ -1,26 +1,44 @@
-/** Non-blocking worker-thread writer for Gateway audit metadata. */
-import path from "node:path";
-import { fileURLToPath, pathToFileURL } from "node:url";
-import { Worker } from "node:worker_threads";
+/** Non-blocking process-owned queue for audit metadata persistence. */
 import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
 import type { DecisionReceiptV1 } from "../../packages/gateway-protocol/src/index.js";
 import { resolveStateDir } from "../config/paths.js";
 import { redactSensitiveText } from "../logging/redact.js";
-import { OPENCLAW_SQLITE_BUSY_TIMEOUT_MS } from "../state/openclaw-state-db.js";
-import type { AuditEventInput } from "./audit-event-types.js";
+import {
+  OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+  runWithOpenClawStateBusyTimeout,
+} from "../state/openclaw-state-db.js";
+import { isOpenClawStateWriteContentionError } from "../state/openclaw-state-ownership.js";
+import { pruneExpiredAuditEvents, recordAuditEvent } from "./audit-event-store.js";
+import { isOutboundMessageProgressInput, type AuditEventInput } from "./audit-event-types.js";
+import {
+  pruneExpiredExecutionDecisionFacts,
+  recordExecutionDecisionFact,
+} from "./execution-decision-facts.js";
 import type { ExecutionIdentityAdmissionWork } from "./execution-identity-admission.js";
+import {
+  processExecutionIdentityAdmissionWork,
+  pruneExpiredExecutionIdentityContexts,
+} from "./execution-identity-context.js";
+import {
+  pruneExpiredOutboundMessageProgress,
+  recordOutboundMessageProgress,
+} from "./message-delivery-progress-store.js";

 const MAX_PENDING_AUDIT_EVENTS = 4_096;
-// The worker can be synchronously blocked inside SQLite's busy timeout. Keep
-// shutdown beyond that window so a queued stop cannot kill an accepted write.
+const AUDIT_MAINTENANCE_INTERVAL_MS = 60 * 60_000;
+const AUDIT_LOCK_RETRY_DELAY_MS = 25;
+const AUDIT_LOCK_RETRY_MAX_DELAY_MS = 1_000;
+const AUDIT_LOCK_CONTENTION_REPORT_MS = 1_000;
+// Bound retries during shutdown without blocking the Gateway event loop.
 const AUDIT_WRITER_SHUTDOWN_TIMEOUT_MS = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS + 5_000;

-type AuditWriterMessage =
-  | { type: "ready" }
-  | { type: "recorded" }
-  | { type: "record-error"; error: string }
-  | { type: "maintenance-error"; error: string }
-  | { type: "stopped" };
+type AuditWriterAttempt = "settled" | "retry";
+type AuditMaintenanceAttempt = "settled" | "more" | "retry";
+
+type AuditWriterRequest =
+  | { type: "record-event"; input: AuditEventInput }
+  | { type: "record-execution-identity"; work: ExecutionIdentityAdmissionWork }
+  | { type: "record-execution-decision"; receipt: DecisionReceiptV1 };

 export type AuditEventWriter = {
  ready: Promise<void>;
@@ -39,64 +57,129 @@ function formatAuditWriterError(error: unknown): string {
  );
 }

-function resolveAuditEventWriterUrl(currentModuleUrl = import.meta.url): URL {
-  const currentPath = fileURLToPath(currentModuleUrl);
-  const normalized = currentPath.replaceAll(path.sep, "/");
-  const distMarker = "/dist/";
-  const distIndex = normalized.lastIndexOf(distMarker);
-  if (distIndex >= 0) {
-    const distRoot = currentPath.slice(0, distIndex + distMarker.length);
-    return pathToFileURL(path.join(distRoot, "audit", "audit-event-writer.worker.js"));
+function executionIdentityFailureMessage(error: unknown): string {
+  const message = error instanceof Error ? error.message : String(error);
+  if (
+    message.includes("audit identity key is missing") ||
+    message.includes("audit identity key is corrupt")
+  ) {
+    return "audit execution identity key unavailable";
  }
-  const extension = path.extname(currentPath) || ".js";
-  return new URL(`./audit-event-writer.worker${extension}`, currentModuleUrl);
+  if (message.includes("execution identity context conflict")) {
+    return "audit execution identity context conflict";
+  }
+  if (message.includes("execution identity recovery evidence unavailable")) {
+    return "audit execution identity recovery evidence unavailable";
+  }
+  if (
+    message.includes("admission envelope") ||
+    message.includes("admission work") ||
+    message.includes("admission token")
+  ) {
+    return "audit execution identity envelope rejected";
+  }
+  return "audit execution identity persistence failed";
 }

-/** Start one bounded worker queue. SQLite contention never blocks the agent-event callback. */
+/** Start one bounded queue on the current process's cached shared-state connection owner. */
 export function createAuditEventWriter(
  options: {
    stateDir?: string;
    maxPending?: number;
-    workerUrl?: URL;
+    onContention?: (message: string) => void;
    onError?: (error: string) => void;
  } = {},
 ): AuditEventWriter {
-  const workerUrl = options.workerUrl ?? resolveAuditEventWriterUrl();
-  const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined;
+  const database = {
+    env: { OPENCLAW_STATE_DIR: options.stateDir ?? resolveStateDir(process.env) },
+  };
  const maxPending = Math.max(1, Math.floor(options.maxPending ?? MAX_PENDING_AUDIT_EVENTS));
-  let worker: Worker;
-  try {
-    worker = new Worker(workerUrl, {
-      workerData: { stateDir: options.stateDir ?? resolveStateDir(process.env) },
-      execArgv: sourceWorkerExecArgv,
-    });
-  } catch (error) {
-    options.onError?.(formatAuditWriterError(error));
-    return {
-      ready: Promise.resolve(),
-      record: () => false,
-      recordExecutionIdentity: () => false,
-      recordExecutionDecision: () => false,
-      stop: async () => {},
-    };
-  }
-  worker.unref?.();
-
-  let pending = 0;
+  const queue: AuditWriterRequest[] = [];
  let stopped = false;
  let unavailable = false;
-  let readyResolved = false;
+  let maintenancePending = true;
+  let readyPending = true;
+  let scheduled: ReturnType<typeof setImmediate> | undefined;
+  let retryTimer: ReturnType<typeof setTimeout> | undefined;
+  let lockRetryAttempt = 0;
+  let lockContentionDelayMs = 0;
+  let lockContentionReported = false;
  let resolveReady!: () => void;
  const ready = new Promise<void>((resolve) => {
    resolveReady = resolve;
  });
+  let stopPromise: Promise<void> | undefined;
  let resolveStop: (() => void) | undefined;
  let stopTimer: ReturnType<typeof setTimeout> | undefined;

-  const markReady = () => {
-    if (!readyResolved) {
-      readyResolved = true;
-      resolveReady();
+  const fail = (error: unknown) => {
+    options.onError?.(formatAuditWriterError(error));
+  };
+  const reportContention = (message: string) => {
+    options.onContention?.(formatAuditWriterError(message));
+  };
+  const runWithoutBusyWait = <T>(operation: () => T): T =>
+    runWithOpenClawStateBusyTimeout(() => operation(), database, 0);
+  const observeLockContention = () => {
+    lockRetryAttempt += 1;
+  };
+  const resetLockContention = () => {
+    lockRetryAttempt = 0;
+    lockContentionDelayMs = 0;
+    lockContentionReported = false;
+  };
+  const reportMaintenance = (): AuditMaintenanceAttempt => {
+    let more = false;
+    for (const maintenance of [
+      () => pruneExpiredAuditEvents({ database }),
+      () => pruneExpiredExecutionIdentityContexts({ database }),
+      () => pruneExpiredExecutionDecisionFacts({ database }),
+      () => pruneExpiredOutboundMessageProgress({ database }),
+    ]) {
+      try {
+        more = runWithoutBusyWait(maintenance) > 0 || more;
+      } catch (error) {
+        if (isOpenClawStateWriteContentionError(error)) {
+          observeLockContention();
+          return "retry";
+        }
+        fail(error);
+      }
+    }
+    return more ? "more" : "settled";
+  };
+  const processRequest = (request: AuditWriterRequest): AuditWriterAttempt => {
+    try {
+      runWithoutBusyWait(() => {
+        if (request.type === "record-event") {
+          if (isOutboundMessageProgressInput(request.input)) {
+            recordOutboundMessageProgress(request.input, database);
+          } else {
+            recordAuditEvent(request.input, database);
+          }
+          return;
+        }
+        if (request.type === "record-execution-identity") {
+          processExecutionIdentityAdmissionWork(request.work, database);
+          return;
+        }
+        recordExecutionDecisionFact(request.receipt, database);
+      });
+      return "settled";
+    } catch (error) {
+      if (isOpenClawStateWriteContentionError(error)) {
+        observeLockContention();
+        return "retry";
+      }
+      resetLockContention();
+      if (request.type === "record-execution-identity") {
+        fail(executionIdentityFailureMessage(error));
+      } else if (request.type === "record-execution-decision") {
+        fail("audit execution decision rejected");
+      } else {
+        fail(error);
+      }
+      return "settled";
    }
  };
  const finishStop = () => {
@@ -108,17 +191,83 @@ export function createAuditEventWriter(
    resolveStop = undefined;
    finish?.();
  };
-  const fail = (error: unknown) => {
-    options.onError?.(formatAuditWriterError(error));
+  const schedule = () => {
+    if (retryTimer) {
+      if (stopped) {
+        retryTimer.ref?.();
+      }
+      return;
+    }
+    if (scheduled) {
+      if (stopped) {
+        scheduled.ref?.();
+      }
+      return;
+    }
+    scheduled = setImmediate(drainOne);
+    if (!stopped) {
+      scheduled.unref?.();
+    }
  };
+  const scheduleRetry = () => {
+    const delayMs = Math.min(
+      AUDIT_LOCK_RETRY_MAX_DELAY_MS,
+      AUDIT_LOCK_RETRY_DELAY_MS * 2 ** Math.min(6, Math.max(0, lockRetryAttempt - 1)),
+    );
+    lockContentionDelayMs += delayMs;
+    if (!lockContentionReported && lockContentionDelayMs >= AUDIT_LOCK_CONTENTION_REPORT_MS) {
+      lockContentionReported = true;
+      reportContention("audit event persistence delayed by SQLite lock contention");
+    }
+    retryTimer = setTimeout(() => {
+      retryTimer = undefined;
+      drainOne();
+    }, delayMs);
+    if (!stopped) {
+      retryTimer.unref?.();
+    }
+  };
+  function drainOne() {
+    scheduled = undefined;
+    if (maintenancePending) {
+      const maintenance = reportMaintenance();
+      if (readyPending) {
+        readyPending = false;
+        resolveReady();
+      }
+      if (maintenance === "retry") {
+        scheduleRetry();
+        return;
+      }
+      maintenancePending = maintenance === "more";
+      resetLockContention();
+    }
+    const request = queue.shift();
+    if (request && processRequest(request) === "retry") {
+      queue.unshift(request);
+      scheduleRetry();
+      return;
+    }
+    if (request) {
+      resetLockContention();
+    }
+    if (queue.length > 0 || maintenancePending) {
+      schedule();
+      return;
+    }
+    if (stopped) {
+      finishStop();
+    }
+  }
+  const maintenanceTimer = setInterval(() => {
+    maintenancePending = true;
+    schedule();
+  }, AUDIT_MAINTENANCE_INTERVAL_MS);
+  maintenanceTimer.unref?.();
+  schedule();

-  const enqueue = (
-    message:
-      | { type: "record-event"; input: AuditEventInput }
-      | { type: "record-execution-identity"; work: ExecutionIdentityAdmissionWork }
-      | { type: "record-execution-decision"; receipt: DecisionReceiptV1 },
-  ): boolean => {
-    if (stopped || unavailable || pending >= maxPending) {
+  const enqueue = (message: AuditWriterRequest): boolean => {
+    if (stopped || unavailable || queue.length >= maxPending) {
      if (!stopped) {
        fail(
          unavailable
@@ -128,14 +277,12 @@ export function createAuditEventWriter(
      }
      return false;
    }
-    pending += 1;
    try {
-      // Node Worker.postMessage is not the browser Window API and has no targetOrigin.
-      // oxlint-disable-next-line unicorn/require-post-message-target-origin
-      worker.postMessage(message);
+      // Preserve the former Worker boundary's clone and prototype-stripping contract.
+      queue.push(structuredClone(message));
+      schedule();
      return true;
    } catch (error) {
-      pending -= 1;
      if (message.type !== "record-event") {
        fail(
          message.type === "record-execution-identity"
@@ -144,78 +291,43 @@ export function createAuditEventWriter(
        );
      } else {
        unavailable = true;
-        void worker.terminate();
        fail(error);
      }
      return false;
    }
  };

-  worker.on("message", (message: AuditWriterMessage) => {
-    switch (message.type) {
-      case "ready":
-        markReady();
-        return;
-      case "recorded":
-        pending = Math.max(0, pending - 1);
-        return;
-      case "record-error":
-        pending = Math.max(0, pending - 1);
-        fail(message.error);
-        return;
-      case "maintenance-error":
-        fail(message.error);
-        return;
-      case "stopped":
-        pending = 0;
-        markReady();
-        finishStop();
-    }
-  });
-  worker.on("error", (error) => {
-    unavailable = true;
-    fail(error);
-    markReady();
-    finishStop();
-  });
-  worker.on("exit", (code) => {
-    unavailable = true;
-    if (!stopped) {
-      fail(`audit event writer exited with code ${code}`);
-    }
-    markReady();
-    finishStop();
-  });
-
  return {
    ready,
    record: (input) => enqueue({ type: "record-event", input }),
    recordExecutionIdentity: (work) => enqueue({ type: "record-execution-identity", work }),
    recordExecutionDecision: (receipt) => enqueue({ type: "record-execution-decision", receipt }),
-    stop: async () => {
-      if (stopped) {
-        return;
+    stop: () => {
+      if (stopPromise) {
+        return stopPromise;
      }
      stopped = true;
-      if (unavailable) {
-        return;
-      }
-      await new Promise<void>((resolve) => {
+      clearInterval(maintenanceTimer);
+      maintenancePending = true;
+      stopPromise = new Promise<void>((resolve) => {
        resolveStop = resolve;
        stopTimer = setTimeout(() => {
+          queue.length = 0;
+          if (scheduled) {
+            clearImmediate(scheduled);
+            scheduled = undefined;
+          }
+          if (retryTimer) {
+            clearTimeout(retryTimer);
+            retryTimer = undefined;
+          }
          fail("audit event writer shutdown timed out; pending metadata may be lost");
-          void worker.terminate();
          finishStop();
        }, AUDIT_WRITER_SHUTDOWN_TIMEOUT_MS);
-        try {
-          // Node Worker.postMessage is not the browser Window API and has no targetOrigin.
-          // oxlint-disable-next-line unicorn/require-post-message-target-origin
-          worker.postMessage({ type: "stop" });
-        } catch (error) {
-          fail(error);
-          finishStop();
-        }
+        stopTimer.unref?.();
+        schedule();
      });
+      return stopPromise;
    },
  };
 }
@@ -1,131 +0,0 @@
-/** Worker-thread entrypoint for serialized audit writes and retention maintenance. */
-import { parentPort, workerData } from "node:worker_threads";
-import { closeOpenClawStateDatabase } from "../state/openclaw-state-db.js";
-import { pruneExpiredAuditEvents, recordAuditEvent } from "./audit-event-store.js";
-import { isOutboundMessageProgressInput, type AuditEventInput } from "./audit-event-types.js";
-import {
-  pruneExpiredExecutionDecisionFacts,
-  recordExecutionDecisionFact,
-} from "./execution-decision-facts.js";
-import {
-  processExecutionIdentityAdmissionWork,
-  pruneExpiredExecutionIdentityContexts,
-} from "./execution-identity-context.js";
-import {
-  pruneExpiredOutboundMessageProgress,
-  recordOutboundMessageProgress,
-} from "./message-delivery-progress-store.js";
-
-const AUDIT_MAINTENANCE_INTERVAL_MS = 60 * 60_000;
-
-type AuditWriterRequest =
-  | { type: "record-event"; input: AuditEventInput }
-  | { type: "record-execution-identity"; work: unknown }
-  | { type: "record-execution-decision"; receipt: unknown }
-  | { type: "stop" };
-
-const stateDir =
-  workerData && typeof workerData === "object" && typeof workerData.stateDir === "string"
-    ? workerData.stateDir
-    : undefined;
-if (!parentPort || !stateDir) {
-  throw new Error("audit event writer requires a parent port and state directory");
-}
-const port = parentPort;
-const database = { env: { OPENCLAW_STATE_DIR: stateDir } };
-
-function executionIdentityFailureMessage(error: unknown): string {
-  const message = error instanceof Error ? error.message : String(error);
-  if (
-    message.includes("audit identity key is missing") ||
-    message.includes("audit identity key is corrupt")
-  ) {
-    return "audit execution identity key unavailable";
-  }
-  if (message.includes("execution identity context conflict")) {
-    return "audit execution identity context conflict";
-  }
-  if (message.includes("execution identity recovery evidence unavailable")) {
-    return "audit execution identity recovery evidence unavailable";
-  }
-  if (
-    message.includes("admission envelope") ||
-    message.includes("admission work") ||
-    message.includes("admission token")
-  ) {
-    return "audit execution identity envelope rejected";
-  }
-  return "audit execution identity persistence failed";
-}
-
-function reportMaintenance(): void {
-  try {
-    pruneExpiredAuditEvents({ database });
-  } catch (error) {
-    port.postMessage({ type: "maintenance-error", error: String(error) });
-  }
-  try {
-    pruneExpiredExecutionIdentityContexts({ database });
-  } catch (error) {
-    port.postMessage({ type: "maintenance-error", error: String(error) });
-  }
-  try {
-    pruneExpiredExecutionDecisionFacts({ database });
-  } catch (error) {
-    port.postMessage({ type: "maintenance-error", error: String(error) });
-  }
-  try {
-    pruneExpiredOutboundMessageProgress({ database });
-  } catch (error) {
-    port.postMessage({ type: "maintenance-error", error: String(error) });
-  }
-}
-
-reportMaintenance();
-const maintenanceTimer = setInterval(reportMaintenance, AUDIT_MAINTENANCE_INTERVAL_MS);
-port.postMessage({ type: "ready" });
-
-port.on("message", (message: AuditWriterRequest) => {
-  if (message.type === "record-event") {
-    try {
-      if (isOutboundMessageProgressInput(message.input)) {
-        recordOutboundMessageProgress(message.input, database);
-      } else {
-        recordAuditEvent(message.input, database);
-      }
-      port.postMessage({ type: "recorded" });
-    } catch (error) {
-      port.postMessage({ type: "record-error", error: String(error) });
-    }
-    return;
-  }
-  if (message.type === "record-execution-identity") {
-    try {
-      processExecutionIdentityAdmissionWork(message.work, database);
-      port.postMessage({ type: "recorded" });
-    } catch (error) {
-      port.postMessage({ type: "record-error", error: executionIdentityFailureMessage(error) });
-    }
-    return;
-  }
-  if (message.type === "record-execution-decision") {
-    try {
-      recordExecutionDecisionFact(message.receipt, database);
-      port.postMessage({ type: "recorded" });
-    } catch {
-      port.postMessage({ type: "record-error", error: "audit execution decision rejected" });
-    }
-    return;
-  }
-  clearInterval(maintenanceTimer);
-  reportMaintenance();
-  try {
-    // The Gateway may still own a live connection. Leave WAL reset to the
-    // final lifecycle owner instead of waiting on or invalidating its readers.
-    closeOpenClawStateDatabase({ checkpointMode: "PASSIVE" });
-  } catch (error) {
-    port.postMessage({ type: "maintenance-error", error: String(error) });
-  }
-  port.postMessage({ type: "stopped" });
-  port.close();
-});
@@ -287,6 +287,32 @@ describe("audit event persistence", () => {
    pruneExpiredAuditEvents({ database, now: expiredAt });
    expect(listAuditEvents({ database, limit: 10, now: occurredAt }).events).toEqual([]);
  });
+
+  it("bounds each expired-row maintenance transaction", () => {
+    const database = createDatabaseOptions();
+    const { db } = openOpenClawStateDatabase(database);
+    const now = Date.now();
+    const expiredAt = now - AUDIT_EVENT_RETENTION_MS_CONTRACT - 1;
+    db.prepare(
+      `WITH RECURSIVE numbers(n) AS (
+         SELECT 1
+         UNION ALL
+         SELECT n + 1 FROM numbers WHERE n < ?
+       )
+       INSERT INTO audit_events (
+         event_id, source_id, source_sequence, occurred_at, kind, action, status,
+         actor_type, actor_id, agent_id, run_id
+       )
+       SELECT 'expired-event-' || n, 'expired-source-' || n, n, ?, 'agent_run',
+              'agent.run.started', 'started', 'agent', 'main', 'main', 'expired-run-' || n
+       FROM numbers`,
+    ).run(AUDIT_EVENT_PRUNE_BATCH_ROWS_CONTRACT + 1, expiredAt);
+
+    expect(pruneExpiredAuditEvents({ database, now })).toBe(AUDIT_EVENT_PRUNE_BATCH_ROWS_CONTRACT);
+    expect(db.prepare("SELECT COUNT(*) AS count FROM audit_events").get()).toEqual({ count: 1 });
+    expect(pruneExpiredAuditEvents({ database, now })).toBe(1);
+    expect(pruneExpiredAuditEvents({ database, now })).toBe(0);
+  });
 });

 describe("agent activity audit projection", () => {
@@ -1,10 +1,41 @@
-import { describe, expect, it } from "vitest";
+import { beforeEach, describe, expect, it, vi } from "vitest";
 import type { AuditEventInput } from "./audit-event-types.js";
 import type { AuditEventWriter } from "./audit-event-writer.js";
 import { createAuditEventRecorder } from "./audit-recorder.js";
 import { emitTrustedMessageAuditEvent } from "./message-audit-events.js";
 import { onTrustedMessageAuditEventForTest as onTrustedMessageAuditEvent } from "./message-audit-events.test-support.js";

+const recorderMocks = vi.hoisted(() => ({
+  onContention: (_message: string) => {},
+  onError: (_error: string) => {},
+  warn: vi.fn(),
+}));
+
+vi.mock("../logging/subsystem.js", () => ({
+  createSubsystemLogger: () => ({
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: recorderMocks.warn,
+  }),
+}));
+
+vi.mock("./audit-event-writer.js", () => ({
+  createAuditEventWriter: (options: {
+    onContention?: (message: string) => void;
+    onError?: (error: string) => void;
+  }) => {
+    recorderMocks.onContention = options.onContention ?? (() => {});
+    recorderMocks.onError = options.onError ?? (() => {});
+    return {
+      ready: Promise.resolve(),
+      record: () => true,
+      recordExecutionIdentity: () => true,
+      recordExecutionDecision: () => true,
+      stop: async () => {},
+    };
+  },
+}));
+
 function captureWriter(inputs: AuditEventInput[]): AuditEventWriter {
  return {
    ready: Promise.resolve(),
@@ -34,6 +65,25 @@ function emitMessage(conversationKind: "direct" | "group") {
 }

 describe("message audit recorder", () => {
+  beforeEach(() => {
+    recorderMocks.warn.mockReset();
+    recorderMocks.onContention = () => {};
+    recorderMocks.onError = () => {};
+  });
+
+  it("keeps recoverable contention separate from metadata-loss warnings", async () => {
+    const recorder = createAuditEventRecorder({ messageMode: "off" });
+
+    recorderMocks.onContention("audit event persistence delayed by SQLite lock contention");
+    recorderMocks.onError("audit writer shutdown timed out with pending metadata");
+
+    expect(recorderMocks.warn.mock.calls).toEqual([
+      ["audit event persistence delayed by SQLite lock contention"],
+      ["audit event persistence failed: audit writer shutdown timed out with pending metadata"],
+    ]);
+    await recorder.stop();
+  });
+
  it("keeps message events off by default policy", async () => {
    const inputs: AuditEventInput[] = [];
    const recorder = createAuditEventRecorder({
@@ -30,6 +30,9 @@ export function createAuditEventRecorder(options: {
    options.writer ??
    createAuditEventWriter({
      ...(options.stateDir ? { stateDir: options.stateDir } : {}),
+      onContention: (message) => {
+        log.warn(message);
+      },
      onError: (error) => {
        if (!persistenceFailureWarned) {
          persistenceFailureWarned = true;
@@ -303,7 +303,7 @@ export function parseExecutionIdentityAdmissionToken(

 function redactDisplayLabel(value: string): string {
  // The shared redactor's secret-prefix pass becomes stable on its second pass.
-  // Stabilizing here lets the worker reject any altered structured-clone payload.
+  // Stabilizing here lets the writer reject any altered structured-clone payload.
  return truncateUtf16Safe(
    redactSensitiveText(redactSensitiveText(value, { mode: "tools" }), { mode: "tools" }),
    128,
@@ -370,7 +370,7 @@ function captureExecutionIdentityAdmissionEnvelope(
  return freezeEnvelope(validateEnvelope(envelope));
 }

-/** Revalidate a structured-cloned worker message before any persistence work. */
+/** Revalidate a structured-cloned queue message before any persistence work. */
 export function parseExecutionIdentityAdmissionEnvelope(
  value: unknown,
 ): ExecutionIdentityAdmissionEnvelope {
@@ -389,7 +389,7 @@ export function parseExecutionIdentityAdmissionEnvelope(
  return parsed;
 }

-/** Revalidate either bounded worker message before schema, key, or database work. */
+/** Revalidate either bounded queue message before schema, key, or database work. */
 export function parseExecutionIdentityAdmissionWork(
  value: unknown,
 ): ExecutionIdentityAdmissionWork {
@@ -1,4 +1,4 @@
-/** Worker-only canonical context construction and bounded value helpers. */
+/** Queue-only canonical context construction and bounded value helpers. */
 import type { DatabaseSync } from "node:sqlite";
 import type { ExecutionIdentityContextV1 } from "../../packages/gateway-protocol/src/index.js";
 import { validateExecutionIdentityContextV1 } from "../../packages/gateway-protocol/src/index.js";
@@ -192,7 +192,7 @@ function pruneExecutionIdentityContextsAfterInsert(
  const remainingPruneBudget = Math.max(0, limits.pruneBatchRows - expiredCount);
  if (remainingPruneBudget > 0) {
    // Derive overflow from committed rows inside this transaction. A process-local
-    // count misses writes from the Gateway worker or a concurrent direct CLI.
+    // count misses writes committed by another state-database connection.
    const retainedIds = kysely
      .selectFrom("execution_identity_contexts")
      .select("context_id")
@@ -241,7 +241,7 @@ export function pruneExpiredExecutionIdentityContexts(
  );
 }

-/** Worker-owned canonicalization and persistence for one accepted admission envelope. */
+/** Queue-owned canonicalization and persistence for one accepted admission envelope. */
 function persistExecutionIdentityAdmissionEnvelope(
  input: unknown,
  options: ExecutionIdentityStoreOptions = {},
@@ -335,7 +335,7 @@ function verifyExecutionIdentityAdmissionRetry(
  return context;
 }

-/** Worker-owned persistence/verification for one accepted bounded queue item. */
+/** Queue-owned persistence/verification for one accepted bounded queue item. */
 export function processExecutionIdentityAdmissionWork(
  input: unknown,
  options: ExecutionIdentityStoreOptions = {},
@@ -26,6 +26,7 @@ import {

 const tempDirs = useAutoCleanupTempDirTracker(afterEach);
 const PINNED_PRE_C04_READER_SHA = "5dc4cf602bc5e263e83cd16a12bb1e100544f4c3";
+const OUTBOUND_PROGRESS_PRUNE_BATCH_ROWS_CONTRACT = 1_024;

 function ensurePinnedReaderCommit(repositoryRoot: string): void {
  try {
@@ -475,4 +476,37 @@ describe("outbound message progress companion", () => {
      (db.prepare("SELECT COUNT(*) AS count FROM audit_events").get() as { count: number }).count,
    ).toBe(1);
  });
+
+  it("bounds each expired progress maintenance transaction", () => {
+    const database = databaseOptions();
+    recordOutboundMessageProgress(progressInput("message.outbound.queued"), database);
+    const { db } = openOpenClawStateDatabase(database);
+    db.exec("DELETE FROM outbound_message_progress");
+    const now = Date.now();
+    const expiredAt = now - 31 * 24 * 60 * 60_000;
+    db.prepare(
+      `WITH RECURSIVE numbers(n) AS (
+         SELECT 1
+         UNION ALL
+         SELECT n + 1 FROM numbers WHERE n < ?
+       )
+       INSERT INTO outbound_message_progress (
+         progress_id, source_id, source_sequence, schema_version, occurred_at, action,
+         outcome, actor_type, actor_id, agent_id, run_id, channel, conversation_kind
+       )
+       SELECT 'expired-progress-' || n, 'expired-progress-source-' || n, n, 1, ?,
+              'message.outbound.queued', 'queued', 'agent', 'main', 'main',
+              'expired-progress-run-' || n, 'qa-channel', 'direct'
+       FROM numbers`,
+    ).run(OUTBOUND_PROGRESS_PRUNE_BATCH_ROWS_CONTRACT + 1, expiredAt);
+
+    expect(pruneExpiredOutboundMessageProgress({ database, now })).toBe(
+      OUTBOUND_PROGRESS_PRUNE_BATCH_ROWS_CONTRACT,
+    );
+    expect(db.prepare("SELECT COUNT(*) AS count FROM outbound_message_progress").get()).toEqual({
+      count: 1,
+    });
+    expect(pruneExpiredOutboundMessageProgress({ database, now })).toBe(1);
+    expect(pruneExpiredOutboundMessageProgress({ database, now })).toBe(0);
+  });
 });
@@ -240,14 +240,24 @@ function countProgressRows(db: DatabaseSync): number {
  return normalizeSqliteNumber(row?.count ?? null) ?? 0;
 }

+function deleteExpiredProgressRows(db: DatabaseSync, now: number, limit: number) {
+  const kysely = progressDb(db);
+  const expiredSequences = kysely
+    .selectFrom("outbound_message_progress")
+    .select("sequence")
+    .where("occurred_at", "<", now - OUTBOUND_MESSAGE_PROGRESS_RETENTION_MS)
+    .orderBy("occurred_at", "asc")
+    .orderBy("sequence", "asc")
+    .limit(limit);
+  return executeSqliteQuerySync(
+    db,
+    kysely.deleteFrom("outbound_message_progress").where("sequence", "in", expiredSequences),
+  );
+}
+
 function pruneProgressAfterInsert(db: DatabaseSync, now: number): void {
  const kysely = progressDb(db);
-  const expired = executeSqliteQuerySync(
-    db,
-    kysely
-      .deleteFrom("outbound_message_progress")
-      .where("occurred_at", "<", now - OUTBOUND_MESSAGE_PROGRESS_RETENTION_MS),
-  );
+  const expired = deleteExpiredProgressRows(db, now, OUTBOUND_MESSAGE_PROGRESS_PRUNE_BATCH_ROWS);
  const cachedCount = progressRowCounts.get(db);
  let rowCount =
    cachedCount === undefined
@@ -454,22 +464,18 @@ export function hasOutboundMessageProgressCursor(params: {
 /** Prune existing progress without creating its lazy table. */
 export function pruneExpiredOutboundMessageProgress(
  params: { now?: number; database?: OpenClawStateDatabaseOptions } = {},
-): void {
+): number {
  const database = openOpenClawStateDatabase(params.database);
  if (!tableExists(database.db, "outbound_message_progress")) {
-    return;
+    return 0;
  }
-  runOpenClawStateWriteTransaction(({ db }) => {
-    executeSqliteQuerySync(
+  return runOpenClawStateWriteTransaction(({ db }) => {
+    const deleted = deleteExpiredProgressRows(
      db,
-      progressDb(db)
-        .deleteFrom("outbound_message_progress")
-        .where(
-          "occurred_at",
-          "<",
-          (params.now ?? Date.now()) - OUTBOUND_MESSAGE_PROGRESS_RETENTION_MS,
-        ),
+      params.now ?? Date.now(),
+      OUTBOUND_MESSAGE_PROGRESS_PRUNE_BATCH_ROWS,
    );
    progressRowCounts.delete(db);
+    return Number(deleted.numAffectedRows ?? 0n);
  }, params.database);
 }
@@ -8,7 +8,7 @@ import { createLocalSqliteSnapshotProvider } from "../snapshot/local-repository.
 import { OPENCLAW_AGENT_SCHEMA_VERSION } from "../state/openclaw-agent-db.js";
 import { resolveOpenClawAgentSqlitePath } from "../state/openclaw-agent-db.paths.js";
 import { OPENCLAW_AGENT_SCHEMA_SQL } from "../state/openclaw-agent-schema.js";
-import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "../state/openclaw-state-schema.js";
 import {
@@ -15,10 +15,8 @@ import {
  type SkillProposalRecord,
  type SkillProposalRollback,
 } from "../skills/workshop/types.js";
-import {
-  OPENCLAW_STATE_SCHEMA_VERSION,
-  openOpenClawStateDatabase,
-} from "../state/openclaw-state-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
+import { openOpenClawStateDatabase } from "../state/openclaw-state-db.js";
 import {
  createOpenClawTestState,
  type OpenClawTestState,
@@ -31,10 +31,10 @@ import {
 import type { InstalledPluginInstallRecordInfo } from "../plugins/installed-plugin-index.js";
 import { EMPTY_LEGACY_SESSION_SURFACES } from "../plugins/legacy-session-surfaces.types.js";
 import { closeOpenClawAgentDatabasesForTest } from "../state/openclaw-agent-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
 } from "../state/openclaw-state-db.js";
 import { loadTaskFlowRegistryStateFromSqlite } from "../tasks/task-flow-registry.store.sqlite.js";
 import { loadTaskRegistryStateFromSqlite } from "../tasks/task-registry.store.sqlite.js";
@@ -8,10 +8,10 @@ import {
  readOpenClawDatabaseQuarantine,
  recordOpenClawDatabaseQuarantine,
 } from "../state/openclaw-quarantine-store.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabase,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
 } from "../state/openclaw-state-db.js";
 import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "../state/openclaw-state-schema.js";
@@ -2,7 +2,7 @@ import fs from "node:fs";
 import path from "node:path";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { requireNodeSqlite } from "../infra/node-sqlite.js";
-import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
 import {
  autoMigrateLegacyStateDir,
@@ -15,15 +15,15 @@ const outro = (message: string) => clackOutro(stylePromptTitle(message) ?? messa
 const loadConfigModule = createLazyRuntimeModule(() => import("../config/config.js"));

 async function assertDoctorDatabaseSchemasCompatible(): Promise<void> {
-  const [databasePreflight, agentDatabase, stateDatabase] = await Promise.all([
+  const [databasePreflight, agentDatabase, stateDatabaseContract] = await Promise.all([
    import("../state/openclaw-database-preflight.js"),
    import("../state/openclaw-agent-db.js"),
-    import("../state/openclaw-state-db.js"),
+    import("../state/openclaw-state-db-contract.js"),
  ]);
  const databaseSchemas = databasePreflight.preflightOpenClawDatabaseSchemas({
    env: process.env,
    supportedVersions: {
-      state: stateDatabase.OPENCLAW_STATE_SCHEMA_VERSION,
+      state: stateDatabaseContract.OPENCLAW_STATE_SCHEMA_VERSION,
      agent: agentDatabase.OPENCLAW_AGENT_SCHEMA_VERSION,
    },
  });
@@ -297,6 +297,27 @@ describe("createGatewayCloseHandler", () => {
    expect(closed).toBe(true);
  });

+  it("drains the audit subscription before the final shared-state close", async () => {
+    const events: string[] = [];
+    const agentUnsub = vi.fn(async () => {
+      events.push("agent-unsub");
+    });
+    const stopMediaCleanup = vi.fn(async () => {
+      events.push("media-cleanup");
+      return "drained" as const;
+    });
+    mocks.closePluginStateDatabase.mockImplementationOnce(async () => {
+      events.push("shared-state-close");
+    });
+    const close = createGatewayCloseHandler(
+      createGatewayCloseTestDeps({ agentUnsub, stopMediaCleanup }),
+    );
+
+    await close({ reason: "test" });
+
+    expect(events).toEqual(["media-cleanup", "agent-unsub", "shared-state-close"]);
+  });
+
  it("retains shared state when media cleanup times out", async () => {
    const stopMediaCleanup = vi.fn(async () => "timed-out" as const);
    const close = createGatewayCloseHandler(createGatewayCloseTestDeps({ stopMediaCleanup }));
@@ -891,9 +891,7 @@ export function createGatewayCloseHandler(
        shutdownLog.warn(`media-cleanup: ${err instanceof Error ? err.message : String(err)}`);
        recordShutdownWarning(warnings, "media-cleanup");
      }
-      if (mediaCleanupStopResult === "drained") {
-        await shutdownStep("plugin-state-store", () => closePluginStateDatabase(), warnings);
-      } else {
+      if (mediaCleanupStopResult !== "drained") {
        // Timed-out cleanup still owns shared SQLite. Keep the process store open
        // so late completion cannot resume against a database torn down by shutdown.
        recordShutdownWarning(warnings, "media-cleanup");
@@ -943,6 +941,12 @@ export function createGatewayCloseHandler(
      if (params.taskUnsub) {
        await shutdownStep("task-unsub", () => params.taskUnsub!(), warnings);
      }
+      if (mediaCleanupStopResult === "drained") {
+        // Audit draining is part of agentUnsub and can reopen the process-cached owner
+        // for final maintenance. Close shared state only after every database-backed
+        // producer has finished.
+        await shutdownStep("plugin-state-store", () => closePluginStateDatabase(), warnings);
+      }
      params.chatRunState.clear();
      let clientCloseFailures = 0;
      for (const c of params.clients) {
@@ -88,7 +88,7 @@ export function startGatewayEventSubscriptions(params: {
  restartRecoveryCandidates: Map<string, RestartRecoveryCandidate>;
  terminalSessions: Pick<TerminalSessionManager, "closeAgentSessions">;
 }) {
-  // The worker always runs retention maintenance. audit.enabled only controls
+  // The audit writer always runs retention maintenance. audit.enabled only controls
  // producer subscriptions, so disabling collection cannot strand expired rows.
  const runtimeConfig = getRuntimeConfig();
  const auditEnabled = isAuditLedgerEnabled(runtimeConfig);
@@ -88,16 +88,16 @@ export async function prepareGatewayServerBootstrap(input: {
      preflightOpenClawDatabaseSchemas,
    },
    agentDatabase,
-    stateDatabase,
+    stateDatabaseContract,
  ] = await Promise.all([
    import("../state/openclaw-database-preflight.js"),
    import("../state/openclaw-agent-db.js"),
-    import("../state/openclaw-state-db.js"),
+    import("../state/openclaw-state-db-contract.js"),
  ]);
  const databaseSchemas = preflightOpenClawDatabaseSchemas({
    env: process.env,
    supportedVersions: {
-      state: stateDatabase.OPENCLAW_STATE_SCHEMA_VERSION,
+      state: stateDatabaseContract.OPENCLAW_STATE_SCHEMA_VERSION,
      agent: agentDatabase.OPENCLAW_AGENT_SCHEMA_VERSION,
    },
  });
@@ -9,12 +9,12 @@ import type {
  WorkerProfile,
  WorkerSshEndpoint,
 } from "../../plugins/types.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../../state/openclaw-state-db-contract.js";
 import { ensureAdditiveStateColumns } from "../../state/openclaw-state-db-schema-additive.js";
 import {
  assertOpenClawStateDatabaseForMaintenance,
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  type OpenClawStateDatabase,
 } from "../../state/openclaw-state-db.js";
 import { hashWorkerCredential } from "./credential.js";
@@ -5,10 +5,8 @@ import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
-import {
-  closeOpenClawStateDatabaseForTest,
-  OPENCLAW_STATE_SCHEMA_VERSION,
-} from "../state/openclaw-state-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
+import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js";
 import { withTempDir } from "../test-utils/temp-dir.js";
 import { resolveDeviceIdentityCoordinatorPaths } from "./device-identity-coordinator-paths.js";
 import { acquireDeviceIdentityCoordinator } from "./device-identity-coordinator.js";
@@ -5,10 +5,10 @@ import os from "node:os";
 import path from "node:path";
 import { DatabaseSync } from "node:sqlite";
 import { afterEach, describe, expect, it, vi } from "vitest";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
 import {
  closeOpenClawStateDatabaseForTest,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  openOpenClawStateDatabase,
 } from "../state/openclaw-state-db.js";
 import {
@@ -0,0 +1,39 @@
+import { DatabaseSync } from "node:sqlite";
+import { afterEach, describe, expect, it } from "vitest";
+import { runWithSqliteBusyTimeout } from "./sqlite-busy-timeout.js";
+
+describe("runWithSqliteBusyTimeout", () => {
+  let database: DatabaseSync | undefined;
+
+  afterEach(() => {
+    database?.close();
+    database = undefined;
+  });
+
+  it("restores the previous timeout after success and failure", () => {
+    database = new DatabaseSync(":memory:");
+    database.exec("PRAGMA busy_timeout = 5000");
+
+    expect(
+      runWithSqliteBusyTimeout(database, 0, () => database?.prepare("PRAGMA busy_timeout").get()),
+    ).toEqual({ timeout: 0 });
+    expect(database.prepare("PRAGMA busy_timeout").get()).toEqual({ timeout: 5000 });
+
+    expect(() =>
+      runWithSqliteBusyTimeout(database!, 25, () => {
+        throw new Error("operation failed");
+      }),
+    ).toThrow("operation failed");
+    expect(database.prepare("PRAGMA busy_timeout").get()).toEqual({ timeout: 5000 });
+  });
+
+  it.each([-1, 1.5, Number.NaN, Number.POSITIVE_INFINITY])(
+    "rejects invalid timeout %s",
+    (timeout) => {
+      database = new DatabaseSync(":memory:");
+      expect(() => runWithSqliteBusyTimeout(database!, timeout, () => undefined)).toThrow(
+        "busyTimeoutMs must be a non-negative integer",
+      );
+    },
+  );
+});
@@ -0,0 +1,59 @@
+import type { DatabaseSync } from "node:sqlite";
+
+export type SqliteLockFailureReporting = "report" | "suppress";
+
+const lockFailureReportingByDatabase = new WeakMap<DatabaseSync, SqliteLockFailureReporting>();
+
+export function normalizeSqliteNonNegativeInteger(value: number, label: string): number {
+  if (!Number.isInteger(value) || value < 0) {
+    throw new Error(`${label} must be a non-negative integer`);
+  }
+  return value;
+}
+
+export function readSqliteBusyTimeout(database: DatabaseSync): number {
+  const row = database // sqlite-allow-raw -- Connection-local policy must be restored after the bounded operation.
+    .prepare("PRAGMA busy_timeout")
+    .get();
+  const value = row?.busy_timeout ?? row?.timeout;
+  return typeof value === "bigint" ? Number(value) : Number(value ?? 0);
+}
+
+export function setSqliteBusyTimeout(database: DatabaseSync, busyTimeoutMs: number): void {
+  const normalizedTimeoutMs = normalizeSqliteNonNegativeInteger(busyTimeoutMs, "busyTimeoutMs");
+  database.exec(`PRAGMA busy_timeout = ${normalizedTimeoutMs}`); // sqlite-allow-raw -- Connection-local lock policy.
+}
+
+export function shouldReportSqliteLockFailure(database: DatabaseSync): boolean {
+  return lockFailureReportingByDatabase.get(database) !== "suppress";
+}
+
+/** Run one synchronous operation with a temporary connection-local busy timeout. */
+export function runWithSqliteBusyTimeout<T>(
+  database: DatabaseSync,
+  busyTimeoutMs: number,
+  operation: () => T,
+  options: { lockFailureReporting?: SqliteLockFailureReporting } = {},
+): T {
+  const normalizedTimeoutMs = normalizeSqliteNonNegativeInteger(busyTimeoutMs, "busyTimeoutMs");
+  const previousBusyTimeoutMs = readSqliteBusyTimeout(database);
+  const previousLockFailureReporting = lockFailureReportingByDatabase.get(database);
+  if (options.lockFailureReporting) {
+    lockFailureReportingByDatabase.set(database, options.lockFailureReporting);
+  }
+  if (previousBusyTimeoutMs !== normalizedTimeoutMs) {
+    setSqliteBusyTimeout(database, normalizedTimeoutMs);
+  }
+  try {
+    return operation();
+  } finally {
+    if (database.isOpen && previousBusyTimeoutMs !== normalizedTimeoutMs) {
+      setSqliteBusyTimeout(database, previousBusyTimeoutMs);
+    }
+    if (previousLockFailureReporting) {
+      lockFailureReportingByDatabase.set(database, previousLockFailureReporting);
+    } else {
+      lockFailureReportingByDatabase.delete(database);
+    }
+  }
+}
@@ -6,6 +6,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
 import { getNodeSqliteKysely } from "./kysely-sync.js";
 import { requireNodeSqlite } from "./node-sqlite.js";
+import { runWithSqliteBusyTimeout } from "./sqlite-busy-timeout.js";
 import {
  runSqliteDeferredTransactionSync,
  runSqliteImmediateTransactionSync,
@@ -263,6 +264,49 @@ describe("runSqliteImmediateTransactionSync", () => {
    );
  });

+  it("suppresses expected lock-probe diagnostics inside a connection-local policy", () => {
+    const logger = { warn: vi.fn() };
+    const lockError = Object.assign(new Error("database is locked"), {
+      code: "ERR_SQLITE_ERROR",
+      errcode: 5,
+    });
+    const db = {
+      isOpen: true,
+      prepare() {
+        return { get: () => ({ timeout: 0 }) };
+      },
+      exec(sql: string) {
+        if (sql === "BEGIN IMMEDIATE") {
+          throw lockError;
+        }
+      },
+    } as unknown as import("node:sqlite").DatabaseSync;
+
+    expect(() =>
+      runWithSqliteBusyTimeout(
+        db,
+        0,
+        () =>
+          runSqliteImmediateTransactionSync(db, () => "blocked", {
+            busyTimeoutMs: 0,
+            logger,
+            operationLabel: "audit.retry-probe",
+          }),
+        { lockFailureReporting: "suppress" },
+      ),
+    ).toThrow(lockError);
+    expect(logger.warn).not.toHaveBeenCalled();
+
+    expect(() =>
+      runSqliteImmediateTransactionSync(db, () => "blocked", {
+        busyTimeoutMs: 0,
+        logger,
+        operationLabel: "terminal-write",
+      }),
+    ).toThrow(lockError);
+    expect(logger.warn).toHaveBeenCalledTimes(1);
+  });
+
  it("does not warn for busyTimeoutMs: 0 with fast successful transactions (regression)", () => {
    const logger = { warn: vi.fn() };
    let now = 0;
@@ -5,6 +5,7 @@ import { createSubsystemLogger, type SubsystemLogger } from "../logging/subsyste
 // The cache-state module keeps this lifecycle edge off the kysely value graph
 // so cold control-plane paths using transactions do not load kysely.
 import { clearNodeSqliteKyselyCacheForDatabase } from "./kysely-sync-cache-state.js";
+import { shouldReportSqliteLockFailure } from "./sqlite-busy-timeout.js";

 const transactionDepthByDatabase = new WeakMap<DatabaseSync, number>();

@@ -150,7 +151,7 @@ function execTimedTransactionStep(params: {
    return elapsedMs;
  } catch (error) {
    const elapsedMs = Date.now() - startedAt;
-    if (isSqliteLockError(error)) {
+    if (isSqliteLockError(error) && shouldReportSqliteLockFailure(params.db)) {
      const sqliteErrcode = sqliteExtendedResultCode(error);
      const sqlitePrimaryCode = sqlitePrimaryResultCode(error);
      transactionLogger(params.options).warn("SQLite transaction lock wait failed", {
@@ -4,6 +4,7 @@ import path from "node:path";
 import type { DatabaseSync } from "node:sqlite";
 import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion";
 import type { Result } from "@openclaw/normalization-core/result";
+import { normalizeSqliteNonNegativeInteger } from "./sqlite-busy-timeout.js";
 import { isSqliteLockError } from "./sqlite-transaction.js";

 // WAL maintenance configures SQLite write-ahead logging and schedules bounded
@@ -58,7 +59,7 @@ export type SqliteConnectionPragmaOptions = SqliteWalMaintenanceOptions & {
 };

 function configureSqliteBusyTimeout(db: DatabaseSync, busyTimeoutMs: number): number {
-  const normalizedTimeoutMs = normalizeNonNegativeInteger(busyTimeoutMs, "busyTimeoutMs");
+  const normalizedTimeoutMs = normalizeSqliteNonNegativeInteger(busyTimeoutMs, "busyTimeoutMs");
  db.exec(`PRAGMA busy_timeout = ${normalizedTimeoutMs};`);
  return normalizedTimeoutMs;
 }
@@ -86,13 +87,6 @@ export function configureSqlitePreSchemaPragmas(
  enableIncrementalAutoVacuumForFreshDatabase(db);
 }

-function normalizeNonNegativeInteger(value: number, label: string): number {
-  if (!Number.isInteger(value) || value < 0) {
-    throw new Error(`${label} must be a non-negative integer`);
-  }
-  return value;
-}
-
 function findExistingVolumePaths(
  targetPath: string,
 ): { canonicalPath: string; originalPath: string } | null {
@@ -439,11 +433,11 @@ export function configureSqliteWalMaintenance(
 ): SqliteWalMaintenance {
  const busyTimeoutMs =
    options.busyTimeoutMs === undefined ? 0 : configureSqliteBusyTimeout(db, options.busyTimeoutMs);
-  const autoCheckpointPages = normalizeNonNegativeInteger(
+  const autoCheckpointPages = normalizeSqliteNonNegativeInteger(
    options.autoCheckpointPages ?? DEFAULT_SQLITE_WAL_AUTOCHECKPOINT_PAGES,
    "autoCheckpointPages",
  );
-  const checkpointIntervalMs = normalizeNonNegativeInteger(
+  const checkpointIntervalMs = normalizeSqliteNonNegativeInteger(
    options.checkpointIntervalMs ?? DEFAULT_SQLITE_WAL_CHECKPOINT_INTERVAL_MS,
    "checkpointIntervalMs",
  );
@@ -3,10 +3,10 @@ import { existsSync, mkdirSync } from "node:fs";
 import path from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
 import {
  closeOpenClawStateDatabaseForTest,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  runOpenClawStateWriteTransaction,
  withOpenClawStateStartupMigrationCheckpointDatabase,
 } from "../state/openclaw-state-db.js";
@@ -23,11 +23,11 @@ import {
  OPENCLAW_AGENT_SCHEMA_VERSION,
  runOpenClawAgentWriteTransaction,
 } from "../state/openclaw-agent-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  runOpenClawStateWriteTransaction,
 } from "../state/openclaw-state-db.js";
 import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
@@ -8,12 +8,12 @@ import {
  clearOpenClawDatabaseQuarantine,
  recordOpenClawDatabaseQuarantine,
 } from "../state/openclaw-quarantine-store.js";
+import { recordOpenClawStateDatabaseOpenFailure } from "../state/openclaw-state-db-cache.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import {
  clearOpenClawStateDatabaseOpenFailure,
  isOpenClawStateDatabaseOpen,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  openOpenClawStateDatabase,
-  recordOpenClawStateDatabaseOpenFailure,
 } from "../state/openclaw-state-db.js";
 import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
 import {
@@ -4,10 +4,10 @@ import path from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { requireNodeSqlite } from "../../infra/node-sqlite.js";
 import { isSecretValueRegisteredForRedaction } from "../../logging/secret-redaction-registry.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../../state/openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
 } from "../../state/openclaw-state-db.js";
 import {
  deleteSecretStoreEntry,
@@ -1,8 +1,8 @@
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
 import { requireNodeSqlite } from "../../infra/node-sqlite.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../../state/openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  openOpenClawStateDatabase,
 } from "../../state/openclaw-state-db.js";
 import {
@@ -9,7 +9,7 @@ import { createPrivateSqliteDirectory } from "../infra/sqlite-private-directory.
 import { runExec } from "../process/exec.js";
 import { OPENCLAW_AGENT_SCHEMA_VERSION } from "../state/openclaw-agent-db.js";
 import { OPENCLAW_AGENT_SCHEMA_SQL } from "../state/openclaw-agent-schema.js";
-import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "../state/openclaw-state-schema.js";
 import { hashSnapshotArtifact, readSnapshotManifest } from "./manifest.js";
 import {
@@ -11,14 +11,12 @@ import {
  CLAW_LAZY_ADDITIVE_STATE_COLUMN_DEFINITIONS,
  CLAW_STARTUP_ADDITIVE_STATE_COLUMN_DEFINITIONS,
 } from "./openclaw-state-db-additive-columns.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js";
 import {
  ensureAdditiveStateColumns,
  ensureDevicePairSetupBootstrapSchema,
 } from "./openclaw-state-db-schema-additive.js";
-import {
-  assertOpenClawStateDatabaseForMaintenance,
-  OPENCLAW_STATE_SCHEMA_VERSION,
-} from "./openclaw-state-db.js";
+import { assertOpenClawStateDatabaseForMaintenance } from "./openclaw-state-db.js";
 import { OPENCLAW_STATE_MAINTENANCE_SCHEMA_COMPATIBILITY } from "./openclaw-state-schema-compatibility.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js";

@@ -13,11 +13,11 @@ import {
 } from "./openclaw-agent-db.js";
 import { resolveOpenClawAgentSqlitePath } from "./openclaw-agent-db.paths.js";
 import { preflightOpenClawDatabaseSchemas } from "./openclaw-database-preflight.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js";
 import { withOpenClawStateDatabaseReadOnly } from "./openclaw-state-db-readonly.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openExistingOpenClawStateDatabaseReadOnly,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  openOpenClawStateDatabase,
 } from "./openclaw-state-db.js";
 import { resolveOpenClawStateSqlitePath } from "./openclaw-state-db.paths.js";
@@ -15,9 +15,9 @@ import {
  preflightOpenClawStateDatabasePath,
  preflightOpenClawDatabaseSchemas,
 } from "./openclaw-database-preflight.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  openOpenClawStateDatabase,
 } from "./openclaw-state-db.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js";
@@ -17,7 +17,7 @@ import { recordOpenClawDatabaseQuarantine } from "./openclaw-quarantine-store.js
 import {
  confirmOpenClawStateDatabaseIntegrity,
  recordOpenClawStateDatabaseOpenFailure,
-} from "./openclaw-state-db.js";
+} from "./openclaw-state-db-cache.js";
 import { resolveOpenClawStateSqlitePath } from "./openclaw-state-db.paths.js";

 export const OPENCLAW_DATABASE_VERIFY_INITIAL_DELAY_MS = 5 * 60_000;
@@ -27,10 +27,10 @@ import {
  readOpenClawDatabaseQuarantine,
  recordOpenClawDatabaseQuarantine,
 } from "./openclaw-quarantine-store.js";
+import { recordOpenClawStateDatabaseOpenFailure } from "./openclaw-state-db-cache.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  recordOpenClawStateDatabaseOpenFailure,
  repairOpenClawStateDatabaseSchema,
 } from "./openclaw-state-db.js";

@@ -4,11 +4,21 @@ import {
  registerNodeSqliteKyselyQueryErrorHandler,
 } from "../infra/kysely-sync.js";
 import type { SqliteFileGeneration } from "../infra/sqlite-file-generation.js";
+import {
+  confirmSqliteFileIntegrity,
+  type SqliteIntegrityConfirmation,
+} from "../infra/sqlite-integrity.js";
 import { createSqliteTerminalOpenLatch } from "../infra/sqlite-terminal-open-latch.js";
 import { isSqliteCorruptionError } from "../infra/sqlite-transaction.js";
 import { readOpenClawDatabaseQuarantine } from "./openclaw-quarantine-store.js";
-import type { OpenClawStateDatabase } from "./openclaw-state-db-contract.js";
-import { createOpenClawDatabaseVerificationError } from "./openclaw-state-db-maintenance.js";
+import type {
+  OpenClawStateDatabase,
+  OpenClawStateDatabaseOptions,
+} from "./openclaw-state-db-contract.js";
+import {
+  createOpenClawDatabaseVerificationError,
+  resolveDatabasePath,
+} from "./openclaw-state-db-maintenance.js";

 const cachedDatabases = new Map<string, OpenClawStateDatabase>();
 type OpenClawStateDatabaseLifecycleEvent =
@@ -80,7 +90,7 @@ function evictCachedOpenClawStateDatabase(database: OpenClawStateDatabase): bool
 }

 /** Evict an exact cached shared-state owner after a proven corruption read. */
-function evictOpenClawStateDatabaseAfterCorruption(
+export function evictOpenClawStateDatabaseAfterCorruption(
  database: OpenClawStateDatabase,
  error: unknown,
 ): boolean {
@@ -120,6 +130,12 @@ function getOpenClawStateDatabaseIfOpenAtPath(pathname: string): OpenClawStateDa
  return cached?.db.isOpen ? cached : undefined;
 }

+export function getOpenClawStateDatabaseIfOpen(
+  options: OpenClawStateDatabaseOptions = {},
+): OpenClawStateDatabase | undefined {
+  return getOpenClawStateDatabaseIfOpenAtPath(resolveDatabasePath(options));
+}
+
 /** Remove a closed cached owner while fresh-open access is held. */
 function closeStaleCachedOpenClawStateDatabase(database: OpenClawStateDatabase): void {
  if (cachedDatabases.get(database.path) !== database) {
@@ -132,7 +148,7 @@ function closeStaleCachedOpenClawStateDatabase(database: OpenClawStateDatabase):
 }

 /** Latch background verification damage so later opens fail without rescanning. */
-function recordOpenClawStateDatabaseOpenFailure(
+export function recordOpenClawStateDatabaseOpenFailure(
  pathname: string,
  error: Error,
  generation?: SqliteFileGeneration,
@@ -141,7 +157,7 @@ function recordOpenClawStateDatabaseOpenFailure(
 }

 /** Clear a terminal open failure after doctor rewrites the database file. */
-function clearOpenClawStateDatabaseOpenFailure(pathname: string): void {
+export function clearOpenClawStateDatabaseOpenFailure(pathname: string): void {
  terminalOpenLatch.clear(pathname);
 }

@@ -182,8 +198,25 @@ function assertOpenClawStateDatabaseFreshOpenAllowedAtPath(
  }
 }

+/** Reject a fresh shared-state open after known corruption until repair clears it. */
+export function assertOpenClawStateDatabaseFreshOpenAllowed(
+  options: OpenClawStateDatabaseOptions = {},
+): void {
+  const env = options.env ?? process.env;
+  assertOpenClawStateDatabaseFreshOpenAllowedAtPath(resolveDatabasePath(options), env);
+}
+
+/** Reconfirm an advisory worker failure on the live owner connection. */
+export function confirmOpenClawStateDatabaseIntegrity(
+  pathname: string,
+): SqliteIntegrityConfirmation {
+  const resolvedPath = path.resolve(pathname);
+  closeOpenClawStateDatabaseByPath(resolvedPath);
+  return confirmSqliteFileIntegrity(resolvedPath, resolvedPath);
+}
+
 /** Close one cached shared state database handle by exact pathname. */
-function closeOpenClawStateDatabaseByPath(pathname: string): boolean {
+export function closeOpenClawStateDatabaseByPath(pathname: string): boolean {
  const resolvedPath = path.resolve(pathname);
  const database = cachedDatabases.get(resolvedPath);
  if (!database) {
@@ -199,7 +232,7 @@ function closeOpenClawStateDatabaseByPath(pathname: string): boolean {
 }

 /** Close all cached shared state database handles. */
-function closeOpenClawStateDatabase(
+export function closeOpenClawStateDatabase(
  options?: Parameters<OpenClawStateDatabase["walMaintenance"]["close"]>[0],
 ): void {
  for (const database of cachedDatabases.values()) {
@@ -213,12 +246,12 @@ function closeOpenClawStateDatabase(
 }

 /** Test whether any cached shared state database handle is still open. */
-function isOpenClawStateDatabaseOpen(): boolean {
+export function isOpenClawStateDatabaseOpen(): boolean {
  return Array.from(cachedDatabases.values()).some((database) => database.db.isOpen);
 }

 /** Close shared state handles and clear terminal failure latches for test isolation. */
-function closeOpenClawStateDatabaseForTest(): void {
+export function closeOpenClawStateDatabaseForTest(): void {
  closeOpenClawStateDatabase();
  terminalOpenLatch.clearAll();
 }
@@ -4,10 +4,10 @@ import { afterEach, describe, expect, it } from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
 import { requireNodeSqlite } from "../infra/node-sqlite.js";
 import { CLAW_FIRST_USE_ADDITIVE_STATE_COLUMN_DEFINITIONS } from "./openclaw-state-db-additive-columns.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  repairOpenClawStateDatabaseSchema,
 } from "./openclaw-state-db.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js";
@@ -1,9 +1,13 @@
-import { statSync } from "node:fs";
+import { existsSync, statSync } from "node:fs";
 import path from "node:path";
 import type { DatabaseSync } from "node:sqlite";
 import { clearNodeSqliteKyselyCacheForDatabase } from "../infra/kysely-sync.js";
 import { openNodeSqliteDatabase } from "../infra/node-sqlite.js";
-import { prepareSqliteReadOnlyLocationSync } from "../infra/sqlite-readonly-location.js";
+import { assertSqliteIntegrity } from "../infra/sqlite-integrity.js";
+import {
+  prepareSqliteReadOnlyLocation,
+  prepareSqliteReadOnlyLocationSync,
+} from "../infra/sqlite-readonly-location.js";
 import {
  createNewerSqliteSchemaVersionError,
  readSqliteUserVersion,
@@ -11,11 +15,15 @@ import {
 import {
  assertOpenClawStateDatabaseFreshOpenAllowed,
  evictOpenClawStateDatabaseAfterCorruption,
-  getOpenClawStateDatabaseIfOpen,
+  openClawStateDatabaseCache,
+} from "./openclaw-state-db-cache.js";
+import {
  OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
  OPENCLAW_STATE_SCHEMA_VERSION,
+  type OpenClawStateDatabase,
  type OpenClawStateDatabaseOptions,
-} from "./openclaw-state-db.js";
+} from "./openclaw-state-db-contract.js";
+import { assertOpenClawStateDatabaseForMaintenance } from "./openclaw-state-db-maintenance.js";
 import { resolveOpenClawStateSqlitePath } from "./openclaw-state-db.paths.js";

 type OpenClawStateReadOnlyDatabase = {
@@ -55,10 +63,9 @@ function assertSupportedSchemaVersion(db: DatabaseSync, pathname: string): void

 function withOpenClawStateDatabaseReadOnlyIfOpen<T>(
  operation: (database: OpenClawStateReadOnlyDatabase) => T,
-  options: OpenClawStateDatabaseOptions,
  pathname: string,
 ): ReusedOpenClawStateReadOnlyDatabase<T> {
-  const opened = getOpenClawStateDatabaseIfOpen(options);
+  const opened = openClawStateDatabaseCache.getOpenClawStateDatabaseIfOpenAtPath(pathname);
  if (!opened || opened.db.isTransaction) {
    return { reused: false };
  }
@@ -108,7 +115,7 @@ export function withOpenClawStateDatabaseReadOnly<T>(
  // and closing a connection per call made shared-state reads scale with row
  // count. An in-flight transaction is skipped so callers never observe
  // uncommitted rows a fresh read-only connection could not have seen.
-  const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, options, pathname);
+  const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, pathname);
  if (reused.reused) {
    return reused.value;
  }
@@ -121,7 +128,7 @@ export function withExistingOpenClawStateDatabaseReadOnly<T>(
  options: OpenClawStateDatabaseOptions = {},
 ): T | undefined {
  const pathname = resolveReadOnlyPath(options);
-  const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, options, pathname);
+  const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, pathname);
  if (reused.reused) {
    return reused.value;
  }
@@ -141,7 +148,7 @@ export function withExistingOpenClawStateDatabaseArtifactPreservingReadOnly<T>(
  options: OpenClawStateDatabaseOptions = {},
 ): T | undefined {
  const pathname = resolveReadOnlyPath(options);
-  const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, options, pathname);
+  const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, pathname);
  if (reused.reused) {
    return reused.value;
  }
@@ -161,3 +168,64 @@ export function withExistingOpenClawStateDatabaseArtifactPreservingReadOnly<T>(
    prepared.cleanup();
  }
 }
+
+/** Open existing shared state without creating, migrating, chmodding, or configuring it. */
+export async function openExistingOpenClawStateDatabaseReadOnly(
+  options: OpenClawStateDatabaseOptions = {},
+): Promise<OpenClawStateDatabase | undefined> {
+  const pathname = resolveReadOnlyPath(options);
+  if (!existsSync(pathname)) {
+    return undefined;
+  }
+  assertOpenClawStateDatabaseFreshOpenAllowed(options);
+  const prepared = await prepareSqliteReadOnlyLocation(pathname);
+  let db: DatabaseSync;
+  try {
+    db = openNodeSqliteDatabase(prepared.location, { readOnly: true });
+  } catch (error) {
+    prepared.cleanup();
+    throw error;
+  }
+  try {
+    db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`);
+    assertSupportedSchemaVersion(db, pathname);
+    assertSqliteIntegrity(db, pathname);
+    if (readSqliteUserVersion(db) === OPENCLAW_STATE_SCHEMA_VERSION) {
+      assertOpenClawStateDatabaseForMaintenance(db, { pathname });
+    }
+  } catch (error) {
+    try {
+      clearNodeSqliteKyselyCacheForDatabase(db);
+      db.close();
+    } catch {
+      // Preserve the verification failure that explains why the database was refused.
+    }
+    prepared.cleanup();
+    throw error;
+  }
+  let cleanupComplete = false;
+  return {
+    db,
+    path: pathname,
+    walMaintenance: {
+      checkpoint: () => false,
+      // Cleanup can fail transiently after the database closes. Keep the
+      // close contract retryable until one call finishes both responsibilities.
+      close: () => {
+        const wasOpen = db.isOpen;
+        if (!wasOpen && cleanupComplete) {
+          return false;
+        }
+        try {
+          if (wasOpen) {
+            clearNodeSqliteKyselyCacheForDatabase(db);
+            db.close();
+          }
+        } finally {
+          cleanupComplete = prepared.cleanup();
+        }
+        return cleanupComplete;
+      },
+    },
+  };
+}
@@ -2,11 +2,11 @@
 import { afterEach, describe, expect, it } from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
 import { requireNodeSqlite } from "../infra/node-sqlite.js";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js";
 import {
  closeOpenClawStateDatabaseForTest,
  detectOpenClawStateDatabaseSchemaMigrations,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  repairOpenClawStateDatabaseSchema,
 } from "./openclaw-state-db.js";

@@ -7,12 +7,14 @@ import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
 import { executeSqliteQuerySync, getNodeSqliteKysely } from "../infra/kysely-sync.js";
 import { requireNodeSqlite } from "../infra/node-sqlite.js";
 import { isSqliteCorruptionError } from "../infra/sqlite-transaction.js";
+import {
+  evictOpenClawStateDatabaseAfterCorruption,
+  getOpenClawStateDatabaseIfOpen,
+} from "./openclaw-state-db-cache.js";
 import { withOpenClawStateDatabaseReadOnly } from "./openclaw-state-db-readonly.js";
 import type { DB as OpenClawStateKyselyDatabase } from "./openclaw-state-db.generated.js";
 import {
  closeOpenClawStateDatabaseForTest,
-  evictOpenClawStateDatabaseAfterCorruption,
-  getOpenClawStateDatabaseIfOpen,
  openOpenClawStateDatabase,
  runOpenClawStateWriteTransaction,
 } from "./openclaw-state-db.js";
@@ -26,7 +26,10 @@ import { loadTaskRegistryStateFromSqlite } from "../tasks/task-registry.store.sq
 import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js";
 import { VERSION } from "../version.js";
 import { listOpenClawRegisteredAgentDatabases } from "./openclaw-agent-db-registry.js";
-import { FIRST_USE_STATE_TABLES } from "./openclaw-state-db-contract.js";
+import {
+  FIRST_USE_STATE_TABLES,
+  OPENCLAW_STATE_SCHEMA_VERSION,
+} from "./openclaw-state-db-contract.js";
 import {
  findOpenClawStateDatabaseSchemaMigrationRequiredError,
  OpenClawStateDatabaseSchemaMigrationRequiredError,
@@ -40,7 +43,6 @@ import {
  OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
  openExistingOpenClawStateDatabaseReadOnly,
  openOpenClawStateDatabase,
-  OPENCLAW_STATE_SCHEMA_VERSION,
  repairOpenClawStateDatabaseSchema,
  repairOpenClawStateDatabaseSchemaIfNeeded,
  runOpenClawStateWriteTransaction,
@@ -1,6 +1,5 @@
 // OpenClaw state database manages shared persisted state and migrations.
 import { existsSync } from "node:fs";
-import path from "node:path";
 import type { DatabaseSync } from "node:sqlite";
 import {
  clearNodeSqliteKyselyCacheForDatabase,
@@ -9,19 +8,22 @@ import {
  getNodeSqliteKysely,
 } from "../infra/kysely-sync.js";
 import { openNodeSqliteDatabase } from "../infra/node-sqlite.js";
+import {
+  normalizeSqliteNonNegativeInteger,
+  readSqliteBusyTimeout,
+  runWithSqliteBusyTimeout,
+  setSqliteBusyTimeout,
+  type SqliteLockFailureReporting,
+} from "../infra/sqlite-busy-timeout.js";
 import { createSqliteLifecycleAggregateError } from "../infra/sqlite-coordinator.js";
-import type { SqliteFileGeneration } from "../infra/sqlite-file-generation.js";
 import {
  repairCanonicalSqliteIndexes,
  verifyAndRepairCanonicalSqliteIndexes,
 } from "../infra/sqlite-index-schema.js";
 import {
  assertSqliteIntegrity,
-  confirmSqliteFileIntegrity,
  isTerminalSqliteIntegrityError,
-  type SqliteIntegrityConfirmation,
 } from "../infra/sqlite-integrity.js";
-import { prepareSqliteReadOnlyLocation } from "../infra/sqlite-readonly-location.js";
 import { assertSqliteSchemaTablesPresent } from "../infra/sqlite-schema-contract.js";
 import { migrateSqliteSchemaToStrictInTransaction } from "../infra/sqlite-strict.js";
 import {
@@ -40,7 +42,10 @@ import { createSubsystemLogger } from "../logging/subsystem.js";
 import { VERSION } from "../version.js";
 import { clearOpenClawDatabaseQuarantine } from "./openclaw-quarantine-store.js";
 import { repairAuditEventsSchema } from "./openclaw-state-db-audit-migration.js";
-import { openClawStateDatabaseCache as stateDbCache } from "./openclaw-state-db-cache.js";
+import {
+  getOpenClawStateDatabaseIfOpen,
+  openClawStateDatabaseCache as stateDbCache,
+} from "./openclaw-state-db-cache.js";
 import {
  OPENCLAW_DATABASE_SCHEMA_DOCS_URL,
  LAZY_ADDITIVE_STATE_TABLES,
@@ -83,12 +88,21 @@ import type { DB as OpenClawStateKyselyDatabase } from "./openclaw-state-db.gene
 import { describeAgentPathMigration, warnAgentPathMigration } from "./openclaw-state-db.paths.js";
 import {
  assertOpenClawStateWriteAllowed,
+  isOpenClawStateWriteContentionError,
  OpenClawStateOwnershipError,
  runWithOpenClawStateWriteAccess,
 } from "./openclaw-state-ownership.js";
 import { getOpenClawStateRuntimeSchema } from "./openclaw-state-schema-compatibility.js";
 import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js";
-export { registerOpenClawStateDatabaseLifecycleListener } from "./openclaw-state-db-cache.js";
+export {
+  clearOpenClawStateDatabaseOpenFailure,
+  closeOpenClawStateDatabase,
+  closeOpenClawStateDatabaseByPath,
+  closeOpenClawStateDatabaseForTest,
+  isOpenClawStateDatabaseOpen,
+  registerOpenClawStateDatabaseLifecycleListener,
+} from "./openclaw-state-db-cache.js";
+export { openExistingOpenClawStateDatabaseReadOnly } from "./openclaw-state-db-readonly.js";

 const STATE_MIGRATION_ASSERTIONS = {
  5: assertOpenClawStateDatabaseV5ForMigration,
@@ -97,11 +111,7 @@ const STATE_MIGRATION_ASSERTIONS = {
  8: assertOpenClawStateDatabaseV8ForMigration,
 } as const;

-export {
-  OPENCLAW_DATABASE_SCHEMA_DOCS_URL,
-  OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
-  OPENCLAW_STATE_SCHEMA_VERSION,
-};
+export { OPENCLAW_DATABASE_SCHEMA_DOCS_URL, OPENCLAW_SQLITE_BUSY_TIMEOUT_MS };
 export type {
  OpenClawStateDatabase,
  OpenClawStateDatabaseOptions,
@@ -115,37 +125,6 @@ export { ensureOpenClawStatePermissions } from "./openclaw-state-db-permissions.
 export { detectOpenClawStateDatabaseSchemaMigrations } from "./openclaw-state-db-schema-repair.js";
 export { withOpenClawStateStartupMigrationCheckpointDatabase } from "./openclaw-state-db-startup-checkpoint.js";

-/** Reconfirm an advisory worker failure on the live owner connection. */
-export function confirmOpenClawStateDatabaseIntegrity(
-  pathname: string,
-): SqliteIntegrityConfirmation {
-  const resolvedPath = path.resolve(pathname);
-  closeOpenClawStateDatabaseByPath(resolvedPath);
-  return confirmSqliteFileIntegrity(resolvedPath, resolvedPath);
-}
-
-/** Latch background verification damage so later opens fail without rescanning. */
-export function recordOpenClawStateDatabaseOpenFailure(
-  pathname: string,
-  error: Error,
-  generation?: SqliteFileGeneration,
-): boolean {
-  return stateDbCache.recordOpenClawStateDatabaseOpenFailure(pathname, error, generation);
-}
-
-/** Clear a terminal open failure after doctor rewrites the database file. */
-export function clearOpenClawStateDatabaseOpenFailure(pathname: string): void {
-  stateDbCache.clearOpenClawStateDatabaseOpenFailure(pathname);
-}
-
-/** Reject a fresh shared-state open after known corruption until repair clears it. */
-export function assertOpenClawStateDatabaseFreshOpenAllowed(
-  options: OpenClawStateDatabaseOptions = {},
-): void {
-  const env = options.env ?? process.env;
-  stateDbCache.assertOpenClawStateDatabaseFreshOpenAllowedAtPath(resolveDatabasePath(options), env);
-}
-
 type OpenClawStateMetadataDatabase = Pick<OpenClawStateKyselyDatabase, "schema_meta">;
 const stateDbLog = createSubsystemLogger("state/db");

@@ -272,7 +251,7 @@ function repairOpenClawStateDatabaseSchemaWithWriteAccess(
      },
    );
    const quarantineCleared = clearOpenClawDatabaseQuarantine(pathname, { env });
-    clearOpenClawStateDatabaseOpenFailure(pathname);
+    stateDbCache.clearOpenClawStateDatabaseOpenFailure(pathname);
    return {
      changes,
      warnings: quarantineCleared
@@ -367,7 +346,12 @@ export function repairOpenClawStateDatabaseSchemaIfNeeded(
  );
 }

-function ensureSchema(db: DatabaseSync, pathname: string, env: NodeJS.ProcessEnv): void {
+function ensureSchema(
+  db: DatabaseSync,
+  pathname: string,
+  env: NodeJS.ProcessEnv,
+  busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+): void {
  const now = Date.now();
  const kysely = getNodeSqliteKysely<OpenClawStateMetadataDatabase>(db);
  // Rebuilding referenced tables requires disabling FK enforcement before BEGIN.
@@ -461,7 +445,7 @@ function ensureSchema(db: DatabaseSync, pathname: string, env: NodeJS.ProcessEnv
        warnAgentPathMigration(stateDbLog, pathMigration, pathname);
      },
      {
-        busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+        busyTimeoutMs,
        databaseLabel: pathname,
        operationLabel: "state.schema.ensure",
      },
@@ -471,69 +455,6 @@ function ensureSchema(db: DatabaseSync, pathname: string, env: NodeJS.ProcessEnv
  }
 }

-/** Open existing shared state without creating, migrating, chmodding, or configuring it. */
-export async function openExistingOpenClawStateDatabaseReadOnly(
-  options: OpenClawStateDatabaseOptions = {},
-): Promise<OpenClawStateDatabase | undefined> {
-  const pathname = resolveDatabasePath(options);
-  if (!existsSync(pathname)) {
-    return undefined;
-  }
-  assertOpenClawStateDatabaseFreshOpenAllowed(options);
-  const prepared = await prepareSqliteReadOnlyLocation(pathname);
-  let db: DatabaseSync;
-  try {
-    db = openNodeSqliteDatabase(prepared.location, {
-      readOnly: true,
-    });
-  } catch (error) {
-    prepared.cleanup();
-    throw error;
-  }
-  try {
-    db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`);
-    assertSupportedSchemaVersion(db, pathname);
-    assertSqliteIntegrity(db, pathname);
-    if (readSqliteUserVersion(db) === OPENCLAW_STATE_SCHEMA_VERSION) {
-      assertOpenClawStateDatabaseForMaintenance(db, { pathname });
-    }
-  } catch (error) {
-    try {
-      clearNodeSqliteKyselyCacheForDatabase(db);
-      db.close();
-    } catch {
-      // Preserve the verification failure that explains why the database was refused.
-    }
-    prepared.cleanup();
-    throw error;
-  }
-  let cleanupComplete = false;
-  return {
-    db,
-    path: pathname,
-    walMaintenance: {
-      checkpoint: () => false,
-      // Cleanup can fail transiently after the database closes. Keep the
-      // close contract retryable until one call finishes both responsibilities.
-      close: () => {
-        const wasOpen = db.isOpen;
-        if (!wasOpen && cleanupComplete) {
-          return false;
-        }
-        try {
-          if (wasOpen) {
-            clearNodeSqliteKyselyCacheForDatabase(db);
-            db.close();
-          }
-        } finally {
-          cleanupComplete = prepared.cleanup();
-        }
-        return cleanupComplete;
-      },
-    },
-  };
-}
-
 function assertCurrentStateRuntimeSchema(database: DatabaseSync, pathname: string): void {
  assertCanonicalStateSchemaShape(database, pathname);
  assertOpenClawStateDatabaseForMaintenance(database, { pathname });
@@ -542,8 +463,9 @@ function assertCurrentStateRuntimeSchema(database: DatabaseSync, pathname: strin
 function assertStateDatabaseIntegrityBeforeMutation(
  database: DatabaseSync,
  pathname: string,
+  busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
 ): void {
-  database.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`);
+  setSqliteBusyTimeout(database, busyTimeoutMs);
  const userVersion = readSqliteUserVersion(database);
  const hasApplicationSchema = database
    .prepare("SELECT 1 FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' LIMIT 1")
@@ -567,48 +489,58 @@ function assertStateDatabaseIntegrityBeforeMutation(
 function openUnpublishedOpenClawStateDatabase(
  pathname: string,
  env: NodeJS.ProcessEnv,
+  busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+  lockFailureReporting: SqliteLockFailureReporting = "report",
 ): OpenClawStateDatabase {
  ensureOpenClawStatePermissions(pathname, env);
  const db = openNodeSqliteDatabase(pathname);
  enableNodeSqliteKyselyStatementCache(db);
-  const walMaintenance = (() => {
-    let maintenance: SqliteWalMaintenance | undefined;
-    try {
-      db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`);
-      assertSupportedSchemaVersion(db, pathname);
-      assertStateDatabaseIntegrityBeforeMutation(db, pathname);
-      configureSqlitePreSchemaPragmas(db, {
-        busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
-      });
-      maintenance = configureSqliteConnectionPragmas(db, {
-        busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
-        databaseLabel: "openclaw-state",
-        databasePath: pathname,
-        foreignKeys: true,
-        synchronous: "NORMAL",
-      });
-      ensureSchema(db, pathname, env);
-      return maintenance;
-    } catch (err) {
-      maintenance?.close();
-      db.close();
-      if (
-        err instanceof Error &&
-        (err.name === "SqliteSchemaVersionError" || isTerminalSqliteIntegrityError(err))
-      ) {
-        recordOpenClawStateDatabaseOpenFailure(pathname, err);
+  setSqliteBusyTimeout(db, busyTimeoutMs);
+  const walMaintenance = runWithSqliteBusyTimeout(
+    db,
+    busyTimeoutMs,
+    () => {
+      let maintenance: SqliteWalMaintenance | undefined;
+      try {
+        setSqliteBusyTimeout(db, busyTimeoutMs);
+        assertSupportedSchemaVersion(db, pathname);
+        assertStateDatabaseIntegrityBeforeMutation(db, pathname, busyTimeoutMs);
+        configureSqlitePreSchemaPragmas(db, {
+          busyTimeoutMs,
+        });
+        maintenance = configureSqliteConnectionPragmas(db, {
+          busyTimeoutMs,
+          databaseLabel: "openclaw-state",
+          databasePath: pathname,
+          foreignKeys: true,
+          synchronous: "NORMAL",
+        });
+        ensureSchema(db, pathname, env, busyTimeoutMs);
+        return maintenance;
+      } catch (err) {
+        maintenance?.close();
+        db.close();
+        if (
+          err instanceof Error &&
+          (err.name === "SqliteSchemaVersionError" || isTerminalSqliteIntegrityError(err))
+        ) {
+          stateDbCache.recordOpenClawStateDatabaseOpenFailure(pathname, err);
+        }
+        throw err;
      }
-      throw err;
-    }
-  })();
+    },
+    { lockFailureReporting },
+  );
  ensureOpenClawStatePermissions(pathname, env);
  return { db, path: pathname, walMaintenance };
 }

 /** Open or return a cached shared state database after schema and migration checks. */

-export function openOpenClawStateDatabase(
+function openOpenClawStateDatabaseWithBusyTimeout(
  options: OpenClawStateDatabaseOptions = {},
+  busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+  lockFailureReporting: SqliteLockFailureReporting = "report",
 ): OpenClawStateDatabase {
  const env = options.env ?? process.env;
  if (options.database) {
@@ -634,7 +566,7 @@ export function openOpenClawStateDatabase(
    return cached;
  }
  try {
-    assertOpenClawStateDatabaseFreshOpenAllowed(options);
+    stateDbCache.assertOpenClawStateDatabaseFreshOpenAllowedAtPath(pathname, env);
  } catch (error) {
    stateDbCache.recordOpenClawStateDatabaseLifecycleOpenError(pathname, error);
    throw error;
@@ -642,18 +574,25 @@ export function openOpenClawStateDatabase(
  let unpublished: OpenClawStateDatabase | undefined;
  try {
    unpublished = runWithOpenClawStateWriteAccess(
-      { databasePath: pathname, env },
+      { databasePath: pathname, busyTimeoutMs, env },
      "fresh state database open",
      () => {
        if (cached) {
          // A closed handle can leave Kysely and WAL helpers cached; clear both under access.
          stateDbCache.closeStaleCachedOpenClawStateDatabase(cached);
        }
-        return (unpublished = openUnpublishedOpenClawStateDatabase(pathname, env));
+        return (unpublished = openUnpublishedOpenClawStateDatabase(
+          pathname,
+          env,
+          busyTimeoutMs,
+          lockFailureReporting,
+        ));
      },
    );
  } catch (error) {
-    stateDbCache.recordOpenClawStateDatabaseLifecycleOpenError(pathname, error);
+    if (lockFailureReporting === "report" || !isOpenClawStateWriteContentionError(error)) {
+      stateDbCache.recordOpenClawStateDatabaseLifecycleOpenError(pathname, error);
+    }
    if (!unpublished) {
      throw error;
    }
@@ -670,6 +609,38 @@ export function openOpenClawStateDatabase(
  return stateDbCache.publishOpenClawStateDatabase(unpublished);
 }

+/** Open or return a cached shared state database after schema and migration checks. */
+export function openOpenClawStateDatabase(
+  options: OpenClawStateDatabaseOptions = {},
+): OpenClawStateDatabase {
+  return openOpenClawStateDatabaseWithBusyTimeout(options);
+}
+
+/** Run one operation through the shared owner without waiting synchronously on SQLite locks. */
+export function runWithOpenClawStateBusyTimeout<T>(
+  operation: (database: OpenClawStateDatabase) => T,
+  options: OpenClawStateDatabaseOptions,
+  busyTimeoutMs: number,
+): T {
+  const normalizedTimeoutMs = normalizeSqliteNonNegativeInteger(busyTimeoutMs, "busyTimeoutMs");
+  const existing = options.database ?? getOpenClawStateDatabaseIfOpen(options);
+  if (existing) {
+    return runWithSqliteBusyTimeout(existing.db, normalizedTimeoutMs, () => operation(existing), {
+      lockFailureReporting: "suppress",
+    });
+  }
+  const opened = openOpenClawStateDatabaseWithBusyTimeout(options, normalizedTimeoutMs, "suppress");
+  try {
+    return runWithSqliteBusyTimeout(opened.db, normalizedTimeoutMs, () => operation(opened), {
+      lockFailureReporting: "suppress",
+    });
+  } finally {
+    if (opened.db.isOpen) {
+      setSqliteBusyTimeout(opened.db, OPENCLAW_SQLITE_BUSY_TIMEOUT_MS);
+    }
+  }
+}
+
 /** Run a synchronous immediate transaction against the shared state database. */
 export function runOpenClawStateWriteTransaction<T>(
  operation: (database: OpenClawStateDatabase) => T,
@@ -702,7 +673,7 @@ export function runOpenClawStateWriteTransaction<T>(
        return operation(database);
      },
      {
-        busyTimeoutMs: transactionOptions.busyTimeoutMs ?? OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+        busyTimeoutMs: transactionOptions.busyTimeoutMs ?? readSqliteBusyTimeout(database.db),
        databaseLabel: database.path,
        ...transactionOptions,
        operationLabel: transactionOptions.operationLabel ?? "state.write",
@@ -722,45 +693,3 @@ export function runOpenClawStateWriteTransaction<T>(
  }
  return result;
 }
-
-/**
- * Return a shared state handle this process already holds open, if any.
- *
- * Read-only callers use this to avoid opening a connection per call; it never
- * creates, repairs, or registers a handle.
- */
-export function getOpenClawStateDatabaseIfOpen(
-  options: OpenClawStateDatabaseOptions = {},
-): OpenClawStateDatabase | undefined {
-  return stateDbCache.getOpenClawStateDatabaseIfOpenAtPath(resolveDatabasePath(options));
-}
-
-/** Evict an exact cached shared-state owner after a proven corruption read. */
-export function evictOpenClawStateDatabaseAfterCorruption(
-  database: OpenClawStateDatabase,
-  error: unknown,
-): boolean {
-  return stateDbCache.evictOpenClawStateDatabaseAfterCorruption(database, error);
-}
-
-/** Close one cached shared state database handle by exact pathname. */
-export function closeOpenClawStateDatabaseByPath(pathname: string): boolean {
-  return stateDbCache.closeOpenClawStateDatabaseByPath(pathname);
-}
-
-/** Close all cached shared state database handles. */
-export function closeOpenClawStateDatabase(
-  options?: Parameters<typeof stateDbCache.closeOpenClawStateDatabase>[0],
-): void {
-  stateDbCache.closeOpenClawStateDatabase(options);
-}
-
-/** Test whether any cached shared state database handle is still open. */
-export function isOpenClawStateDatabaseOpen(): boolean {
-  return stateDbCache.isOpenClawStateDatabaseOpen();
-}
-
-/** Close shared state handles and clear terminal failure latches for test isolation. */
-export function closeOpenClawStateDatabaseForTest(): void {
-  stateDbCache.closeOpenClawStateDatabaseForTest();
-}
@@ -8,6 +8,7 @@ import {
  executeSqliteQueryTakeFirstSync,
  getNodeSqliteKysely,
 } from "../infra/kysely-sync.js";
+import { runWithSqliteBusyTimeout } from "../infra/sqlite-busy-timeout.js";
 import { isSqliteLockError } from "../infra/sqlite-transaction.js";
 import { loggingState } from "../logging/state.js";
 import type { DB as OpenClawStateKyselyDatabase } from "./openclaw-state-db.generated.js";
@@ -180,29 +181,6 @@ function validateOptions(options: OpenClawStateLeaseOptions) {
  };
 }

-function readBusyTimeout(database: DatabaseSync): number {
-  const row = database // sqlite-allow-raw -- Narrow connection primitive for bounded lease admission.
-    .prepare("PRAGMA busy_timeout")
-    .get() as { busy_timeout?: unknown; timeout?: unknown } | undefined;
-  const value = row?.busy_timeout ?? row?.timeout;
-  return typeof value === "bigint" ? Number(value) : Number(value ?? 0);
-}
-
-function withBusyTimeout<T>(database: DatabaseSync, busyTimeoutMs: number, run: () => T): T {
-  const previousBusyTimeoutMs = readBusyTimeout(database);
-  if (previousBusyTimeoutMs === busyTimeoutMs) {
-    return run();
-  }
-  database.exec(`PRAGMA busy_timeout = ${busyTimeoutMs}`); // sqlite-allow-raw -- Bound synchronous lease admission to waitMs.
-  try {
-    return run();
-  } finally {
-    if (database.isOpen) {
-      database.exec(`PRAGMA busy_timeout = ${previousBusyTimeoutMs}`); // sqlite-allow-raw -- Restore canonical connection policy.
-    }
-  }
-}
-
 function withLeaseWriteTransaction<T>(
  database: OpenClawStateLeaseDatabase,
  operationLabel: string,
@@ -216,7 +194,7 @@ function withLeaseWriteTransaction<T>(
      database.options,
      { operationLabel, busyTimeoutMs },
    );
-  return withBusyTimeout(stateDatabase.db, busyTimeoutMs, run);
+  return runWithSqliteBusyTimeout(stateDatabase.db, busyTimeoutMs, run);
 }

 function withLeaseRead<T>(
@@ -15,10 +15,10 @@ import { sha256HexPrefixCore } from "../infra/crypto-digest.js";
 import { requireNodeSqlite, resolveImmutableSqliteFileUri } from "../infra/node-sqlite.js";
 import * as sqliteReadonlyLocation from "../infra/sqlite-readonly-location.js";
 import { withEnv, withEnvAsync } from "../test-utils/env.js";
+import { getOpenClawStateDatabaseIfOpen } from "./openclaw-state-db-cache.js";
 import { withOpenClawStateStartupMigrationCheckpointDatabase } from "./openclaw-state-db-startup-checkpoint.js";
 import {
  closeOpenClawStateDatabaseForTest,
-  getOpenClawStateDatabaseIfOpen,
  openExistingOpenClawStateDatabaseReadOnly,
  openOpenClawStateDatabase,
  repairOpenClawStateDatabaseSchema,
@@ -10,6 +10,7 @@ import {
  openNodeSqliteDatabase,
  tryAcquireExclusiveSqliteCoordinator,
 } from "../infra/node-sqlite.js";
+import { normalizeSqliteNonNegativeInteger } from "../infra/sqlite-busy-timeout.js";
 import {
  createSqliteLifecycleAggregateError,
  ensurePrivateSqliteCoordinatorDirectory,
@@ -21,6 +22,7 @@ import {
  prepareSqliteReadOnlyLocation,
  prepareSqliteReadOnlyLocationSync,
 } from "../infra/sqlite-readonly-location.js";
+import { isSqliteLockError } from "../infra/sqlite-transaction.js";
 import { OPENCLAW_SQLITE_BUSY_TIMEOUT_MS } from "./openclaw-state-db-contract.js";
 import { tableExists } from "./openclaw-state-db-schema-helpers.js";
 import { resolveOpenClawStateDirForDatabasePath } from "./openclaw-state-db.paths.js";
@@ -38,6 +40,17 @@ export type OpenClawExternalStateOwnership = {

 export class OpenClawStateOwnershipError extends Error {}

+class OpenClawStateOwnershipContentionError extends SqliteCoordinatorError {
+  constructor() {
+    super("another OpenClaw process is changing shared state ownership");
+    this.name = "OpenClawStateOwnershipContentionError";
+  }
+}
+
+export function isOpenClawStateWriteContentionError(error: unknown): boolean {
+  return error instanceof OpenClawStateOwnershipContentionError || isSqliteLockError(error);
+}
+
 export class OpenClawStateOwnershipMetadataError extends OpenClawStateOwnershipError {
  constructor(
    readonly databasePath: string,
@@ -160,6 +173,7 @@ function inspectJournalAwarePublicOwnership(

 function inspectOpenClawStateOwnershipAtPathWhileCoordinatorHeld(
  databasePath: string,
+  busyTimeoutMs: number,
 ): OpenClawExternalStateOwnership | null {
  const resolvedPath = path.resolve(databasePath);
  if (!existsSync(resolvedPath)) {
@@ -169,9 +183,7 @@ function inspectOpenClawStateOwnershipAtPathWhileCoordinatorHeld(
  // Inspect the live committed view without cloning a potentially busy family.
  const database = openNodeSqliteDatabase(resolvedPath);
  try {
-    database.exec(
-      `PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS}; PRAGMA trusted_schema = OFF;`,
-    );
+    database.exec(`PRAGMA busy_timeout = ${busyTimeoutMs}; PRAGMA trusted_schema = OFF;`);
    return inspectOpenClawStateOwnershipFromDatabase(database, resolvedPath);
  } finally {
    database.close();
@@ -187,7 +199,10 @@ function resolveOpenClawStateOwnershipCoordinatorPath(databasePath: string): str
  );
 }

-function acquireOpenClawStateOwnershipCoordinator(databasePath: string): {
+function acquireOpenClawStateOwnershipCoordinator(
+  databasePath: string,
+  busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+): {
  release: () => void;
 } {
  const coordinatorPath = resolveOpenClawStateOwnershipCoordinatorPath(databasePath);
@@ -196,10 +211,10 @@ function acquireOpenClawStateOwnershipCoordinator(databasePath: string): {
    "state ownership coordinator",
  );
  const coordinator = tryAcquireExclusiveSqliteCoordinator(coordinatorPath, {
-    busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+    busyTimeoutMs,
  });
  if (!coordinator) {
-    throw new SqliteCoordinatorError("another OpenClaw process is changing shared state ownership");
+    throw new OpenClawStateOwnershipContentionError();
  }
  return coordinator;
 }
@@ -240,14 +255,19 @@ function assertOwnershipAllowsWrite(
 /** Fence and hold one path-based mutation until its main-file preamble is complete. */
 function acquireOpenClawStateWriteAccess(options: {
  databasePath: string;
+  busyTimeoutMs?: number;
  env?: NodeJS.ProcessEnv;
 }): { release: () => void } {
  const resolvedPath = path.resolve(options.databasePath);
-  const access = acquireOpenClawStateOwnershipCoordinator(resolvedPath);
+  const busyTimeoutMs = normalizeSqliteNonNegativeInteger(
+    options.busyTimeoutMs ?? OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
+    "busyTimeoutMs",
+  );
+  const access = acquireOpenClawStateOwnershipCoordinator(resolvedPath, busyTimeoutMs);
  try {
    quarantineOrphanedSqliteSidecars(resolvedPath);
    assertOwnershipAllowsWrite(
-      inspectOpenClawStateOwnershipAtPathWhileCoordinatorHeld(resolvedPath),
+      inspectOpenClawStateOwnershipAtPathWhileCoordinatorHeld(resolvedPath, busyTimeoutMs),
      resolvedPath,
      options.env ?? process.env,
    );
@@ -273,7 +293,7 @@ function acquireOpenClawStateWriteAccess(options: {
 }

 export function runWithOpenClawStateWriteAccess<T>(
-  options: { databasePath: string; env?: NodeJS.ProcessEnv },
+  options: { databasePath: string; busyTimeoutMs?: number; env?: NodeJS.ProcessEnv },
  operationLabel: string,
  operation: () => T,
 ): T {
@@ -2,9 +2,9 @@ import { mkdtempSync, readFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
+import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js";
 import { tableExists } from "./openclaw-state-db-schema-helpers.js";
 import {
-  OPENCLAW_STATE_SCHEMA_VERSION,
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
 } from "./openclaw-state-db.js";
@@ -726,7 +726,6 @@ describe("collectMissingPackPaths", () => {
      "dist/agents/compaction-planning.worker.js",
      "dist/agents/model-provider-auth.worker.js",
      "dist/agents/prepared-model-catalog.worker.js",
-      "dist/audit/audit-event-writer.worker.js",
      "dist/config/sessions/session-accessor.sqlite-archive.worker.js",
      "dist/config/sessions/session-transcript-reconcile.worker.js",
      "dist/state/openclaw-database-verify.worker.js",
@@ -767,7 +766,6 @@ describe("collectMissingPackPaths", () => {
        "dist/agents/compaction-planning.worker.js",
        "dist/agents/model-provider-auth.worker.js",
        "dist/agents/prepared-model-catalog.worker.js",
-        "dist/audit/audit-event-writer.worker.js",
        "dist/config/sessions/session-accessor.sqlite-archive.worker.js",
        "dist/config/sessions/session-transcript-reconcile.worker.js",
        "dist/state/openclaw-database-verify.worker.js",
@@ -201,7 +201,6 @@ describe("production lint suppressions", () => {
        "src/agents/agent-bundle-mcp-runtime.ts|unicorn/prefer-add-event-listener|1",
        "src/agents/agent-tools.abort.ts|typescript/prefer-promise-reject-errors|1",
        "src/agents/sessions/session-manager-entries.ts|unicorn/prefer-structured-clone|1",
-        "src/audit/audit-event-writer.ts|unicorn/require-post-message-target-origin|2",
        "src/channels/plugins/channel-runtime-surface.types.ts|typescript/no-unnecessary-type-parameters|1",
        "src/channels/plugins/contracts/test-helpers.ts|typescript/no-unnecessary-type-parameters|1",
        "src/channels/plugins/types.plugin.ts|typescript/no-explicit-any|1",
@@ -358,7 +358,6 @@ function buildCoreDistEntries(): Record<string, string> {
    "agents/compaction-planning.worker": "src/agents/compaction-planning.worker.ts",
    "agents/model-provider-auth.worker": "src/agents/model-provider-auth.worker.ts",
    "agents/prepared-model-catalog.worker": "src/agents/prepared-model-catalog.worker.ts",
-    "audit/audit-event-writer.worker": "src/audit/audit-event-writer.worker.ts",
    "config/sessions/session-accessor.sqlite-archive.worker":
      "src/config/sessions/session-accessor.sqlite-archive.worker.ts",
    "config/sessions/session-transcript-reconcile.worker":