fix(whatsapp): bound reconnect catch-up replies (#80642)

* fix(whatsapp): bound reconnect catch-up replies

Co-authored-by: Vishal Jain <jainvishal2212@gmail.com>

* docs(changelog): defer reconnect entry to aggregate

---------

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

docs/channels/whatsapp.md

@@ -125,7 +125,7 @@ A separate WhatsApp number is recommended (setup and metadata are optimized for
 ## Runtime model
 
 - The gateway owns the WhatsApp socket and reconnect loop.
-- A watchdog tracks two signals independently: raw WhatsApp Web transport activity and application-message activity. A quiet-but-connected session is not restarted just because no message arrived recently; it forces reconnect only when transport frames stop arriving for a fixed internal window (not user-configurable) or application messages stay silent past 4x the normal message timeout. Right after a reconnect for a recently active session, that first window uses the shorter normal message timeout instead of the 4x window.
+- A watchdog tracks two signals independently: raw WhatsApp Web transport activity and application-message activity. A quiet-but-connected session is not restarted just because no message arrived recently; it forces reconnect only when transport frames stop arriving for a fixed internal window (not user-configurable) or application messages stay silent past 4x the normal message timeout. Right after a reconnect for a recently active session, that first window uses the shorter normal message timeout instead of the 4x window. OpenClaw can auto-reply to offline messages that Baileys delivers early in that reconnect, bounded by the inbound message-ID dedupe lifetime; initial startup keeps the short stale-history guard.
 - Baileys socket timings are explicit under `web.whatsapp.*`: `keepAliveIntervalMs` (application ping interval), `connectTimeoutMs` (opening handshake timeout), `defaultQueryTimeoutMs` (Baileys query waits, plus OpenClaw's outbound send/presence and inbound read-receipt timeouts).
 - Outbound sends require an active WhatsApp listener for the target account; sends fail fast otherwise.
 - Group sends attach native mention metadata for `@+<digits>` and `@<digits>` tokens (in text and media captions) when the token matches current participant metadata, including LID-backed groups.

extensions/whatsapp/src/auto-reply.test-harness.ts

@@ -244,6 +244,7 @@ export function createWebListenerFactoryCapture(): AnyExport {
         onMessage: (msg: WebInboundMessageInput) => Promise<void>;
         shouldDebounce?: (msg: WebInboundMessageInput) => boolean;
         debounceMs?: number;
+        appendReplyWindow?: { afterMs: number; untilMs: number; maxAgeMs: number };
         selfChatMode?: boolean;
       }
     | undefined;
@@ -251,6 +252,7 @@ export function createWebListenerFactoryCapture(): AnyExport {
     onMessage: (msg: WebInboundMessageInput) => Promise<void>;
     shouldDebounce?: (msg: WebInboundMessageInput) => boolean;
     debounceMs?: number;
+    appendReplyWindow?: { afterMs: number; untilMs: number; maxAgeMs: number };
     selfChatMode?: boolean;
   }) => {
     capturedOnMessage = opts.onMessage;

extensions/whatsapp/src/auto-reply.web-auto-reply.connection-and-logging.e2e.test.ts

@@ -213,6 +213,52 @@ describe("web auto-reply connection", () => {
     }
   });
 
+  it("widens append catch-up only after reconnecting a recently active listener", async () => {
+    const scripted = createScriptedWebListenerFactory();
+    const { controller, run } = startWebAutoReplyMonitor({
+      monitorWebChannelFn: monitorWebChannel as never,
+      listenerFactory: scripted.listenerFactory,
+      sleep: vi.fn(async () => {}),
+      messageTimeoutMs: 30 * 60_000,
+    });
+
+    await vi.waitFor(() => expect(scripted.getListenerCount()).toBe(1));
+    expect(
+      (
+        mockCallArg(scripted.listenerFactory, 0, 0) as {
+          appendReplyWindow?: { afterMs: number; untilMs: number; maxAgeMs: number };
+        }
+      ).appendReplyWindow,
+    ).toBeUndefined();
+
+    await sendWebDirectInboundMessage({
+      onMessage: requireOnMessage(scripted.getOnMessage()),
+      body: "hi before reconnect",
+      from: "+1",
+      to: "+2",
+      id: "active-before-reconnect",
+      spies: createWebInboundDeliverySpies(),
+    });
+
+    const reconnectStartedAt = Date.now();
+    scripted.resolveClose(0, { status: 408, isLoggedOut: false, error: new Error("timeout") });
+    await vi.waitFor(() => expect(scripted.getListenerCount()).toBe(2));
+
+    const appendReplyWindow = (
+      mockCallArg(scripted.listenerFactory, 1, 0) as {
+        appendReplyWindow?: { afterMs: number; untilMs: number; maxAgeMs: number };
+      }
+    ).appendReplyWindow;
+    expect(appendReplyWindow?.afterMs).toBeGreaterThanOrEqual(reconnectStartedAt - 20 * 60_000);
+    expect(appendReplyWindow?.afterMs).toBeLessThanOrEqual(Date.now() - 20 * 60_000);
+    expect(appendReplyWindow?.untilMs).toBeGreaterThanOrEqual(reconnectStartedAt + 20 * 60_000);
+    expect(appendReplyWindow?.maxAgeMs).toBe(20 * 60_000);
+
+    controller.abort();
+    scripted.resolveClose(1, { status: 499, isLoggedOut: false, error: "aborted" });
+    await run;
+  });
+
   it("retries opening-phase Boom 428 through the reconnect policy", async () => {
     const boom428 = {
       output: {

extensions/whatsapp/src/auto-reply/monitor.ts

@@ -28,6 +28,7 @@ import {
   type ManagedWhatsAppListener,
 } from "../connection-controller.js";
 import { resolveWhatsAppInboundPolicy } from "../inbound-policy.js";
+import { WHATSAPP_INBOUND_DEDUPE_TTL_MS } from "../inbound/dedupe.js";
 import { normalizeWebInboundMessage } from "../inbound/message-aliases.js";
 import {
   attachWebInboxToSocket,
@@ -221,6 +222,10 @@ export async function monitorWebChannel(
 
   const transportTimeoutMs = tuning.transportTimeoutMs ?? DEFAULT_TRANSPORT_TIMEOUT_MS;
   const messageTimeoutMs = tuning.messageTimeoutMs ?? 30 * 60 * 1000;
+  const reconnectCatchUpWindowMs = Math.min(
+    Math.max(messageTimeoutMs, 60_000),
+    WHATSAPP_INBOUND_DEDUPE_TTL_MS,
+  );
   const watchdogCheckMs = tuning.watchdogCheckMs ?? 60 * 1000;
   const controller = new WhatsAppConnectionController({
     accountId: account.accountId,
@@ -302,7 +307,6 @@ export async function monitorWebChannel(
               baseMentionConfig,
               account,
             });
-
             return (await (listenerFactory ?? attachWebInboxToSocket)({
               cfg,
               loadConfig: loadCurrentMonitorConfig,
@@ -314,6 +318,13 @@ export async function monitorWebChannel(
               sendReadReceipts: account.sendReadReceipts,
               socketTiming,
               debounceMs: inboundDebounceMs,
+              appendReplyWindow: connectionLocal.openedAfterRecentInbound
+                ? {
+                    afterMs: connectionLocal.startedAt - reconnectCatchUpWindowMs,
+                    untilMs: connectionLocal.startedAt + reconnectCatchUpWindowMs,
+                    maxAgeMs: reconnectCatchUpWindowMs,
+                  }
+                : undefined,
               shouldDebounce,
               socketRef: controller.socketRef,
               shouldRetryDisconnect: () => !sigintStop && controller.shouldRetryDisconnect(),

extensions/whatsapp/src/inbound/dedupe.ts

@@ -1,13 +1,13 @@
 // Whatsapp plugin module implements dedupe behavior.
 import { createClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe";
 
-const RECENT_WEB_MESSAGE_TTL_MS = 20 * 60_000;
+export const WHATSAPP_INBOUND_DEDUPE_TTL_MS = 20 * 60_000;
 const RECENT_WEB_MESSAGE_MAX = 5000;
 const RECENT_OUTBOUND_MESSAGE_TTL_MS = 20 * 60_000;
 const RECENT_OUTBOUND_MESSAGE_MAX = 5000;
 
 const claimableInboundMessages = createClaimableDedupe({
-  ttlMs: RECENT_WEB_MESSAGE_TTL_MS,
+  ttlMs: WHATSAPP_INBOUND_DEDUPE_TTL_MS,
   memoryMaxSize: RECENT_WEB_MESSAGE_MAX,
 });
 const recentOutboundMessages = createRecentMessageCache({

extensions/whatsapp/src/inbound/monitor.ts

@@ -265,6 +265,12 @@ type AdmittedWebInboundCallbackMessage = WebInboundMessage & {
   admission: AdmittedWebInboundMessage["admission"];
 };
 
+type AppendReplyWindow = {
+  afterMs: number;
+  untilMs: number;
+  maxAgeMs: number;
+};
+
 type MonitorWebInboxOptions = {
   cfg: OpenClawConfig;
   loadConfig?: () => OpenClawConfig;
@@ -280,6 +286,8 @@ type MonitorWebInboxOptions = {
   sendReadReceipts?: boolean;
   /** Debounce window (ms) for batching rapid consecutive messages from the same sender (0 to disable). */
   debounceMs?: number;
+  /** Bounded reconnect window for offline append auto-replies. */
+  appendReplyWindow?: AppendReplyWindow;
   /** Optional debounce gating predicate. */
   shouldDebounce?: (msg: AdmittedWebInboundCallbackMessage) => boolean;
   /** Optional shared socket reference so reply closures can follow reconnects. */
@@ -992,7 +1000,14 @@ export async function attachWebInboxToSocket(
     const APPEND_RECENT_GRACE_MS = 60_000;
     const msgTsSeconds = parseWhatsAppTimestampSeconds(msg.messageTimestamp);
     const msgTsMs = msgTsSeconds !== undefined ? msgTsSeconds * 1000 : 0;
-    return msgTsMs < connectedAtMs - APPEND_RECENT_GRACE_MS;
+    // Reconnect catch-up is temporary; after it expires, preserve steady-state
+    // handling for fresh appends instead of rejecting every later append.
+    const nowMs = Date.now();
+    const appendAfterMs =
+      options.appendReplyWindow && nowMs <= options.appendReplyWindow.untilMs
+        ? Math.max(options.appendReplyWindow.afterMs, nowMs - options.appendReplyWindow.maxAgeMs)
+        : connectedAtMs - APPEND_RECENT_GRACE_MS;
+    return msgTsMs < appendAfterMs;
   };
 
   const processDurableInboundMessage = async (

extensions/whatsapp/src/monitor-inbox.append-upsert.test-support.ts

@@ -59,6 +59,119 @@ describe("append upsert handling (#20952)", () => {
     await listener.close();
   });
 
+  it("processes only reconnect catch-up appends within the dedupe age", async () => {
+    const onMessage = vi.fn(async () => {});
+    const { listener, sock } = await startInboxMonitor(onMessage, {
+      appendReplyWindow: {
+        afterMs: Date.now() - 30 * 60_000,
+        untilMs: Date.now() + 30 * 60_000,
+        maxAgeMs: 20 * 60_000,
+      },
+    });
+
+    sock.ev.emit("messages.upsert", {
+      type: "append",
+      messages: [
+        {
+          key: { id: "catch-up-1", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "missed while reconnecting" },
+          messageTimestamp: Math.floor(Date.now() / 1000) - 15 * 60,
+          pushName: "Reconnect Tester",
+        },
+      ],
+    });
+    await waitForMessageCalls(onMessage, 1);
+
+    sock.ev.emit("messages.upsert", {
+      type: "append",
+      messages: [
+        {
+          key: { id: "catch-up-old", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "before the recovery window" },
+          messageTimestamp: Math.floor(Date.now() / 1000) - 25 * 60,
+          pushName: "Reconnect Tester",
+        },
+      ],
+    });
+    await settleInboundWork();
+
+    expect(onMessage).toHaveBeenCalledTimes(1);
+
+    await listener.close();
+  });
+
+  it("ends reconnect catch-up while preserving fresh appends after the window expires", async () => {
+    const onMessage = vi.fn(async () => {});
+    const { listener, sock } = await startInboxMonitor(onMessage, {
+      appendReplyWindow: {
+        afterMs: Date.now() - 30 * 60_000,
+        untilMs: Date.now() - 1,
+        maxAgeMs: 20 * 60_000,
+      },
+    });
+
+    sock.ev.emit("messages.upsert", {
+      type: "append",
+      messages: [
+        {
+          key: { id: "catch-up-late", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "arrived after recovery" },
+          messageTimestamp: Math.floor(Date.now() / 1000) - 5 * 60,
+          pushName: "Reconnect Tester",
+        },
+      ],
+    });
+    await settleInboundWork();
+
+    expect(onMessage).not.toHaveBeenCalled();
+
+    sock.ev.emit("messages.upsert", {
+      type: "append",
+      messages: [
+        {
+          key: { id: "fresh-after-catch-up", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "fresh after recovery" },
+          messageTimestamp: Math.floor(Date.now() / 1000),
+          pushName: "Reconnect Tester",
+        },
+      ],
+    });
+    await waitForMessageCalls(onMessage, 1);
+
+    expect(onMessage).toHaveBeenCalledTimes(1);
+
+    await listener.close();
+  });
+
+  it("processes a distinct catch-up message from the boundary second", async () => {
+    const onMessage = vi.fn(async () => {});
+    const boundarySeconds = Math.floor(Date.now() / 1000) - 20 * 60 + 1;
+    const { listener, sock } = await startInboxMonitor(onMessage, {
+      appendReplyWindow: {
+        afterMs: boundarySeconds * 1000,
+        untilMs: Date.now() + 30 * 60_000,
+        maxAgeMs: 20 * 60_000,
+      },
+    });
+
+    sock.ev.emit("messages.upsert", {
+      type: "append",
+      messages: [
+        {
+          key: { id: "catch-up-same-second", fromMe: false, remoteJid: "999@s.whatsapp.net" },
+          message: { conversation: "same second, different message" },
+          messageTimestamp: boundarySeconds,
+          pushName: "Reconnect Tester",
+        },
+      ],
+    });
+    await waitForMessageCalls(onMessage, 1);
+
+    expect(onMessage).toHaveBeenCalledTimes(1);
+
+    await listener.close();
+  });
+
   it("skips append messages with NaN/non-finite timestamps", async () => {
     const onMessage = vi.fn(async () => {});
     const { listener, sock } = await startInboxMonitor(onMessage);