fix(whatsapp): preserve inbound order through adoption

extensions/whatsapp/src/inbound/monitor.ts

@@ -13,9 +13,11 @@ import { recordChannelActivity } from "openclaw/plugin-sdk/channel-activity-runt
 import {
   formatInboundMediaUnavailableText,
   formatLocationText,
+  shouldDebounceTextInbound,
   type MediaPlaceholderTextFact,
 } from "openclaw/plugin-sdk/channel-inbound";
 import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { getChildLogger } from "openclaw/plugin-sdk/logging-core";
 import {
   asDateTimestampMs,
@@ -427,9 +429,25 @@ export async function attachWebInboxToSocket(
     }
     return successor.getCurrentSock();
   };
+  type InboundTurnSettlement = {
+    promise: Promise<void>;
+    resolve: () => void;
+    settled: boolean;
+  };
   type QueuedInboundMessageMetadata = {
     admission: AdmittedWebInboundCallbackMessage["admission"];
     debounceKey?: string;
+    debounceConversationKey?: string;
+    debounceCompletion?: {
+      key: string;
+      conversationKey: string;
+      promise: Promise<void>;
+      resolve: () => void;
+      flushing: boolean;
+      settled: boolean;
+    };
+    debounceEligible?: boolean;
+    turnSettlement?: InboundTurnSettlement;
     turnAdoptionLifecycle?: WhatsAppIngressLifecycle;
     readReceipt?: WhatsAppReadReceiptTarget;
     receiveOrder?: number;
@@ -439,6 +457,14 @@ export async function attachWebInboxToSocket(
     options.durableInboundQueue ?? createWhatsAppDurableInboundQueue(options.accountId);
   const inboundDebounceMs = Math.max(0, Math.trunc(options.debounceMs ?? 0));
   const pendingDebounceKeys = new Set<string>();
+  const pendingDebounceCompletionsByKey = new Map<
+    string,
+    Set<NonNullable<QueuedInboundMessage["debounceCompletion"]>>
+  >();
+  const pendingDebounceCompletionsByConversation = new Map<
+    string,
+    Set<NonNullable<QueuedInboundMessage["debounceCompletion"]>>
+  >();
   const activeInboundFlushes = new Set<Promise<void>>();
   const pendingMessageHandlers = new Set<Promise<void>>();
   let durableIngressActive = false;
@@ -472,6 +498,7 @@ export async function attachWebInboxToSocket(
     });
   };
   let nextReceiveOrder = 0;
+  let messagesUpsertTail: Promise<void> = Promise.resolve();
   const publishPendingWorkState = (at = Date.now()) => {
     options.onPendingWorkChanged?.(
       pendingMessageHandlers.size +
@@ -497,8 +524,33 @@ export async function attachWebInboxToSocket(
     }
     return `${admission.accountId}:${admission.conversation.id}:${senderKey}`;
   };
+  const buildInboundDebounceConversationKey = (msg: QueuedInboundMessage): string => {
+    const admission = requireWhatsAppInboundAdmission(msg);
+    return `${admission.accountId}:${admission.conversation.id}`;
+  };
   const shouldDebounceInboundMessage = (msg: AdmittedWebInboundCallbackMessage): boolean =>
-    options.shouldDebounce?.(msg) ?? true;
+    shouldDebounceTextInbound({
+      text: msg.payload.commandBody ?? msg.payload.body,
+      cfg: options.cfg,
+      hasMedia: Boolean(msg.payload.media),
+      allowDebounce: !msg.payload.location && !msg.quote,
+    }) &&
+    (options.shouldDebounce?.(msg) ?? true);
+  // Eligible raw text claims need independent durable lanes until the channel
+  // buffer adopts them together. The conversation queue and flush barrier below
+  // preserve ordering when full normalization or a custom predicate vetoes them.
+  const canEnterDebounceBeforeConversationAdoption = (msg: WAMessage): boolean => {
+    if (inboundDebounceMs <= 0) {
+      return false;
+    }
+    const rawMessage = msg.message ?? undefined;
+    return shouldDebounceTextInbound({
+      text: extractText(rawMessage),
+      cfg: options.cfg,
+      hasMedia: Boolean(extractMediaKind(rawMessage)),
+      allowDebounce: !extractLocationData(rawMessage) && !describeReplyContext(rawMessage),
+    });
+  };
   const orderDebouncedInboundEntries = (entries: QueuedInboundMessage[]) =>
     entries.toSorted((a, b) => {
       const timestampDiff = (a.event.timestamp ?? 0) - (b.event.timestamp ?? 0);
@@ -508,6 +560,110 @@ export async function attachWebInboxToSocket(
       return (a.receiveOrder ?? 0) - (b.receiveOrder ?? 0);
     });
 
+  const createInboundTurnSettlement = (): InboundTurnSettlement => {
+    let resolvePromise!: () => void;
+    const settlement: InboundTurnSettlement = {
+      promise: new Promise<void>((resolve) => {
+        resolvePromise = resolve;
+      }),
+      resolve: () => {
+        if (settlement.settled) {
+          return;
+        }
+        settlement.settled = true;
+        resolvePromise();
+      },
+      settled: false,
+    };
+    return settlement;
+  };
+
+  const waitForInboundTurnSettlement = async (
+    settlement: InboundTurnSettlement,
+    abortSignal?: AbortSignal,
+  ): Promise<void> => {
+    if (!abortSignal) {
+      await settlement.promise;
+      return;
+    }
+    if (abortSignal.aborted) {
+      return;
+    }
+    await new Promise<void>((resolve) => {
+      const finish = () => {
+        abortSignal.removeEventListener("abort", finish);
+        resolve();
+      };
+      abortSignal.addEventListener("abort", finish, { once: true });
+      void settlement.promise.then(finish, finish);
+    });
+  };
+
+  const registerDebounceCompletion = (params: {
+    key: string;
+    conversationKey: string;
+  }): NonNullable<QueuedInboundMessage["debounceCompletion"]> => {
+    let resolvePromise!: () => void;
+    const completion: NonNullable<QueuedInboundMessage["debounceCompletion"]> = {
+      ...params,
+      promise: new Promise<void>((resolve) => {
+        resolvePromise = resolve;
+      }),
+      resolve: () => {
+        if (completion.settled) {
+          return;
+        }
+        completion.settled = true;
+        resolvePromise();
+        const keyCompletions = pendingDebounceCompletionsByKey.get(completion.key);
+        keyCompletions?.delete(completion);
+        if (keyCompletions?.size === 0) {
+          pendingDebounceCompletionsByKey.delete(completion.key);
+          pendingDebounceKeys.delete(completion.key);
+        }
+        const conversationCompletions = pendingDebounceCompletionsByConversation.get(
+          completion.conversationKey,
+        );
+        conversationCompletions?.delete(completion);
+        if (conversationCompletions?.size === 0) {
+          pendingDebounceCompletionsByConversation.delete(completion.conversationKey);
+        }
+      },
+      flushing: false,
+      settled: false,
+    };
+    const keyCompletions = pendingDebounceCompletionsByKey.get(params.key) ?? new Set();
+    keyCompletions.add(completion);
+    pendingDebounceCompletionsByKey.set(params.key, keyCompletions);
+    const conversationCompletions =
+      pendingDebounceCompletionsByConversation.get(params.conversationKey) ?? new Set();
+    conversationCompletions.add(completion);
+    pendingDebounceCompletionsByConversation.set(params.conversationKey, conversationCompletions);
+    pendingDebounceKeys.add(params.key);
+    return completion;
+  };
+
+  const flushPriorConversationDebounce = async (
+    conversationKey: string,
+    retainedKey?: string,
+  ): Promise<void> => {
+    const completions = [
+      ...(pendingDebounceCompletionsByConversation.get(conversationKey) ?? []),
+    ].filter((completion) => completion.key !== retainedKey || completion.flushing);
+    const keys = new Set(completions.map((completion) => completion.key));
+    await Promise.all([...keys].map((key) => debouncer.flushKey(key)));
+    await Promise.all(completions.map((completion) => completion.promise));
+    if (!retainedKey) {
+      return;
+    }
+    // The retained buffer may cross into onFlush while the awaits above yield. Recheck before
+    // enqueueing: if it is flushing, this message must form a later batch after adoption settles.
+    const racedRetainedCompletions = [
+      ...(pendingDebounceCompletionsByConversation.get(conversationKey) ?? []),
+    ].filter((completion) => completion.key === retainedKey && completion.flushing);
+    await Promise.all(racedRetainedCompletions.map((completion) => completion.promise));
+  };
+
   const buildFlushIngressLifecycle = (entries: QueuedInboundMessage[]) => {
     const lifecycles = entries
       .map((entry) => entry.turnAdoptionLifecycle)
@@ -518,20 +674,41 @@ export async function attachWebInboxToSocket(
         lifecycle: undefined,
         settle: async () => {},
         abandon: async () => {},
+        waitForSettlement: async () => {},
       };
     }
     let handedOff = false;
+    let settled = false;
+    let resolveSettlement!: () => void;
+    const settlement = new Promise<void>((resolve) => {
+      resolveSettlement = resolve;
+    });
+    const markSettled = () => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      resolveSettlement();
+    };
+    const abortSignal =
+      lifecycles.length === 1
+        ? firstLifecycle.abortSignal
+        : AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal));
+    if (abortSignal.aborted) {
+      markSettled();
+    } else {
+      abortSignal.addEventListener("abort", markSettled, { once: true });
+    }
     const adoptAll = async () => {
-      for (const lifecycle of lifecycles) {
-        await lifecycle.onAdopted();
+      try {
+        await Promise.all(lifecycles.map((lifecycle) => lifecycle.onAdopted()));
+      } finally {
+        markSettled();
       }
     };
     return {
       lifecycle: {
-        abortSignal:
-          lifecycles.length === 1
-            ? firstLifecycle.abortSignal
-            : AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)),
+        abortSignal,
         onAdopted: async () => {
           handedOff = true;
           await adoptAll();
@@ -549,9 +726,13 @@ export async function attachWebInboxToSocket(
         },
         onAbandoned: async () => {
           handedOff = true;
-          await Promise.all(
-            lifecycles.map((lifecycle) => Promise.resolve(lifecycle.onAbandoned())),
-          );
+          try {
+            await Promise.all(
+              lifecycles.map((lifecycle) => Promise.resolve(lifecycle.onAbandoned())),
+            );
+          } finally {
+            markSettled();
+          }
         },
       } satisfies WhatsAppIngressLifecycle,
       // Gated or otherwise terminal no-dispatch turns still own every merged claim.
@@ -563,19 +744,32 @@ export async function attachWebInboxToSocket(
       abandon: async () => {
         if (!handedOff) {
           handedOff = true;
-          await Promise.all(
-            lifecycles.map((lifecycle) => Promise.resolve(lifecycle.onAbandoned())),
-          );
+          try {
+            await Promise.all(
+              lifecycles.map((lifecycle) => Promise.resolve(lifecycle.onAbandoned())),
+            );
+          } finally {
+            markSettled();
+          }
         }
       },
+      waitForSettlement: async () => {
+        await settlement;
+      },
     };
   };
 
   const debouncer = createInboundDebouncer<QueuedInboundMessage>({
     debounceMs: inboundDebounceMs,
     buildKey: (msg) => msg.debounceKey ?? buildInboundDebounceKey(msg),
-    shouldDebounce: shouldDebounceInboundMessage,
+    shouldDebounce: (msg) => msg.debounceEligible === true,
+    serializeImmediate: true,
     onFlush: async (entries) => {
+      for (const entry of entries) {
+        if (entry.debounceCompletion) {
+          entry.debounceCompletion.flushing = true;
+        }
+      }
       let finishFlush!: () => void;
       const flushTask = new Promise<void>((resolve) => {
         finishFlush = resolve;
@@ -583,13 +777,16 @@ export async function attachWebInboxToSocket(
       activeInboundFlushes.add(flushTask);
       publishPendingWorkState();
       notifyDebounceWork();
+      let waitForSettlement = async () => {};
       try {
         const orderedEntries = orderDebouncedInboundEntries(entries);
         const last = orderedEntries.at(-1);
         if (!last) {
           return;
         }
-        const { lifecycle, settle, abandon } = buildFlushIngressLifecycle(orderedEntries);
+        const flushLifecycle = buildFlushIngressLifecycle(orderedEntries);
+        const { lifecycle, settle, abandon } = flushLifecycle;
+        waitForSettlement = flushLifecycle.waitForSettlement;
         try {
           if (orderedEntries.length === 1) {
             await options.onMessage(attachWhatsAppIngressLifecycle(last, lifecycle));
@@ -654,11 +851,27 @@ export async function attachWebInboxToSocket(
           throw error;
         }
       } finally {
-        for (const entry of entries) {
-          if (entry.debounceKey) {
-            pendingDebounceKeys.delete(entry.debounceKey);
+        // Durable adoption can remain deferred after dispatch returns. Keep the
+        // ordering completion pending without holding the debouncer's key task.
+        const resolveOrderingCompletions = () => {
+          // Adoption requests the durable monitor's terminal drain before this settlement resolves;
+          // its activity callback publishes the final count after these keys clear.
+          for (const entry of entries) {
+            entry.debounceCompletion?.resolve();
+            entry.turnSettlement?.resolve();
           }
-        }
+        };
+        void waitForSettlement().then(resolveOrderingCompletions, (error: unknown) => {
+          resolveOrderingCompletions();
+          try {
+            inboundLogger.error(
+              { error: String(error) },
+              "failed waiting for WhatsApp ingress adoption settlement",
+            );
+          } catch {
+            // Bookkeeping must complete even if failure reporting is unavailable.
+          }
+        });
         activeInboundFlushes.delete(flushTask);
         finishFlush();
         publishPendingWorkState();
@@ -1294,7 +1507,7 @@ export async function attachWebInboxToSocket(
       receiveOrder?: number;
       turnAdoptionLifecycle?: WhatsAppIngressLifecycle;
     },
-  ) => {
+  ): Promise<"debounced" | "immediate"> => {
     const chatJid = inbound.remoteJid;
     const sendComposing = async () => {
       const currentSock = getCurrentSock();
@@ -1449,10 +1662,24 @@ export async function attachWebInboxToSocket(
       receiveOrder: durable.receiveOrder,
     });
     const debounceKey = buildInboundDebounceKey(inboundMessage);
+    const debounceConversationKey = buildInboundDebounceConversationKey(inboundMessage);
+    const debounceEligible =
+      inboundDebounceMs > 0 && Boolean(debounceKey) && shouldDebounceInboundMessage(inboundMessage);
+    inboundMessage.debounceConversationKey = debounceConversationKey;
+    inboundMessage.debounceEligible = debounceEligible;
+    const turnSettlement = createInboundTurnSettlement();
+    inboundMessage.turnSettlement = turnSettlement;
+    await flushPriorConversationDebounce(
+      debounceConversationKey,
+      debounceEligible ? (debounceKey ?? undefined) : undefined,
+    );
     if (debounceKey) {
       inboundMessage.debounceKey = debounceKey;
-      if (inboundDebounceMs > 0 && shouldDebounceInboundMessage(inboundMessage)) {
-        pendingDebounceKeys.add(debounceKey);
+      if (debounceEligible) {
+        inboundMessage.debounceCompletion = registerDebounceCompletion({
+          key: debounceKey,
+          conversationKey: debounceConversationKey,
+        });
         publishPendingWorkState();
         notifyDebounceWork();
       }
@@ -1475,7 +1702,20 @@ export async function attachWebInboxToSocket(
         },
       );
     }
-    await debouncer.enqueue(inboundMessage);
+    try {
+      await debouncer.enqueue(inboundMessage);
+    } catch (error) {
+      inboundMessage.debounceCompletion?.resolve();
+      turnSettlement.resolve();
+      throw error;
+    }
+    if (!debounceEligible) {
+      await waitForInboundTurnSettlement(
+        turnSettlement,
+        durable.turnAdoptionLifecycle?.abortSignal,
+      );
+    }
+    return debounceEligible ? "debounced" : "immediate";
   };
 
   const processDurableInboundMessage = async (
@@ -1540,12 +1780,15 @@ export async function attachWebInboxToSocket(
     return "deferred";
   };
 
+  const durableConversationQueue = new KeyedAsyncQueue();
+
   const durableInboundMonitor = createWhatsAppIngressMonitor({
     queue: durableInboundQueue,
     dispatch: async (msg, payload, lifecycle) => {
       const remoteJid = msg.key?.remoteJid;
       const id = msg.key?.id;
-      return {
+      const conversationKey = remoteJid?.trim() || id?.trim() || "invalid-whatsapp-conversation";
+      return await durableConversationQueue.enqueue(conversationKey, async () => ({
         kind: await processDurableInboundMessage(msg, {
           ...payload,
           ...(remoteJid && id
@@ -1553,7 +1796,7 @@ export async function attachWebInboxToSocket(
             : {}),
           lifecycle,
         }),
-      };
+      }));
     },
     pollIntervalMs: WHATSAPP_INGRESS_DRAIN_INTERVAL_MS,
     onLog: (message) => inboundLogger.warn({ message }, "whatsapp ingress drain"),
@@ -1569,10 +1812,14 @@ export async function attachWebInboxToSocket(
     if (upsert.type !== "notify" && upsert.type !== "append") {
       return;
     }
-    for (const msg of upsert.messages ?? []) {
+    // Stamp the whole transport batch before any async policy work. receiveOrder
+    // preserves intra-socket ordering without changing wall-clock age semantics.
+    const receivedMessages = (upsert.messages ?? []).map((msg) => {
+      return { msg, receiveOrder: nextReceiveOrder++, receivedAt: Date.now() };
+    });
+    for (const { msg, receiveOrder, receivedAt } of receivedMessages) {
       rememberBaileysMessage(msg.key?.remoteJid, msg.key?.id, msg.message);
 
-      const receiveOrder = nextReceiveOrder++;
       if (
         await maybeResolveWhatsAppApprovalReaction({
           cfg: options.loadConfig?.() ?? options.cfg,
@@ -1588,7 +1835,6 @@ export async function attachWebInboxToSocket(
         continue;
       }
 
-      const receivedAt = Date.now();
       const skipStaleAppend = shouldSkipStaleAppend(msg, upsert.type);
       const skipRecentOutboundEcho = shouldSkipRecentOutboundEcho(msg);
       const remoteJid = msg.key?.remoteJid;
@@ -1647,6 +1893,7 @@ export async function attachWebInboxToSocket(
             skipRecentOutboundEcho,
             receivedAt,
             receiveOrder,
+            allowConcurrentDebounce: canEnterDebounceBeforeConversationAdoption(msg),
           });
           appendError = undefined;
           break;
@@ -1693,20 +1940,47 @@ export async function attachWebInboxToSocket(
     }
   };
   const handleMessagesUpsertEvent = (upsert: { type?: string; messages?: Array<WAMessage> }) => {
-    const task = handleMessagesUpsert(upsert).catch((err: unknown) => {
-      inboundLogger.error({ error: String(err) }, "messages.upsert handler error");
-      inboundConsoleLog.error(`Messages upsert handler error: ${String(err)}`);
-    });
+    // Baileys listeners are fire-and-forget. Preserve emitter order through
+    // durable admission so a later callback cannot persist and drain first.
+    const task = messagesUpsertTail
+      .then(() => handleMessagesUpsert(upsert))
+      .catch((err: unknown) => {
+        try {
+          inboundLogger.error({ error: String(err) }, "messages.upsert handler error");
+        } catch {
+          // Logging must not poison the serialized admission tail.
+        }
+        try {
+          inboundConsoleLog.error(`Messages upsert handler error: ${String(err)}`);
+        } catch {
+          // Logging must not poison the serialized admission tail.
+        }
+      });
+    messagesUpsertTail = task.then(
+      () => undefined,
+      () => undefined,
+    );
     pendingMessageHandlers.add(task);
     publishPendingWorkState();
-    void task.finally(() => {
-      pendingMessageHandlers.delete(task);
-      publishPendingWorkState();
-    });
+    void task.then(
+      () => {
+        pendingMessageHandlers.delete(task);
+        publishPendingWorkState();
+      },
+      () => {
+        pendingMessageHandlers.delete(task);
+        publishPendingWorkState();
+      },
+    );
   };
   const drainDebouncedInboundMessages = async () => {
-    while (pendingDebounceKeys.size > 0 || activeInboundFlushes.size > 0) {
-      const debounceKeys = Array.from(pendingDebounceKeys);
+    for (;;) {
+      const debounceKeys = [...pendingDebounceCompletionsByKey.entries()]
+        .filter(([, completions]) => [...completions].some((completion) => !completion.flushing))
+        .map(([key]) => key);
+      if (debounceKeys.length === 0 && activeInboundFlushes.size === 0) {
+        return;
+      }
       if (debounceKeys.length > 0) {
         await Promise.all(debounceKeys.map((key) => debouncer.flushKey(key)));
       }

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

@@ -1044,6 +1044,41 @@ describe("web monitor inbox", () => {
     }
   });
 
+  it("batches rapid durable text messages before dispatching the conversation", async () => {
+    vi.useFakeTimers();
+    try {
+      const onMessage = vi.fn(async () => undefined);
+      const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+        debounceMs: 250,
+      });
+      const first = buildNotifyMessageUpsert({
+        id: nextMessageId("durable-debounce-1"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "first",
+        timestamp: 1_700_000_000,
+        pushName: "Tester",
+      });
+      const second = buildNotifyMessageUpsert({
+        id: nextMessageId("durable-debounce-2"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "second",
+        timestamp: 1_700_000_001,
+        pushName: "Tester",
+      });
+      sock.ev.emit("messages.upsert", {
+        type: "notify",
+        messages: [...first.messages, ...second.messages],
+      });
+      await vi.advanceTimersByTimeAsync(250);
+      await waitForMessageCalls(onMessage, 1);
+
+      expect(inboundMessage(onMessage).payload.body).toBe("first\nsecond");
+      await listener.close();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("completes shutdown under a long durable debounce without waiting for the window", async () => {
     // Durable pump tasks await claim flush waiters; close must force-flush
     // debounced batches before waiting on those pumps (socket-close timeout).
@@ -2034,6 +2069,415 @@ describe("web monitor inbox", () => {
     await listener.close();
   });
 
+  it("keeps control commands serialized when text debounce is enabled", async () => {
+    let adoptFirst: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      if (!adoptFirst) {
+        const lifecycle = resolveWhatsAppIngressLifecycle(message);
+        if (!lifecycle) {
+          throw new Error("expected durable ingress lifecycle");
+        }
+        lifecycle.onDeferred();
+        adoptFirst = lifecycle.onAdopted;
+      }
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs: 250,
+    });
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("serialized-command-1"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "/status",
+        timestamp: 1_700_000_000,
+      }),
+    );
+    await waitForMessageCalls(onMessage, 1);
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("serialized-command-2"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "/status plugins",
+        timestamp: 1_700_000_001,
+      }),
+    );
+    await settleInboundWork();
+
+    expect(onMessage).toHaveBeenCalledTimes(1);
+    if (!adoptFirst) {
+      throw new Error("expected first adoption callback");
+    }
+    await adoptFirst();
+    await waitForMessageCalls(onMessage, 2);
+    expect(inboundMessage(onMessage, 1).payload.body).toBe("/status plugins");
+    await listener.close();
+  });
+
+  it("flushes and adopts buffered text before dispatching a later control command", async () => {
+    let adoptText: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      if (!adoptText) {
+        const lifecycle = resolveWhatsAppIngressLifecycle(message);
+        if (!lifecycle) {
+          throw new Error("expected durable ingress lifecycle");
+        }
+        lifecycle.onDeferred();
+        adoptText = lifecycle.onAdopted;
+      }
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs: 250,
+    });
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("text-before-command"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "first text",
+        timestamp: 1_700_000_000,
+      }),
+    );
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("command-after-text"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "/status",
+        timestamp: 1_700_000_001,
+      }),
+    );
+
+    await waitForMessageCalls(onMessage, 1);
+    expect(inboundMessage(onMessage).payload.body).toBe("first text");
+    await settleInboundWork();
+    expect(onMessage).toHaveBeenCalledTimes(1);
+
+    if (!adoptText) {
+      throw new Error("expected text adoption callback");
+    }
+    await adoptText();
+    await waitForMessageCalls(onMessage, 2);
+    expect(inboundMessage(onMessage, 1).payload.body).toBe("/status");
+    await listener.close();
+  });
+
+  it("serializes a custom debounce veto with later eligible text", async () => {
+    let adoptImmediate: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      if (!adoptImmediate) {
+        const lifecycle = resolveWhatsAppIngressLifecycle(message);
+        if (!lifecycle) {
+          throw new Error("expected durable ingress lifecycle");
+        }
+        lifecycle.onDeferred();
+        adoptImmediate = lifecycle.onAdopted;
+      }
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs: 50,
+      shouldDebounce: (message) => message.payload.body !== "do not debounce",
+    });
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("custom-veto"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "do not debounce",
+        timestamp: 1_700_000_000,
+      }),
+    );
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("after-custom-veto"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "debounce me",
+        timestamp: 1_700_000_001,
+      }),
+    );
+
+    await waitForMessageCalls(onMessage, 1);
+    expect(inboundMessage(onMessage).payload.body).toBe("do not debounce");
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    expect(onMessage).toHaveBeenCalledTimes(1);
+
+    if (!adoptImmediate) {
+      throw new Error("expected immediate adoption callback");
+    }
+    await adoptImmediate();
+    await waitForMessageCalls(onMessage, 2);
+    expect(inboundMessage(onMessage, 1).payload.body).toBe("debounce me");
+    await listener.close();
+  });
+
+  it("holds a new same-sender debounce batch behind an in-flight adoption", async () => {
+    let adoptFirstBatch: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      if (!adoptFirstBatch) {
+        const lifecycle = resolveWhatsAppIngressLifecycle(message);
+        if (!lifecycle) {
+          throw new Error("expected durable ingress lifecycle");
+        }
+        lifecycle.onDeferred();
+        adoptFirstBatch = lifecycle.onAdopted;
+      }
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs: 20,
+    });
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("first-debounce-batch"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "first batch",
+        timestamp: 1_700_000_000,
+      }),
+    );
+    await waitForMessageCalls(onMessage, 1);
+
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("second-debounce-batch"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "second batch",
+        timestamp: 1_700_000_001,
+      }),
+    );
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    expect(onMessage).toHaveBeenCalledTimes(1);
+
+    if (!adoptFirstBatch) {
+      throw new Error("expected first batch adoption callback");
+    }
+    await adoptFirstBatch();
+    await waitForMessageCalls(onMessage, 2);
+    expect(inboundMessage(onMessage, 1).payload.body).toBe("second batch");
+    await listener.close();
+  });
+
+  it("atomically joins or waits when a same-sender debounce timer starts flushing", async () => {
+    const debounceMs = 12_345;
+    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    let debounceChecks = 0;
+    let firstDebounceTimerIndex = -1;
+    let adoptFirstBatch: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      if (adoptFirstBatch) {
+        return;
+      }
+      const lifecycle = resolveWhatsAppIngressLifecycle(message);
+      if (!lifecycle) {
+        throw new Error("expected durable ingress lifecycle");
+      }
+      lifecycle.onDeferred();
+      adoptFirstBatch = lifecycle.onAdopted;
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs,
+      shouldDebounce: () => {
+        debounceChecks += 1;
+        if (debounceChecks === 2) {
+          firstDebounceTimerIndex = setTimeoutSpy.mock.calls.findLastIndex(
+            (call) => call[1] === debounceMs,
+          );
+          const timer = setTimeoutSpy.mock.results[firstDebounceTimerIndex]?.value as
+            | ReturnType<typeof setTimeout>
+            | undefined;
+          const flush = setTimeoutSpy.mock.calls[firstDebounceTimerIndex]?.[0];
+          if (timer) {
+            clearTimeout(timer);
+          }
+          queueMicrotask(() => {
+            if (typeof flush === "function") {
+              flush();
+            }
+          });
+        }
+        return true;
+      },
+    });
+    try {
+      sock.ev.emit(
+        "messages.upsert",
+        buildNotifyMessageUpsert({
+          id: nextMessageId("debounce-boundary-first"),
+          remoteJid: "999@s.whatsapp.net",
+          text: "first boundary batch",
+          timestamp: 1_700_000_000,
+        }),
+      );
+      await vi.waitFor(() => expect(debounceChecks).toBe(1));
+
+      sock.ev.emit(
+        "messages.upsert",
+        buildNotifyMessageUpsert({
+          id: nextMessageId("debounce-boundary-second"),
+          remoteJid: "999@s.whatsapp.net",
+          text: "second boundary batch",
+          timestamp: 1_700_000_001,
+        }),
+      );
+      await waitForMessageCalls(onMessage, 1);
+      expect(inboundMessage(onMessage).payload.body).toBe("first boundary batch");
+      expect(firstDebounceTimerIndex).toBeGreaterThanOrEqual(0);
+      expect(
+        setTimeoutSpy.mock.calls.findIndex(
+          (call, index) => index > firstDebounceTimerIndex && call[1] === debounceMs,
+        ),
+      ).toBe(-1);
+
+      if (!adoptFirstBatch) {
+        throw new Error("expected first boundary batch adoption callback");
+      }
+      await adoptFirstBatch();
+      await vi.waitFor(() => {
+        expect(
+          setTimeoutSpy.mock.calls.findIndex(
+            (call, index) => index > firstDebounceTimerIndex && call[1] === debounceMs,
+          ),
+        ).toBeGreaterThan(firstDebounceTimerIndex);
+      });
+      const secondDebounceTimerIndex = setTimeoutSpy.mock.calls.findIndex(
+        (call, index) => index > firstDebounceTimerIndex && call[1] === debounceMs,
+      );
+      const secondTimer = setTimeoutSpy.mock.results[secondDebounceTimerIndex]?.value as
+        | ReturnType<typeof setTimeout>
+        | undefined;
+      const flushSecond = setTimeoutSpy.mock.calls[secondDebounceTimerIndex]?.[0];
+      if (secondTimer) {
+        clearTimeout(secondTimer);
+      }
+      if (typeof flushSecond !== "function") {
+        throw new Error("expected second boundary debounce timer callback");
+      }
+      flushSecond();
+      await waitForMessageCalls(onMessage, 2);
+      expect(inboundMessage(onMessage, 1).payload.body).toBe("second boundary batch");
+    } finally {
+      await adoptFirstBatch?.();
+      await listener.close();
+      setTimeoutSpy.mockRestore();
+    }
+  });
+
+  it("publishes idle after deferred debounce adoption settles", async () => {
+    const debounceMs = 12_345;
+    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    const onPendingWorkChanged = vi.fn();
+    let adoptBatch: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      const lifecycle = resolveWhatsAppIngressLifecycle(message);
+      if (!lifecycle) {
+        throw new Error("expected durable ingress lifecycle");
+      }
+      lifecycle.onDeferred();
+      adoptBatch = lifecycle.onAdopted;
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs,
+      onPendingWorkChanged,
+    });
+    try {
+      sock.ev.emit(
+        "messages.upsert",
+        buildNotifyMessageUpsert({
+          id: nextMessageId("deferred-debounce-pending-work"),
+          remoteJid: "999@s.whatsapp.net",
+          text: "deferred batch",
+          timestamp: 1_700_000_000,
+        }),
+      );
+      await vi.waitFor(() => {
+        expect(setTimeoutSpy.mock.calls.some((call) => call[1] === debounceMs)).toBe(true);
+      });
+      const timerIndex = setTimeoutSpy.mock.calls.findIndex((call) => call[1] === debounceMs);
+      const timer = setTimeoutSpy.mock.results[timerIndex]?.value as
+        | ReturnType<typeof setTimeout>
+        | undefined;
+      const flush = setTimeoutSpy.mock.calls[timerIndex]?.[0];
+      if (timer) {
+        clearTimeout(timer);
+      }
+      if (typeof flush !== "function") {
+        throw new Error("expected debounce timer callback");
+      }
+      flush();
+      await waitForMessageCalls(onMessage, 1);
+      await vi.waitFor(() => {
+        expect(onPendingWorkChanged.mock.calls.at(-1)?.[0]).toBe(1);
+      });
+
+      if (!adoptBatch) {
+        throw new Error("expected deferred batch adoption callback");
+      }
+      await adoptBatch();
+      await vi.waitFor(() => {
+        expect(onPendingWorkChanged.mock.calls.at(-1)?.[0]).toBe(0);
+      });
+    } finally {
+      await adoptBatch?.();
+      await listener.close();
+      setTimeoutSpy.mockRestore();
+    }
+  });
+
+  it("does not block an unrelated conversation on a deferred debounce adoption", async () => {
+    let adoptFirstConversation: (() => void | Promise<void>) | undefined;
+    const onMessage = vi.fn(async (message: WebInboundMessage) => {
+      if (message.platform.chatJid !== "999@s.whatsapp.net" || adoptFirstConversation) {
+        return;
+      }
+      const lifecycle = resolveWhatsAppIngressLifecycle(message);
+      if (!lifecycle) {
+        throw new Error("expected durable ingress lifecycle");
+      }
+      lifecycle.onDeferred();
+      adoptFirstConversation = lifecycle.onAdopted;
+    });
+
+    const { listener, sock } = await startInboxMonitor(onMessage as InboxOnMessage, {
+      debounceMs: 20,
+    });
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("deferred-first-conversation"),
+        remoteJid: "999@s.whatsapp.net",
+        text: "first conversation",
+        timestamp: 1_700_000_000,
+      }),
+    );
+    await waitForMessageCalls(onMessage, 1);
+
+    sock.ev.emit(
+      "messages.upsert",
+      buildNotifyMessageUpsert({
+        id: nextMessageId("independent-conversation"),
+        remoteJid: "888@s.whatsapp.net",
+        text: "/status",
+        timestamp: 1_700_000_001,
+      }),
+    );
+    await waitForMessageCalls(onMessage, 2);
+    expect(inboundMessage(onMessage, 1).platform.chatJid).toBe("888@s.whatsapp.net");
+
+    if (!adoptFirstConversation) {
+      throw new Error("expected first conversation adoption callback");
+    }
+    await adoptFirstConversation();
+    await listener.close();
+  });
+
   it("captures reply context from quoted messages", async () => {
     await expectQuotedReplyContext({ conversation: "original" });
   });