retry whatsapp session init conflicts (#101106)

andersonjeccel · web-flow · commit 87a9e27025b7 · 2026-07-06T23:03:10.000-03:00
diff --git a/extensions/whatsapp/src/inbound/monitor.ts b/extensions/whatsapp/src/inbound/monitor.ts
@@ -14,6 +14,7 @@ import {
   formatLocationText,
 } from "openclaw/plugin-sdk/channel-inbound";
 import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce";
+import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
 import { getChildLogger } from "openclaw/plugin-sdk/logging-core";
 import {
   asDateTimestampMs,
@@ -104,12 +105,31 @@ const RECONNECT_IN_PROGRESS_ERROR = "no active socket - reconnection in progress
 const GROUP_META_TTL_MS = 5 * 60 * 1000;
 const BAILEYS_MESSAGE_TTL_MS = 10 * 60 * 1000;
 const INBOUND_CLOSE_DRAIN_TIMEOUT_MS = 5_000;
+const REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE = /reply session initialization conflicted for \S+/u;
 export const WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES = 500;
 
 type WhatsAppGroupMetadataCacheEntry = {
   subject?: string;
   expires: number;
 };
+
+function resolveRetryableWhatsAppInboundError(
+  error: unknown,
+): WhatsAppRetryableInboundError | null {
+  if (error instanceof WhatsAppRetryableInboundError) {
+    return error;
+  }
+  const hasSessionInitConflict = collectErrorGraphCandidates(error, (current) => [
+    current.cause,
+    current.error,
+  ]).some((candidate) =>
+    REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE.test(formatErrorMessage(candidate)),
+  );
+  if (!hasSessionInitConflict) {
+    return null;
+  }
+  return new WhatsAppRetryableInboundError(formatErrorMessage(error), { cause: error });
+}
 export type WhatsAppGroupMetadataCache = Map<string, WhatsAppGroupMetadataCacheEntry>;
 export type WhatsAppBaileysCacheEntry<T> = {
   expiresAt: number;
@@ -465,12 +485,13 @@ export async function attachWebInboxToSocket(
       (entry): entry is QueuedInboundMessage & { readReceipt: WhatsAppReadReceiptTarget } =>
         Boolean(entry.readReceipt),
     );
-    if (error instanceof WhatsAppRetryableInboundError) {
-      dedupeKeys.forEach((dedupeKey) => releaseRecentInboundMessage(dedupeKey, error));
+    const retryableError = resolveRetryableWhatsAppInboundError(error);
+    if (retryableError) {
+      dedupeKeys.forEach((dedupeKey) => releaseRecentInboundMessage(dedupeKey, retryableError));
       await Promise.all(
         durableEntries.map((entry) =>
           durableInboundJournal.release(entry.durableId, {
-            lastError: formatError(error),
+            lastError: formatError(retryableError),
           }),
         ),
       );
diff --git a/extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts b/extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts
@@ -1596,6 +1596,39 @@ describe("web monitor inbox", () => {
     await listener.close();
   });
 
+  it("retries redelivered messages after reply session initialization conflicts", async () => {
+    let attempts = 0;
+    const onMessage = vi.fn(async () => {
+      attempts += 1;
+      if (attempts === 1) {
+        throw new Error(
+          "reply session initialization conflicted for agent:main:whatsapp:direct:+15551234567",
+        );
+      }
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage);
+    const upsert = buildNotifyMessageUpsert({
+      id: nextMessageId("session-init-conflict"),
+      remoteJid: "999@s.whatsapp.net",
+      text: "ping",
+      timestamp: 1_700_000_000,
+      pushName: "Tester",
+    });
+
+    sock.ev.emit("messages.upsert", upsert);
+    await waitForMessageCalls(onMessage, 1);
+    expect(sock.readMessages).not.toHaveBeenCalled();
+
+    sock.ev.emit("messages.upsert", upsert);
+    await waitForMessageCalls(onMessage, 2);
+    await vi.waitFor(() => {
+      expect(sock.readMessages).toHaveBeenCalledTimes(1);
+    });
+
+    await listener.close();
+  });
+
   it("resolves LID JIDs using Baileys LID mapping store", async () => {
     const onMessage = vi.fn(async () => {});
 
