fix(outbound): publish conversation bindings after SQLite commit (#101023)

* fix(outbound): restore current-conversation binding map when persist fails

The generic current-conversation binding mutators (bind/touch/unbind/
prune-on-read) changed the process-wide in-memory map before calling
writePersistedBindings(), with no rollback. A throwing SQLite write left
the map ahead of disk; because bindingsLoaded is a one-time flag, the
diverged bindings were served until restart, routing inbound messages to
wrong or deleted session targets while the caller already saw the throw.

Add persistBindingsOrRestore(): snapshot the map before mutation and, on
write failure, restore it and rethrow. Covers all six write sites. Mirrors
the cron rollback precedent (#99960, src/cron/service/ops.ts persistOrRestore).

Fault-injection tests assert the map reverts for each path.

* fix(outbound): publish conversation bindings after SQLite commit

* chore(outbound): align binding files with current main

* fix(outbound): publish conversation bindings after SQLite commit

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
Co-authored-by: Peter Steinberger <peter@steipete.me>

src/infra/outbound/current-conversation-bindings.test.ts

@@ -8,6 +8,7 @@ import { setActivePluginRegistry } from "../../plugins/runtime.js";
 import type { DB as OpenClawStateKyselyDatabase } from "../../state/openclaw-state-db.generated.js";
 import {
   closeOpenClawStateDatabaseForTest,
+  openOpenClawStateDatabase,
   runOpenClawStateWriteTransaction,
 } from "../../state/openclaw-state-db.js";
 import { createTestRegistry } from "../../test-utils/channel-plugins.js";
@@ -130,6 +131,45 @@ function setMinimalCurrentConversationRegistry(): void {
   );
 }
 
+async function withReadOnlyStateDatabase<T>(run: () => T | Promise<T>): Promise<T> {
+  const { db } = openOpenClawStateDatabase();
+  db.exec("PRAGMA query_only = ON");
+  try {
+    return await run();
+  } finally {
+    db.exec("PRAGMA query_only = OFF");
+  }
+}
+
+function workspaceConversation(conversationId: string) {
+  return {
+    channel: "workspace",
+    accountId: "default",
+    conversationId,
+  };
+}
+
+async function bindWorkspaceConversation(
+  conversationId: string,
+  options: {
+    targetSessionKey?: string;
+    ttlMs?: number;
+    metadata?: Record<string, unknown>;
+  } = {},
+): Promise<SessionBindingRecord | null> {
+  return bindGenericCurrentConversation({
+    targetSessionKey: options.targetSessionKey ?? "agent:codex:acp:workspace-dm",
+    targetKind: "session",
+    conversation: workspaceConversation(conversationId),
+    ...(options.ttlMs === undefined ? {} : { ttlMs: options.ttlMs }),
+    ...(options.metadata === undefined ? {} : { metadata: options.metadata }),
+  });
+}
+
+function resolveWorkspaceConversation(conversationId: string): SessionBindingRecord | null {
+  return resolveGenericCurrentConversationBinding(workspaceConversation(conversationId));
+}
+
 describe("generic current-conversation bindings", () => {
   let previousStateDir: string | undefined;
   let testStateDir = "";
@@ -458,4 +498,156 @@ describe("generic current-conversation bindings", () => {
       },
     );
   });
+
+  describe("SQLite write failures", () => {
+    it("keeps a replacement bind out of memory and disk", async () => {
+      await bindWorkspaceConversation("user:U1", {
+        targetSessionKey: "agent:codex:acp:session-a",
+      });
+
+      await expect(
+        withReadOnlyStateDatabase(() =>
+          bindWorkspaceConversation("user:U1", {
+            targetSessionKey: "agent:codex:acp:session-b",
+          }),
+        ),
+      ).rejects.toThrow();
+
+      expect(resolveWorkspaceConversation("user:U1")?.targetSessionKey).toBe(
+        "agent:codex:acp:session-a",
+      );
+      testing.resetCurrentConversationBindingsForTests();
+      closeOpenClawStateDatabaseForTest();
+      expect(resolveWorkspaceConversation("user:U1")?.targetSessionKey).toBe(
+        "agent:codex:acp:session-a",
+      );
+    });
+
+    it("keeps a failed touch out of memory and disk", async () => {
+      const bound = expectSessionBinding(
+        await bindWorkspaceConversation("user:U1", { metadata: { label: "workspace-dm" } }),
+      );
+      const originalActivity = bound.metadata?.lastActivityAt;
+
+      await expect(
+        withReadOnlyStateDatabase(() =>
+          touchGenericCurrentConversationBinding(bound.bindingId, 9_999_999),
+        ),
+      ).rejects.toThrow();
+
+      expect(resolveWorkspaceConversation("user:U1")?.metadata?.lastActivityAt).toBe(
+        originalActivity,
+      );
+      testing.resetCurrentConversationBindingsForTests();
+      expect(resolveWorkspaceConversation("user:U1")?.metadata?.lastActivityAt).toBe(
+        originalActivity,
+      );
+    });
+
+    it("keeps a binding when unbind by id fails", async () => {
+      const bound = expectSessionBinding(await bindWorkspaceConversation("user:U1"));
+
+      await expect(
+        withReadOnlyStateDatabase(() =>
+          unbindGenericCurrentConversationBindings({
+            bindingId: bound.bindingId,
+            reason: "test cleanup",
+          }),
+        ),
+      ).rejects.toThrow();
+
+      expect(resolveWorkspaceConversation("user:U1")).not.toBeNull();
+      testing.resetCurrentConversationBindingsForTests();
+      expect(resolveWorkspaceConversation("user:U1")).not.toBeNull();
+    });
+
+    it("keeps every matching binding when unbind by session fails", async () => {
+      const targetSessionKey = "agent:codex:acp:shared";
+      await bindWorkspaceConversation("user:U1", { targetSessionKey });
+      await bindWorkspaceConversation("user:U2", { targetSessionKey });
+
+      await expect(
+        withReadOnlyStateDatabase(() =>
+          unbindGenericCurrentConversationBindings({
+            targetSessionKey,
+            reason: "test cleanup",
+          }),
+        ),
+      ).rejects.toThrow();
+
+      expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2);
+      testing.resetCurrentConversationBindingsForTests();
+      expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2);
+    });
+
+    it("keeps an expired binding when prune-on-resolve fails", async () => {
+      vi.useFakeTimers();
+      vi.setSystemTime(new Date(1_000_000));
+      await bindWorkspaceConversation("user:U1", { ttlMs: 1_000 });
+
+      vi.setSystemTime(new Date(1_002_000));
+      await expect(
+        withReadOnlyStateDatabase(() => resolveWorkspaceConversation("user:U1")),
+      ).rejects.toThrow();
+
+      vi.setSystemTime(new Date(1_000_500));
+      expect(resolveWorkspaceConversation("user:U1")).not.toBeNull();
+      testing.resetCurrentConversationBindingsForTests();
+      expect(resolveWorkspaceConversation("user:U1")).not.toBeNull();
+    });
+
+    it("keeps expired list entries when their cleanup write fails", async () => {
+      vi.useFakeTimers();
+      vi.setSystemTime(new Date(1_000_000));
+      const targetSessionKey = "agent:codex:acp:shared";
+      await bindWorkspaceConversation("user:U1", { targetSessionKey });
+      await bindWorkspaceConversation("user:U2", { targetSessionKey, ttlMs: 1_000 });
+
+      vi.setSystemTime(new Date(1_002_000));
+      await expect(
+        withReadOnlyStateDatabase(() =>
+          listGenericCurrentConversationBindingsBySession(targetSessionKey),
+        ),
+      ).rejects.toThrow();
+
+      vi.setSystemTime(new Date(1_000_500));
+      expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2);
+      testing.resetCurrentConversationBindingsForTests();
+      expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2);
+    });
+
+    it("does not partially prune an unbind-by-session batch", async () => {
+      vi.useFakeTimers();
+      vi.setSystemTime(new Date(1_000_000));
+      const targetSessionKey = "agent:codex:acp:shared";
+      await bindWorkspaceConversation("user:U1", { targetSessionKey });
+      await bindWorkspaceConversation("user:U2", { targetSessionKey, ttlMs: 1_000 });
+
+      vi.setSystemTime(new Date(1_002_000));
+      await expect(
+        withReadOnlyStateDatabase(() =>
+          unbindGenericCurrentConversationBindings({
+            targetSessionKey,
+            reason: "test cleanup",
+          }),
+        ),
+      ).rejects.toThrow();
+
+      vi.setSystemTime(new Date(1_000_500));
+      expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2);
+      testing.resetCurrentConversationBindingsForTests();
+      expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2);
+    });
+
+    it("retries the initial cache load after its SQLite cleanup fails", async () => {
+      await bindWorkspaceConversation("user:U1");
+      testing.resetCurrentConversationBindingsForTests();
+
+      await expect(
+        withReadOnlyStateDatabase(() => resolveWorkspaceConversation("user:U1")),
+      ).rejects.toThrow();
+
+      expect(resolveWorkspaceConversation("user:U1")).not.toBeNull();
+    });
+  });
 });

src/infra/outbound/current-conversation-bindings.ts

@@ -34,7 +34,7 @@ type CurrentConversationBindingDatabase = Pick<
 >;
 
 let bindingsLoaded = false;
-const bindingsByConversationKey = new Map<string, SessionBindingRecord>();
+let bindingsByConversationKey = new Map<string, SessionBindingRecord>();
 
 function buildConversationKey(ref: ConversationRef): string {
   const normalized = normalizeConversationRef(ref);
@@ -122,8 +122,8 @@ function targetAgentIdForSessionKey(targetSessionKey: string): string {
   return resolveAgentIdFromSessionKey(targetSessionKey);
 }
 
-function writePersistedBindings(): void {
-  const records = [...bindingsByConversationKey.values()]
+function writePersistedBindings(nextBindings: ReadonlyMap<string, SessionBindingRecord>): void {
+  const records = [...nextBindings.values()]
     .filter((record) => !isBindingExpired(record))
     .toSorted((a, b) => a.bindingId.localeCompare(b.bindingId));
   const updatedAt = Date.now();
@@ -163,29 +163,23 @@ function writePersistedBindings(): void {
   });
 }
 
+function commitBindings(nextBindings: Map<string, SessionBindingRecord>): void {
+  // SQLite is canonical: publish the prepared map only after its transaction
+  // commits, so a storage error cannot leave runtime routing ahead of disk.
+  writePersistedBindings(nextBindings);
+  bindingsByConversationKey = nextBindings;
+}
+
 function loadBindingsIntoMemory(): void {
   if (bindingsLoaded) {
     return;
   }
-  bindingsLoaded = true;
-  bindingsByConversationKey.clear();
+  const nextBindings = new Map<string, SessionBindingRecord>();
   for (const record of readPersistedBindings()) {
-    bindingsByConversationKey.set(buildConversationKey(record.conversation), record);
+    nextBindings.set(buildConversationKey(record.conversation), record);
   }
-}
-
-function pruneExpiredBinding(key: string): SessionBindingRecord | null {
-  loadBindingsIntoMemory();
-  const record = bindingsByConversationKey.get(key) ?? null;
-  if (!record) {
-    return null;
-  }
-  if (!isBindingExpired(record)) {
-    return record;
-  }
-  bindingsByConversationKey.delete(key);
-  writePersistedBindings();
-  return null;
+  bindingsByConversationKey = nextBindings;
+  bindingsLoaded = true;
 }
 
 function resolveChannelSupportsCurrentConversationBinding(params: {
@@ -295,7 +289,8 @@ export async function bindGenericCurrentConversation(
     return null;
   }
   const key = buildConversationKey(conversation);
-  const existing = pruneExpiredBinding(key);
+  const existing = bindingsByConversationKey.get(key);
+  const activeExisting = existing && !isBindingExpired(existing) ? existing : undefined;
   const record: SessionBindingRecord = {
     bindingId: buildBindingId(conversation),
     targetSessionKey,
@@ -305,13 +300,14 @@ export async function bindGenericCurrentConversation(
     boundAt: now,
     ...(expiresAt !== undefined ? { expiresAt } : {}),
     metadata: {
-      ...existing?.metadata,
+      ...activeExisting?.metadata,
       ...input.metadata,
       lastActivityAt: now,
     },
   };
-  bindingsByConversationKey.set(key, record);
-  writePersistedBindings();
+  const nextBindings = new Map(bindingsByConversationKey);
+  nextBindings.set(key, record);
+  commitBindings(nextBindings);
   return record;
 }
 
@@ -322,7 +318,16 @@ export function resolveGenericCurrentConversationBinding(
   if (!supportsGenericCurrentConversationBinding(ref)) {
     return null;
   }
-  return pruneExpiredBinding(buildConversationKey(ref));
+  loadBindingsIntoMemory();
+  const key = buildConversationKey(ref);
+  const record = bindingsByConversationKey.get(key) ?? null;
+  if (!record || !isBindingExpired(record)) {
+    return record;
+  }
+  const nextBindings = new Map(bindingsByConversationKey);
+  nextBindings.delete(key);
+  commitBindings(nextBindings);
+  return null;
 }
 
 /** Lists non-expired current-conversation bindings owned by one target session. */
@@ -331,10 +336,14 @@ export function listGenericCurrentConversationBindingsBySession(
 ): SessionBindingRecord[] {
   loadBindingsIntoMemory();
   const results: SessionBindingRecord[] = [];
-  for (const key of bindingsByConversationKey.keys()) {
-    const record = pruneExpiredBinding(key);
+  let nextBindings: Map<string, SessionBindingRecord> | undefined;
+  for (const [key, record] of bindingsByConversationKey) {
+    if (isBindingExpired(record)) {
+      nextBindings ??= new Map(bindingsByConversationKey);
+      nextBindings.delete(key);
+      continue;
+    }
     if (
-      !record ||
       record.targetSessionKey !== targetSessionKey ||
       !supportsGenericCurrentConversationBinding(record.conversation)
     ) {
@@ -342,6 +351,9 @@ export function listGenericCurrentConversationBindingsBySession(
     }
     results.push(record);
   }
+  if (nextBindings) {
+    commitBindings(nextBindings);
+  }
   return results;
 }
 
@@ -353,18 +365,23 @@ export function touchGenericCurrentConversationBinding(bindingId: string, at = D
   }
   loadBindingsIntoMemory();
   const key = bindingId.slice(CURRENT_BINDINGS_ID_PREFIX.length);
-  const record = pruneExpiredBinding(key);
+  const record = bindingsByConversationKey.get(key);
   if (!record) {
     return;
   }
-  bindingsByConversationKey.set(key, {
-    ...record,
-    metadata: {
-      ...record.metadata,
-      lastActivityAt: at,
-    },
-  });
-  writePersistedBindings();
+  const nextBindings = new Map(bindingsByConversationKey);
+  if (isBindingExpired(record)) {
+    nextBindings.delete(key);
+  } else {
+    nextBindings.set(key, {
+      ...record,
+      metadata: {
+        ...record.metadata,
+        lastActivityAt: at,
+      },
+    });
+  }
+  commitBindings(nextBindings);
 }
 
 /** Removes generic current-conversation bindings by binding id or target session key. */
@@ -381,11 +398,14 @@ export async function unbindGenericCurrentConversationBindings(
     }
     loadBindingsIntoMemory();
     const key = normalizedBindingId.slice(CURRENT_BINDINGS_ID_PREFIX.length);
-    const record = pruneExpiredBinding(key);
+    const record = bindingsByConversationKey.get(key);
     if (record) {
-      bindingsByConversationKey.delete(key);
-      removed.push(record);
-      writePersistedBindings();
+      const nextBindings = new Map(bindingsByConversationKey);
+      nextBindings.delete(key);
+      if (!isBindingExpired(record)) {
+        removed.push(record);
+      }
+      commitBindings(nextBindings);
     }
     return removed;
   }
@@ -393,20 +413,23 @@ export async function unbindGenericCurrentConversationBindings(
     return removed;
   }
   loadBindingsIntoMemory();
-  for (const key of bindingsByConversationKey.keys()) {
-    const record = pruneExpiredBinding(key);
+  const nextBindings = new Map(bindingsByConversationKey);
+  for (const [key, record] of bindingsByConversationKey) {
+    if (isBindingExpired(record)) {
+      nextBindings.delete(key);
+      continue;
+    }
     if (
-      !record ||
       record.targetSessionKey !== normalizedTargetSessionKey ||
       !supportsGenericCurrentConversationBinding(record.conversation)
     ) {
       continue;
     }
-    bindingsByConversationKey.delete(key);
+    nextBindings.delete(key);
     removed.push(record);
   }
-  if (removed.length > 0) {
-    writePersistedBindings();
+  if (nextBindings.size !== bindingsByConversationKey.size) {
+    commitBindings(nextBindings);
   }
   return removed;
 }
@@ -417,7 +440,7 @@ export const testing = {
     env?: NodeJS.ProcessEnv;
   }) {
     bindingsLoaded = false;
-    bindingsByConversationKey.clear();
+    bindingsByConversationKey = new Map();
     if (params?.deletePersistedFile) {
       runOpenClawStateWriteTransaction(
         ({ db }) => {