fix(whatsapp): wire missing Baileys retry/cache hooks for group message reliability (#94338)

Merged via squash.

Prepared head SHA: ee6de071f7
Co-authored-by: xialonglee <22994703+xialonglee@users.noreply.github.com>
Co-authored-by: mcaxtr <7562095+mcaxtr@users.noreply.github.com>
Reviewed-by: @mcaxtr

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

@@ -1,4 +1,5 @@
 // Whatsapp plugin module implements monitor behavior.
+import type { WAMessageKey } from "baileys";
 import { resolveAccountEntry } from "openclaw/plugin-sdk/account-core";
 import { CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY } from "openclaw/plugin-sdk/approval-handler-runtime";
 import { resolveInboundDebounceMs } from "openclaw/plugin-sdk/channel-inbound-debounce";
@@ -27,7 +28,13 @@ import {
 } from "../connection-controller.js";
 import { resolveWhatsAppInboundPolicy } from "../inbound-policy.js";
 import { normalizeWebInboundMessage } from "../inbound/message-aliases.js";
-import { attachWebInboxToSocket, type WhatsAppGroupMetadataCache } from "../inbound/monitor.js";
+import {
+  attachWebInboxToSocket,
+  readWhatsAppBaileysCacheEntry,
+  type WhatsAppBaileysGroupMetadataCache,
+  type WhatsAppBaileysMessageCache,
+  type WhatsAppGroupMetadataCache,
+} from "../inbound/monitor.js";
 import type { WebInboundMessageInput } from "../inbound/types.js";
 import {
   newConnectionId,
@@ -193,6 +200,8 @@ export async function monitorWebChannel(
   >();
   const groupMemberNames = new Map<string, Map<string, string>>();
   const groupMetadataCache: WhatsAppGroupMetadataCache = new Map();
+  const recentMessageKeys: WhatsAppBaileysMessageCache = new Map();
+  const baileysGroupMetaCache: WhatsAppBaileysGroupMetadataCache = new Map();
   const echoTracker = createEchoTracker({ maxItems: 100, logVerbose });
 
   const sleep =
@@ -268,6 +277,14 @@ export async function monitorWebChannel(
       try {
         connection = await controller.openConnection({
           connectionId,
+          getMessage: async (key: WAMessageKey) =>
+            key.id && key.remoteJid
+              ? readWhatsAppBaileysCacheEntry(recentMessageKeys, `${key.remoteJid}:${key.id}`)
+              : undefined,
+          cachedGroupMetadata: async (jid: string) => {
+            const meta = readWhatsAppBaileysCacheEntry(baileysGroupMetaCache, jid);
+            return meta?.participants?.length ? meta : undefined;
+          },
           createListener: async ({ sock, connection: connectionLocal }) => {
             const onMessage = createWebOnMessageHandler({
               cfg,
@@ -303,6 +320,8 @@ export async function monitorWebChannel(
               disconnectRetryPolicy: reconnectPolicy,
               disconnectRetryAbortSignal: controller.getDisconnectRetryAbortSignal(),
               groupMetadataCache,
+              recentMessageKeys,
+              baileysGroupMetaCache,
               onMessage: async (msg: WebInboundMessageInput) => {
                 const normalized = normalizeWebInboundMessage(msg);
                 const inboundAt = Date.now();

extensions/whatsapp/src/connection-controller.ts

@@ -1,5 +1,5 @@
 // Whatsapp plugin module implements connection controller behavior.
-import type { WASocket } from "baileys";
+import type { GroupMetadata, WASocket, WAMessageKey, proto } from "baileys";
 import { info } from "openclaw/plugin-sdk/runtime-env";
 import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
 import {
@@ -528,6 +528,8 @@ export class WhatsAppConnectionController {
     }) => Promise<ManagedWhatsAppListener>;
     onHeartbeat?: (snapshot: WhatsAppConnectionSnapshot) => void;
     onWatchdogTimeout?: (snapshot: WhatsAppConnectionSnapshot) => void;
+    getMessage?: (key: WAMessageKey) => Promise<proto.IMessage | undefined>;
+    cachedGroupMetadata?: (jid: string) => Promise<GroupMetadata | undefined>;
   }): Promise<WhatsAppLiveConnection> {
     if (this.current) {
       await this.closeCurrentConnection();
@@ -539,6 +541,8 @@ export class WhatsAppConnectionController {
       sock = await createWaSocket(false, this.verbose, {
         authDir: this.authDir,
         ...this.socketTiming,
+        ...(params.getMessage ? { getMessage: params.getMessage } : {}),
+        ...(params.cachedGroupMetadata ? { cachedGroupMetadata: params.cachedGroupMetadata } : {}),
       });
       await waitForWaConnection(sock, { timeoutMs: this.socketTiming.connectTimeoutMs });
 

extensions/whatsapp/src/inbound/monitor.ts

@@ -5,6 +5,7 @@ import type {
   proto,
   GroupMetadata,
   WAMessage,
+  WAMessageKey,
   WASocket,
 } from "baileys";
 import { recordChannelActivity } from "openclaw/plugin-sdk/channel-activity-runtime";
@@ -97,7 +98,8 @@ import type {
 
 const LOGGED_OUT_STATUS = DisconnectReason?.loggedOut ?? 401;
 const RECONNECT_IN_PROGRESS_ERROR = "no active socket - reconnection in progress";
-const GROUP_META_TTL_MS = 5 * 60 * 1000; // 5 minutes
+const GROUP_META_TTL_MS = 5 * 60 * 1000;
+const BAILEYS_MESSAGE_TTL_MS = 10 * 60 * 1000;
 const INBOUND_CLOSE_DRAIN_TIMEOUT_MS = 5_000;
 export const WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES = 500;
 
@@ -106,6 +108,15 @@ type WhatsAppGroupMetadataCacheEntry = {
   expires: number;
 };
 export type WhatsAppGroupMetadataCache = Map<string, WhatsAppGroupMetadataCacheEntry>;
+export type WhatsAppBaileysCacheEntry<T> = {
+  expiresAt: number;
+  value: T;
+};
+export type WhatsAppBaileysMessageCache = Map<string, WhatsAppBaileysCacheEntry<proto.IMessage>>;
+export type WhatsAppBaileysGroupMetadataCache = Map<
+  string,
+  WhatsAppBaileysCacheEntry<GroupMetadata>
+>;
 type LocalGroupMetadataCacheEntry = WhatsAppGroupMetadataCacheEntry & {
   participants?: string[];
   mentionParticipants?: WhatsAppOutboundMentionParticipant[];
@@ -171,6 +182,48 @@ function readGroupMetadataCacheEntry<T extends WhatsAppGroupMetadataCacheEntry>(
   return entry;
 }
 
+function rememberWhatsAppBaileysCacheEntry<T>(
+  cache: Map<string, WhatsAppBaileysCacheEntry<T>> | undefined,
+  key: string,
+  value: T,
+  ttlMs: number,
+): void {
+  if (!cache) {
+    return;
+  }
+  if (cache.has(key)) {
+    cache.delete(key);
+  }
+  cache.set(key, {
+    expiresAt: Date.now() + ttlMs,
+    value,
+  });
+  while (cache.size > WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES) {
+    const oldest = cache.keys().next();
+    if (oldest.done) {
+      break;
+    }
+    cache.delete(oldest.value);
+  }
+}
+
+export function readWhatsAppBaileysCacheEntry<T>(
+  cache: Map<string, WhatsAppBaileysCacheEntry<T>>,
+  key: string,
+): T | undefined {
+  const entry = cache.get(key);
+  if (!entry) {
+    return undefined;
+  }
+  if (entry.expiresAt <= Date.now()) {
+    cache.delete(key);
+    return undefined;
+  }
+  cache.delete(key);
+  cache.set(key, entry);
+  return entry.value;
+}
+
 function logWhatsAppVerbose(enabled: boolean | undefined, message: string) {
   if (!enabled) {
     return;
@@ -242,6 +295,8 @@ type MonitorWebInboxOptions = {
   disconnectRetryAbortSignal?: AbortSignal;
   /** Shared group metadata cache used only for inbound metadata fallback after fetch failures. */
   groupMetadataCache?: WhatsAppGroupMetadataCache;
+  recentMessageKeys?: WhatsAppBaileysMessageCache;
+  baileysGroupMetaCache?: WhatsAppBaileysGroupMetadataCache;
 };
 
 type AttachWebInboxToSocketOptions = Omit<
@@ -495,10 +550,29 @@ export async function attachWebInboxToSocket(
   const groupMetadataCache = options.groupMetadataCache ?? new Map();
   const groupMetaCache = new Map<string, LocalGroupMetadataCacheEntry>();
   const lidLookup = sock.signalRepository?.lidMapping;
+  const publishedGroupMetadataJids = new Set<string>();
+  const invalidatedGroupMetadataJids = new Set<string>();
+  let groupMetadataCacheClosed = false;
 
   const resolveInboundJid = async (jid: string | null | undefined): Promise<string | null> =>
     resolveJidToE164(jid, { authDir: options.authDir, lidLookup });
 
+  const rememberBaileysMessage = (
+    remoteJid: string | null | undefined,
+    messageId: string | null | undefined,
+    message: proto.IMessage | null | undefined,
+  ) => {
+    if (!options.recentMessageKeys || !remoteJid || !messageId || !message) {
+      return;
+    }
+    rememberWhatsAppBaileysCacheEntry(
+      options.recentMessageKeys,
+      `${remoteJid}:${messageId}`,
+      message,
+      BAILEYS_MESSAGE_TTL_MS,
+    );
+  };
+
   const rememberOutboundMessage = (remoteJid: string, result: unknown) => {
     const messageId =
       typeof result === "object" && result && "key" in result
@@ -512,6 +586,11 @@ export async function attachWebInboxToSocket(
       remoteJid,
       messageId,
     });
+    const message =
+      typeof result === "object" && result && "message" in result
+        ? (result as { message?: proto.IMessage }).message
+        : undefined;
+    rememberBaileysMessage(remoteJid, messageId, message);
   };
   const trackLateAcceptedSend = (jid: string, promise: Promise<WAMessage | undefined>) => {
     // The local send has failed terminally, but Baileys may still deliver it.
@@ -629,6 +708,13 @@ export async function attachWebInboxToSocket(
     }
     try {
       const meta = await (getCurrentSock() ?? sock).groupMetadata(jid);
+      rememberWhatsAppBaileysCacheEntry(
+        options.baileysGroupMetaCache,
+        jid,
+        meta,
+        GROUP_META_TTL_MS,
+      );
+      publishedGroupMetadataJids.add(jid);
       const entry = await summarizeGroupMeta(meta);
       rememberGroupMetadataCacheEntry(groupMetadataCache, jid, {
         subject: entry.subject,
@@ -1260,6 +1346,8 @@ export async function attachWebInboxToSocket(
       return;
     }
     for (const msg of upsert.messages ?? []) {
+      rememberBaileysMessage(msg.key?.remoteJid, msg.key?.id, msg.message);
+
       const receiveOrder = nextReceiveOrder++;
       if (
         await maybeResolveWhatsAppApprovalReaction({
@@ -1309,6 +1397,7 @@ export async function attachWebInboxToSocket(
     }
   };
   const drainInboundBeforeSocketClose = async () => {
+    groupMetadataCacheClosed = true;
     await waitForPendingMessageHandlers();
     await drainDebouncedInboundMessages();
   };
@@ -1352,25 +1441,82 @@ export async function attachWebInboxToSocket(
       resolveClose({ status: undefined, isLoggedOut: false, error: err });
     }
   };
-  const detachMessagesUpsert = attachEmitterListener(
-    sock.ev as unknown as {
-      on: (event: string, listener: (...args: unknown[]) => void) => void;
-      off?: (event: string, listener: (...args: unknown[]) => void) => void;
-      removeListener?: (event: string, listener: (...args: unknown[]) => void) => void;
-    },
+  const attachSockListener = (event: string, listener: (...args: unknown[]) => void) =>
+    attachEmitterListener(
+      sock.ev as unknown as {
+        on: (event: string, listener: (...args: unknown[]) => void) => void;
+        off?: (event: string, listener: (...args: unknown[]) => void) => void;
+        removeListener?: (event: string, listener: (...args: unknown[]) => void) => void;
+      },
+      event,
+      listener,
+    );
+  const detachMessagesUpsert = attachSockListener(
     "messages.upsert",
     handleMessagesUpsertEvent as unknown as (...args: unknown[]) => void,
   );
-  const detachConnectionUpdate = attachEmitterListener(
-    sock.ev as unknown as {
-      on: (event: string, listener: (...args: unknown[]) => void) => void;
-      off?: (event: string, listener: (...args: unknown[]) => void) => void;
-      removeListener?: (event: string, listener: (...args: unknown[]) => void) => void;
-    },
+  const detachConnectionUpdate = attachSockListener(
     "connection.update",
     handleConnectionUpdate as unknown as (...args: unknown[]) => void,
   );
 
+  const isFullGroupMetadataUpdate = (update: Partial<GroupMetadata>): update is GroupMetadata =>
+    typeof update.id === "string" &&
+    typeof update.subject === "string" &&
+    Array.isArray(update.participants);
+
+  const rememberFullGroupMetadataUpdate = (jid: string, meta: GroupMetadata) => {
+    if (groupMetadataCacheClosed) {
+      return;
+    }
+    rememberWhatsAppBaileysCacheEntry(options.baileysGroupMetaCache, jid, meta, GROUP_META_TTL_MS);
+    publishedGroupMetadataJids.add(jid);
+    invalidatedGroupMetadataJids.delete(jid);
+    rememberGroupMetadataCacheEntry(
+      groupMetadataCache,
+      jid,
+      summarizeGroupMetaForReconnectCache(meta),
+    );
+    groupMetaCache.delete(jid);
+  };
+
+  const forgetFullGroupMetadata = (jid: string) => {
+    options.baileysGroupMetaCache?.delete(jid);
+    groupMetadataCache.delete(jid);
+    groupMetaCache.delete(jid);
+    publishedGroupMetadataJids.delete(jid);
+    invalidatedGroupMetadataJids.add(jid);
+  };
+
+  const detachGroupsUpsert = attachSockListener("groups.upsert", ((groups: GroupMetadata[]) => {
+    for (const group of groups) {
+      if (group.id) {
+        rememberFullGroupMetadataUpdate(group.id, group);
+      }
+    }
+  }) as unknown as (...args: unknown[]) => void);
+
+  const detachGroupsUpdate = attachSockListener("groups.update", ((
+    updates: Partial<GroupMetadata>[],
+  ) => {
+    for (const update of updates) {
+      if (!update.id) {
+        continue;
+      }
+      if (isFullGroupMetadataUpdate(update)) {
+        rememberFullGroupMetadataUpdate(update.id, update);
+        continue;
+      }
+      forgetFullGroupMetadata(update.id);
+    }
+  }) as unknown as (...args: unknown[]) => void);
+
+  const detachGroupParticipantsUpdate = attachSockListener("group-participants.update", ((update: {
+    id: string;
+  }) => {
+    forgetFullGroupMetadata(update.id);
+  }) as unknown as (...args: unknown[]) => void);
+
   const replayTask = replayPendingDurableInboundMessages().catch((err: unknown) => {
     inboundLogger.error({ error: String(err) }, "failed replaying durable WhatsApp inbound");
     inboundConsoleLog.error(`Failed replaying durable WhatsApp inbound: ${String(err)}`);
@@ -1380,16 +1526,30 @@ export async function attachWebInboxToSocket(
     pendingMessageHandlers.delete(replayTask);
   });
 
-  void (async () => {
+  const groupHydrationTask = (async () => {
     try {
       const groups = await sock.groupFetchAllParticipating();
+      if (groupMetadataCacheClosed) {
+        return;
+      }
       for (const [jid, meta] of Object.entries(groups ?? {})) {
-        if (meta) {
+        if (
+          meta &&
+          !publishedGroupMetadataJids.has(jid) &&
+          !invalidatedGroupMetadataJids.has(jid)
+        ) {
           rememberGroupMetadataCacheEntry(
             groupMetadataCache,
             jid,
             summarizeGroupMetaForReconnectCache(meta),
           );
+          rememberWhatsAppBaileysCacheEntry(
+            options.baileysGroupMetaCache,
+            jid,
+            meta,
+            GROUP_META_TTL_MS,
+          );
+          publishedGroupMetadataJids.add(jid);
         }
       }
       logWhatsAppVerbose(
@@ -1406,6 +1566,7 @@ export async function attachWebInboxToSocket(
       );
     }
   })();
+  void groupHydrationTask;
 
   const sendApi = createWebSendApi({
     sock: sendApiSocketOperations,
@@ -1419,6 +1580,9 @@ export async function attachWebInboxToSocket(
       try {
         detachMessagesUpsert();
         detachConnectionUpdate();
+        detachGroupsUpsert();
+        detachGroupsUpdate();
+        detachGroupParticipantsUpdate();
         await drainInboundBeforeSocketCloseWithTimeout();
       } catch (err) {
         logWhatsAppVerbose(options.verbose, `Inbound close drain failed: ${String(err)}`);
@@ -1442,9 +1606,21 @@ export async function attachWebInboxToSocket(
 
 export async function monitorWebInbox(options: MonitorWebInboxOptions) {
   const socketTiming = options.socketTiming ?? resolveWhatsAppSocketTiming(options.cfg);
+  const recentMessageKeys: WhatsAppBaileysMessageCache = options.recentMessageKeys ?? new Map();
+  const baileysGroupMetaCache: WhatsAppBaileysGroupMetadataCache =
+    options.baileysGroupMetaCache ?? new Map();
+
   const sock = await createWaSocket(false, options.verbose, {
     authDir: options.authDir,
     ...socketTiming,
+    getMessage: async (key: WAMessageKey) =>
+      key.id && key.remoteJid
+        ? readWhatsAppBaileysCacheEntry(recentMessageKeys, `${key.remoteJid}:${key.id}`)
+        : undefined,
+    cachedGroupMetadata: async (jid: string) => {
+      const meta = readWhatsAppBaileysCacheEntry(baileysGroupMetaCache, jid);
+      return meta?.participants?.length ? meta : undefined;
+    },
   });
   try {
     await waitForWaConnection(sock, { timeoutMs: socketTiming.connectTimeoutMs });
@@ -1469,5 +1645,7 @@ export async function monitorWebInbox(options: MonitorWebInboxOptions) {
       : undefined,
     socketTiming,
     sock,
+    recentMessageKeys,
+    baileysGroupMetaCache,
   });
 }

extensions/whatsapp/src/monitor-inbox.streams-inbound-messages.test-support.ts

@@ -1,6 +1,7 @@
 // Whatsapp plugin module implements monitor inbox.streams inbound messages support behavior.
 import fsSync from "node:fs";
 import path from "node:path";
+import type { GroupMetadata, WAMessageKey } from "baileys";
 import "./monitor-inbox.test-harness.js";
 import { defaultRuntime } from "openclaw/plugin-sdk/runtime-env";
 import { beforeEach, describe, expect, it, vi } from "vitest";
@@ -9,7 +10,12 @@ import {
   unregisterWhatsAppConnectionController,
 } from "./connection-controller-registry.js";
 import { WhatsAppRetryableInboundError } from "./inbound/dedupe.js";
-import { WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES } from "./inbound/monitor.js";
+import {
+  readWhatsAppBaileysCacheEntry,
+  type WhatsAppBaileysGroupMetadataCache,
+  type WhatsAppBaileysMessageCache,
+  WHATSAPP_GROUP_METADATA_CACHE_MAX_ENTRIES,
+} from "./inbound/monitor.js";
 import type { WebInboundMessage } from "./inbound/types.js";
 import {
   type InboxMonitorOptions,
@@ -106,17 +112,70 @@ async function expectSocketOperationTimeout(
   await rejection;
 }
 
+function groupMetadata(params: {
+  id?: string;
+  subject: string;
+  participants?: string[];
+}): GroupMetadata {
+  return {
+    id: params.id ?? "123@g.us",
+    subject: params.subject,
+    owner: undefined,
+    participants: (params.participants ?? ["555@s.whatsapp.net"]).map((id) => ({ id })),
+  };
+}
+
+function createBaileysCacheSupport() {
+  const recentMessageKeys: WhatsAppBaileysMessageCache = new Map();
+  const baileysGroupMetaCache: WhatsAppBaileysGroupMetadataCache = new Map();
+  const socketOptions = {
+    getMessage: async (key: WAMessageKey) =>
+      key.id && key.remoteJid
+        ? readWhatsAppBaileysCacheEntry(recentMessageKeys, `${key.remoteJid}:${key.id}`)
+        : undefined,
+    cachedGroupMetadata: async (jid: string) => {
+      const meta = readWhatsAppBaileysCacheEntry(baileysGroupMetaCache, jid);
+      return meta?.participants?.length ? meta : undefined;
+    },
+  };
+  return { recentMessageKeys, baileysGroupMetaCache, socketOptions };
+}
+
+async function startInboxMonitorWithBaileysCache(
+  options: Partial<Pick<InboxMonitorOptions, "groupMetadataCache">> = {},
+) {
+  const baileysCache = createBaileysCacheSupport();
+  const started = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, {
+    ...options,
+    recentMessageKeys: baileysCache.recentMessageKeys,
+    baileysGroupMetaCache: baileysCache.baileysGroupMetaCache,
+  });
+  return { ...started, baileysCache };
+}
+
+async function expectCachedGroupMetadata(
+  baileysCache: ReturnType<typeof createBaileysCacheSupport>,
+  expected: Pick<GroupMetadata, "id" | "subject" | "participants">,
+) {
+  await expect(baileysCache.socketOptions.cachedGroupMetadata(expected.id)).resolves.toMatchObject(
+    expected,
+  );
+}
+
 async function primeInboundReplyHandle(params: {
   onMessage: ReturnType<typeof vi.fn>;
   socketRef: NonNullable<InboxMonitorOptions["socketRef"]>;
   upsertId: string;
   retryPolicy: NonNullable<InboxMonitorOptions["disconnectRetryPolicy"]>;
+  baileysCache?: ReturnType<typeof createBaileysCacheSupport>;
   useCurrentSock?: boolean;
 }) {
   const { listener, sock } = await startInboxMonitor(params.onMessage as InboxOnMessage, {
     socketRef: params.socketRef,
     shouldRetryDisconnect: () => true,
     disconnectRetryPolicy: params.retryPolicy,
+    recentMessageKeys: params.baileysCache?.recentMessageKeys,
+    baileysGroupMetaCache: params.baileysCache?.baileysGroupMetaCache,
   });
   const sourceSock = params.useCurrentSock ? getSock() : sock;
   sourceSock.ev.emit(
@@ -480,6 +539,174 @@ describe("web monitor inbox", () => {
     await second.listener.close();
   });
 
+  it("keeps full participating group metadata available to Baileys", async () => {
+    const sock = getSock();
+    sock.groupFetchAllParticipating.mockResolvedValueOnce({
+      "123@g.us": groupMetadata({
+        subject: "Recovered Group",
+        participants: ["444@s.whatsapp.net"],
+      }),
+    });
+
+    const { listener, baileysCache } = await startInboxMonitorWithBaileysCache();
+
+    await vi.waitFor(async () => {
+      await expectCachedGroupMetadata(baileysCache, {
+        id: "123@g.us",
+        subject: "Recovered Group",
+        participants: [{ id: "444@s.whatsapp.net" }],
+      });
+    });
+
+    await listener.close();
+  });
+
+  it("invalidates cached group metadata on partial group and participant updates", async () => {
+    const groupMetadataCache: NonNullable<InboxMonitorOptions["groupMetadataCache"]> = new Map();
+    const { listener, sock, baileysCache } = await startInboxMonitorWithBaileysCache({
+      groupMetadataCache,
+    });
+    sock.ev.emit("groups.update", [
+      groupMetadata({
+        subject: "Fresh Group",
+      }),
+    ]);
+    await expectCachedGroupMetadata(baileysCache, {
+      id: "123@g.us",
+      subject: "Fresh Group",
+      participants: [{ id: "555@s.whatsapp.net" }],
+    });
+    expect(groupMetadataCache.has("123@g.us")).toBe(true);
+
+    sock.ev.emit("groups.update", [{ id: "123@g.us" }]);
+    expect(groupMetadataCache.has("123@g.us")).toBe(false);
+    await expect(
+      baileysCache.socketOptions.cachedGroupMetadata("123@g.us"),
+    ).resolves.toBeUndefined();
+    sock.ev.emit("groups.update", [
+      groupMetadata({
+        subject: "Fresh Again",
+      }),
+    ]);
+    expect(groupMetadataCache.has("123@g.us")).toBe(true);
+    sock.ev.emit("group-participants.update", { id: "123@g.us" });
+    expect(groupMetadataCache.has("123@g.us")).toBe(false);
+    await expect(
+      baileysCache.socketOptions.cachedGroupMetadata("123@g.us"),
+    ).resolves.toBeUndefined();
+
+    await listener.close();
+  });
+
+  it("expires Baileys retry and group metadata cache entries", async () => {
+    const now = vi.spyOn(Date, "now").mockReturnValue(1_700_000_000_000);
+    const baileysCache = createBaileysCacheSupport();
+    const onMessage = vi.fn(async (_msg: Parameters<InboxOnMessage>[0]) => {});
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      recentMessageKeys: baileysCache.recentMessageKeys,
+      baileysGroupMetaCache: baileysCache.baileysGroupMetaCache,
+    });
+    const messageId = nextMessageId("baileys-expiry");
+    try {
+      sock.ev.emit(
+        "messages.upsert",
+        buildNotifyMessageUpsert({
+          id: messageId,
+          remoteJid: "999@s.whatsapp.net",
+          text: "retry me",
+          timestamp: 1_700_000_000,
+          pushName: "Tester",
+        }),
+      );
+      sock.ev.emit("groups.update", [
+        groupMetadata({
+          subject: "Expiring Group",
+        }),
+      ]);
+      await waitForMessageCalls(onMessage, 1);
+
+      await expect(
+        baileysCache.socketOptions.getMessage({
+          id: messageId,
+          remoteJid: "999@s.whatsapp.net",
+        }),
+      ).resolves.toEqual({ conversation: "retry me" });
+      await expectCachedGroupMetadata(baileysCache, {
+        id: "123@g.us",
+        subject: "Expiring Group",
+        participants: [{ id: "555@s.whatsapp.net" }],
+      });
+
+      now.mockReturnValue(1_700_000_000_000 + 5 * 60 * 1000 + 1);
+      await expect(
+        baileysCache.socketOptions.cachedGroupMetadata("123@g.us"),
+      ).resolves.toBeUndefined();
+
+      now.mockReturnValue(1_700_000_000_000 + 10 * 60 * 1000 + 1);
+      await expect(
+        baileysCache.socketOptions.getMessage({
+          id: messageId,
+          remoteJid: "999@s.whatsapp.net",
+        }),
+      ).resolves.toBeUndefined();
+    } finally {
+      now.mockRestore();
+      await listener.close();
+    }
+  });
+
+  it("does not republish invalidated group metadata from pending hydration", async () => {
+    const groupMetadataCache: NonNullable<InboxMonitorOptions["groupMetadataCache"]> = new Map();
+    const baileysCache = createBaileysCacheSupport();
+    const sock = getSock();
+    let resolveHydration!: (groups: Record<string, GroupMetadata>) => void;
+    sock.groupFetchAllParticipating.mockImplementationOnce(
+      async () =>
+        await new Promise<Record<string, GroupMetadata>>((resolve) => {
+          resolveHydration = resolve;
+        }),
+    );
+
+    const { listener } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, {
+      groupMetadataCache,
+      recentMessageKeys: baileysCache.recentMessageKeys,
+      baileysGroupMetaCache: baileysCache.baileysGroupMetaCache,
+    });
+    sock.ev.emit("groups.update", [{ id: "123@g.us" }]);
+
+    resolveHydration({
+      "123@g.us": groupMetadata({
+        subject: "Stale Hydration Group",
+      }),
+    });
+    await settleInboundWork();
+
+    expect(groupMetadataCache.has("123@g.us")).toBe(false);
+    await expect(
+      baileysCache.socketOptions.cachedGroupMetadata("123@g.us"),
+    ).resolves.toBeUndefined();
+
+    await listener.close();
+  });
+
+  it("cleans up Baileys group metadata listeners on close", async () => {
+    const baileysCache = createBaileysCacheSupport();
+    const { listener, sock } = await startInboxMonitor(vi.fn(async () => {}) as InboxOnMessage, {
+      recentMessageKeys: baileysCache.recentMessageKeys,
+      baileysGroupMetaCache: baileysCache.baileysGroupMetaCache,
+    });
+
+    expect(sock.ev.listenerCount("groups.upsert")).toBe(1);
+    expect(sock.ev.listenerCount("groups.update")).toBe(1);
+    expect(sock.ev.listenerCount("group-participants.update")).toBe(1);
+
+    await listener.close();
+
+    expect(sock.ev.listenerCount("groups.upsert")).toBe(0);
+    expect(sock.ev.listenerCount("groups.update")).toBe(0);
+    expect(sock.ev.listenerCount("group-participants.update")).toBe(0);
+  });
+
   it("bounds cached group metadata kept across reconnects", async () => {
     const groupMetadataCache: NonNullable<InboxMonitorOptions["groupMetadataCache"]> = new Map();
     const groups = Object.fromEntries(
@@ -934,16 +1161,50 @@ describe("web monitor inbox", () => {
     }
   });
 
-  it("suppresses self-echo when a timed-out socket send is later accepted", async () => {
+  it("records outbound replies for Baileys retry lookup", async () => {
     const onMessage = vi.fn(async () => undefined);
     const socketRef = createSocketRef();
+    const baileysCache = createBaileysCacheSupport();
+    const message = { conversation: "pong" };
     const { listener, sock, inbound } = await primeInboundReplyHandle({
       onMessage,
       socketRef,
+      baileysCache,
+      upsertId: "outbound-retry-cache",
+      retryPolicy: fastReconnectPolicy(2),
+    });
+    sock.sendMessage.mockResolvedValueOnce({
+      key: { id: "outbound-cached" },
+      message,
+    });
+
+    await inbound.platform.reply("pong");
+
+    await expect(
+      baileysCache.socketOptions.getMessage({
+        id: "outbound-cached",
+        remoteJid: "999@s.whatsapp.net",
+      }),
+    ).resolves.toBe(message);
+
+    await listener.close();
+  });
+
+  it("suppresses self-echo when a timed-out socket send is later accepted", async () => {
+    const onMessage = vi.fn(async () => undefined);
+    const socketRef = createSocketRef();
+    const baileysCache = createBaileysCacheSupport();
+    const { listener, sock, inbound } = await primeInboundReplyHandle({
+      onMessage,
+      socketRef,
+      baileysCache,
       upsertId: "late-accept",
       retryPolicy: fastReconnectPolicy(2),
     });
-    let acceptLateSend: ((value: { key: { id: string } }) => void) | undefined;
+    const message = { conversation: "pong" };
+    let acceptLateSend:
+      | ((value: { key: { id: string }; message: { conversation: string } }) => void)
+      | undefined;
     vi.useFakeTimers();
     try {
       sock.sendMessage.mockImplementationOnce(
@@ -959,8 +1220,14 @@ describe("web monitor inbox", () => {
       vi.useRealTimers();
     }
 
-    acceptLateSend?.({ key: { id: "late-accepted" } });
+    acceptLateSend?.({ key: { id: "late-accepted" }, message });
     await settleInboundWork();
+    await expect(
+      baileysCache.socketOptions.getMessage({
+        id: "late-accepted",
+        remoteJid: "999@s.whatsapp.net",
+      }),
+    ).resolves.toBe(message);
     sock.ev.emit("messages.upsert", {
       type: "notify",
       messages: [

extensions/whatsapp/src/session.ts

@@ -1,6 +1,7 @@
 // Whatsapp plugin module implements session behavior.
 import { randomUUID } from "node:crypto";
 import type { Agent } from "node:https";
+import type { GroupMetadata, WAMessageKey, proto } from "baileys";
 import { formatCliCommand } from "openclaw/plugin-sdk/cli-runtime";
 import { VERSION } from "openclaw/plugin-sdk/cli-runtime";
 import {
@@ -134,6 +135,8 @@ export async function createWaSocket(
   opts: {
     authDir?: string;
     onQr?: (qr: string) => void;
+    getMessage?: (key: WAMessageKey) => Promise<proto.IMessage | undefined>;
+    cachedGroupMetadata?: (jid: string) => Promise<GroupMetadata | undefined>;
   } & WhatsAppSocketTimingOptions = {},
 ): Promise<ReturnType<typeof makeWASocket>> {
   const baseLogger = getChildLogger(
@@ -185,6 +188,8 @@ export async function createWaSocket(
     // Baileys types still model `fetchAgent` as a Node agent even though the
     // runtime path accepts an undici dispatcher for upload fetches.
     fetchAgent: fetchAgent as Agent | undefined,
+    ...(opts.getMessage ? { getMessage: opts.getMessage } : {}),
+    ...(opts.cachedGroupMetadata ? { cachedGroupMetadata: opts.cachedGroupMetadata } : {}),
   });
 
   sock.ev.on("creds.update", () => enqueueSaveCreds(authDir, saveCreds, sessionLogger));