fix(whatsapp): serialize overlapping Web sends (#99050)

* fix(whatsapp): serialize web sendMessage calls

* fix(whatsapp): make send serialization socket-owned

* test(qa): narrow live gateway connection fields

---------

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

docs/concepts/qa-e2e-automation.md

@@ -893,6 +893,7 @@ Scenario catalog (`extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.
   sticker, and reaction events through the driver.
 - Direct Gateway contract probes: `whatsapp-outbound-media-matrix`,
   `whatsapp-outbound-document-preserves-filename`, `whatsapp-outbound-poll`,
+  `whatsapp-outbound-send-serialization`,
   `whatsapp-group-outbound-media`, `whatsapp-group-outbound-poll`,
   `whatsapp-message-actions`, `whatsapp-reply-context-isolation`,
   `whatsapp-reply-delivery-shape`. These bypass model prompting on purpose
@@ -908,7 +909,7 @@ Scenario catalog (`extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.
 - Status reactions: `whatsapp-status-reactions`,
   `whatsapp-status-reaction-lifecycle`.
 
-The catalog currently contains 51 scenarios. The `live-frontier` default lane
+The catalog currently contains 52 scenarios. The `live-frontier` default lane
 is kept small at 10 scenarios for fast smoke coverage. The `mock-openai`
 default lane runs 45 scenarios deterministically through the real WhatsApp
 transport while mocking only model output; approval scenarios and a few

extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.test.ts

@@ -139,6 +139,7 @@ const DIRECT_GATEWAY_SCENARIO_IDS = [
   "whatsapp-outbound-media-matrix",
   "whatsapp-outbound-document-preserves-filename",
   "whatsapp-outbound-poll",
+  "whatsapp-outbound-send-serialization",
   "whatsapp-message-actions",
   "whatsapp-group-outbound-media",
   "whatsapp-group-outbound-audio",

extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.ts

@@ -18,6 +18,7 @@ import { z } from "zod";
 import { createQaArtifactRunId } from "../../artifact-run-id.js";
 import { QA_EVIDENCE_FILENAME, buildLiveTransportEvidenceSummary } from "../../evidence-summary.js";
 import { startQaGatewayChild } from "../../gateway-child.js";
+import { startQaGatewayRpcClient } from "../../gateway-rpc-client.js";
 import { isTruthyOptIn } from "../../mantis-options.runtime.js";
 import { DEFAULT_QA_LIVE_PROVIDER_MODE } from "../../providers/index.js";
 import { fingerprintQaCredentialId } from "../../qa-credentials-fingerprint.runtime.js";
@@ -93,6 +94,7 @@ type WhatsAppQaScenarioId =
   | "whatsapp-outbound-document-preserves-filename"
   | "whatsapp-outbound-media-matrix"
   | "whatsapp-outbound-poll"
+  | "whatsapp-outbound-send-serialization"
   | "whatsapp-pairing-block"
   | "whatsapp-mention-gating"
   | "whatsapp-reply-delivery-shape"
@@ -158,6 +160,7 @@ const WHATSAPP_QA_SCENARIO_POSTURES = {
   "whatsapp-outbound-document-preserves-filename": "direct-gateway",
   "whatsapp-outbound-media-matrix": "direct-gateway",
   "whatsapp-outbound-poll": "direct-gateway",
+  "whatsapp-outbound-send-serialization": "direct-gateway",
   "whatsapp-pairing-block": "user-path",
   "whatsapp-reply-context-isolation": "direct-gateway",
   "whatsapp-reply-delivery-shape": "direct-gateway",
@@ -186,7 +189,8 @@ type WhatsAppQaMessageSendMode =
     };
 
 type WhatsAppQaGateway = Awaited<ReturnType<typeof startQaGatewayChild>>;
-type WhatsAppQaGatewayRuntime = Pick<WhatsAppQaGateway, "call" | "restart" | "workspaceDir">;
+type WhatsAppQaGatewayRuntime = Pick<WhatsAppQaGateway, "call" | "restart" | "workspaceDir"> &
+  Partial<Pick<WhatsAppQaGateway, "logs" | "token" | "wsUrl">>;
 type WhatsAppQaGatewayCallContext = {
   gateway: Pick<WhatsAppQaGatewayRuntime, "call">;
   gatewayTarget: string;
@@ -1530,6 +1534,43 @@ const WHATSAPP_QA_SCENARIOS: WhatsAppQaScenarioDefinition[] = [
       };
     },
   },
+  {
+    id: "whatsapp-outbound-send-serialization",
+    title: "WhatsApp parallel Gateway sends deliver every outbound message",
+    defaultEnabled: false,
+    defaultProviderModes: ["mock-openai"],
+    timeoutMs: 90_000,
+    buildRun: () => {
+      const token = `WHATSAPP_QA_SERIAL_SEND_${randomUUID().slice(0, 8).toUpperCase()}`;
+      const markers = Array.from({ length: 5 }, (_, index) => `${token}_${index + 1}`);
+      return {
+        afterReply: async (_reply, context) => {
+          const sendsStartedAt = new Date();
+          await callWhatsAppGatewaySendConcurrently(
+            context,
+            markers.map((marker, index) => ({
+              label: `parallel-${index + 1}`,
+              message: marker,
+            })),
+          );
+          await Promise.all(
+            markers.map((marker) =>
+              waitForScenarioObservedMessage(context, {
+                observedAfter: sendsStartedAt,
+                match: (message) => message.kind === "text" && message.text.includes(marker),
+              }),
+            ),
+          );
+          return `gateway parallel send delivered ${markers.length}/${markers.length} messages`;
+        },
+        configMode: "allowlist",
+        expectReply: true,
+        input: `Reply with only this exact marker before parallel send checks: ${token}`,
+        matchText: token,
+        target: "dm",
+      };
+    },
+  },
   {
     id: "whatsapp-outbound-poll",
     title: "WhatsApp direct Gateway poll delivers outbound native poll",
@@ -2878,6 +2919,16 @@ function buildWhatsAppQaIdempotencyKey(scenarioId: WhatsAppQaScenarioId, label:
   return `${scenarioId}:${label}:${randomUUID()}`;
 }
 
+type WhatsAppQaGatewaySendParams = {
+  asVoice?: boolean;
+  forceDocument?: boolean;
+  label: string;
+  mediaUrl?: string;
+  mediaUrls?: string[];
+  message?: string;
+  replyToId?: string;
+};
+
 async function writeWhatsAppQaWorkspaceFixture(
   context: WhatsAppQaMessageScenarioContext,
   params: {
@@ -2894,33 +2945,70 @@ async function writeWhatsAppQaWorkspaceFixture(
 
 async function callWhatsAppGatewaySend(
   context: WhatsAppQaGatewayCallContext,
-  params: {
-    asVoice?: boolean;
-    forceDocument?: boolean;
-    label: string;
-    mediaUrl?: string;
-    mediaUrls?: string[];
-    message?: string;
-    replyToId?: string;
-  },
+  params: WhatsAppQaGatewaySendParams,
 ) {
-  return await context.gateway.call(
-    "send",
-    {
-      accountId: context.sutAccountId,
-      agentId: "main",
-      channel: "whatsapp",
-      idempotencyKey: buildWhatsAppQaIdempotencyKey(context.scenarioId, params.label),
-      to: context.gatewayTarget,
-      ...(params.message !== undefined ? { message: params.message } : {}),
-      ...(params.mediaUrl ? { mediaUrl: params.mediaUrl } : {}),
-      ...(params.mediaUrls ? { mediaUrls: params.mediaUrls } : {}),
-      ...(params.asVoice !== undefined ? { asVoice: params.asVoice } : {}),
-      ...(params.forceDocument !== undefined ? { forceDocument: params.forceDocument } : {}),
-      ...(params.replyToId ? { replyToId: params.replyToId } : {}),
-    },
-    { timeoutMs: 60_000 },
+  return await context.gateway.call("send", buildWhatsAppGatewaySendRequest(context, params), {
+    timeoutMs: 60_000,
+  });
+}
+
+function buildWhatsAppGatewaySendRequest(
+  context: WhatsAppQaGatewayCallContext,
+  params: WhatsAppQaGatewaySendParams,
+) {
+  return {
+    accountId: context.sutAccountId,
+    agentId: "main",
+    channel: "whatsapp",
+    idempotencyKey: buildWhatsAppQaIdempotencyKey(context.scenarioId, params.label),
+    to: context.gatewayTarget,
+    ...(params.message !== undefined ? { message: params.message } : {}),
+    ...(params.mediaUrl ? { mediaUrl: params.mediaUrl } : {}),
+    ...(params.mediaUrls ? { mediaUrls: params.mediaUrls } : {}),
+    ...(params.asVoice !== undefined ? { asVoice: params.asVoice } : {}),
+    ...(params.forceDocument !== undefined ? { forceDocument: params.forceDocument } : {}),
+    ...(params.replyToId ? { replyToId: params.replyToId } : {}),
+  };
+}
+
+async function callWhatsAppGatewaySendConcurrently(
+  context: WhatsAppQaMessageScenarioContext,
+  sends: WhatsAppQaGatewaySendParams[],
+) {
+  // Each QA RPC client serializes its own requests. Separate clients preserve
+  // real Gateway overlap so this probe reaches the shared WhatsApp socket concurrently.
+  const connection = resolveWhatsAppGatewayRpcConnection(context.gateway);
+  const clients = await Promise.all(
+    sends.map(() =>
+      startQaGatewayRpcClient({
+        logs: connection.logs,
+        token: connection.token,
+        wsUrl: connection.wsUrl,
+      }),
+    ),
   );
+  try {
+    await Promise.all(
+      clients.map((client, index) =>
+        client.request("send", buildWhatsAppGatewaySendRequest(context, sends[index]), {
+          timeoutMs: 60_000,
+        }),
+      ),
+    );
+  } finally {
+    await Promise.all(clients.map((client) => client.stop()));
+  }
+}
+
+function resolveWhatsAppGatewayRpcConnection(gateway: WhatsAppQaGatewayRuntime) {
+  if (!gateway.logs || !gateway.token || !gateway.wsUrl) {
+    throw new Error("WhatsApp concurrent Gateway probe requires a live RPC connection.");
+  }
+  return {
+    logs: gateway.logs,
+    token: gateway.token,
+    wsUrl: gateway.wsUrl,
+  };
 }
 
 async function callWhatsAppGatewayPoll(

extensions/whatsapp/src/socket-timing.test.ts

@@ -1,9 +1,11 @@
-import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
 // Whatsapp tests cover socket timing plugin behavior.
+import type { AnyMessageContent, WAMessage } from "baileys";
+import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import {
   DEFAULT_WHATSAPP_SOCKET_TIMING,
   WhatsAppSocketOperationTimeoutError,
+  createWhatsAppSocketOperationTimeoutAdapter,
   isWhatsAppSocketOperationTimeoutError,
   resolveWhatsAppSocketOperationTimeoutMs,
   resolveWhatsAppSocketTiming,
@@ -115,3 +117,85 @@ describe("withWhatsAppSocketOperationTimeout", () => {
     expect(vi.getTimerCount()).toBe(0);
   });
 });
+
+describe("createWhatsAppSocketOperationTimeoutAdapter", () => {
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("serializes sendMessage calls across adapter instances for the same socket", async () => {
+    const started: string[] = [];
+    const resolves: Array<(value: WAMessage) => void> = [];
+    const sock = {
+      sendMessage: vi.fn(
+        async (jid: string, content: AnyMessageContent): Promise<WAMessage | undefined> =>
+          await new Promise((resolve) => {
+            const text =
+              typeof content === "object" && content && "text" in content ? content.text : "";
+            started.push(`${jid}:${text}`);
+            resolves.push(resolve);
+          }),
+      ),
+      sendPresenceUpdate: vi.fn(async () => undefined),
+    };
+
+    const first = createWhatsAppSocketOperationTimeoutAdapter(sock, 30_000).sendMessage(
+      "111@s.whatsapp.net",
+      { text: "first" },
+    );
+    await vi.waitFor(() => expect(started).toEqual(["111@s.whatsapp.net:first"]));
+
+    const second = createWhatsAppSocketOperationTimeoutAdapter(sock, 30_000).sendMessage(
+      "222@s.whatsapp.net",
+      { text: "second" },
+    );
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(started).toEqual(["111@s.whatsapp.net:first"]);
+
+    resolves[0]?.({ key: { id: "msg-1" } } as WAMessage);
+    await expect(first).resolves.toMatchObject({ key: { id: "msg-1" } });
+    await vi.waitFor(() =>
+      expect(started).toEqual(["111@s.whatsapp.net:first", "222@s.whatsapp.net:second"]),
+    );
+
+    resolves[1]?.({ key: { id: "msg-2" } } as WAMessage);
+    await expect(second).resolves.toMatchObject({ key: { id: "msg-2" } });
+  });
+
+  it("releases the send queue after a socket operation timeout", async () => {
+    vi.useFakeTimers();
+    const sendMessage = vi
+      .fn<(jid: string, content: AnyMessageContent) => Promise<WAMessage | undefined>>()
+      .mockImplementationOnce(async () => await new Promise(() => {}))
+      .mockResolvedValueOnce({ key: { id: "msg-2" } } as WAMessage);
+    const sock = {
+      sendMessage,
+      sendPresenceUpdate: vi.fn(async () => undefined),
+    };
+
+    const first = createWhatsAppSocketOperationTimeoutAdapter(sock, 1_000).sendMessage(
+      "111@s.whatsapp.net",
+      { text: "first" },
+    );
+    const firstRejection = expect(first).rejects.toMatchObject({
+      name: "WhatsAppSocketOperationTimeoutError",
+      operation: "sendMessage",
+    });
+    await Promise.resolve();
+    expect(sendMessage).toHaveBeenCalledTimes(1);
+
+    const second = createWhatsAppSocketOperationTimeoutAdapter(sock, 1_000).sendMessage(
+      "222@s.whatsapp.net",
+      { text: "second" },
+    );
+    await Promise.resolve();
+    expect(sendMessage).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(1_000);
+    await firstRejection;
+    await expect(second).resolves.toMatchObject({ key: { id: "msg-2" } });
+    expect(sendMessage).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+});

extensions/whatsapp/src/socket-timing.ts

@@ -27,6 +27,8 @@ type WhatsAppSocketOperationTimeoutHooks = {
   onSendMessageTimeout?: (params: { jid: string; promise: Promise<WAMessage | undefined> }) => void;
 };
 
+const socketSendMessageQueueTails = new WeakMap<WhatsAppSocketOperationAdapter, Promise<void>>();
+
 export const DEFAULT_WHATSAPP_SOCKET_TIMING: Required<WhatsAppSocketTimingOptions> = {
   keepAliveIntervalMs: 25_000,
   connectTimeoutMs: 60_000,
@@ -80,6 +82,27 @@ export function resolveWhatsAppSocketOperationTimeoutMs(timeoutMs: number): numb
   return resolveTimerTimeoutMs(timeoutMs, DEFAULT_WHATSAPP_SOCKET_TIMING.defaultQueryTimeoutMs);
 }
 
+async function runSerializedSocketSendMessage<T>(
+  sock: WhatsAppSocketOperationAdapter,
+  run: () => Promise<T>,
+): Promise<T> {
+  const previous = socketSendMessageQueueTails.get(sock) ?? Promise.resolve();
+  // Adapter instances are short-lived, so key the FIFO by the raw socket. A
+  // bounded send releases the queue after timeout to avoid wedging later work.
+  const result = previous.then(run);
+  const tail = result.then(
+    () => undefined,
+    () => undefined,
+  );
+  socketSendMessageQueueTails.set(sock, tail);
+  void tail.then(() => {
+    if (socketSendMessageQueueTails.get(sock) === tail) {
+      socketSendMessageQueueTails.delete(sock);
+    }
+  });
+  return await result;
+}
+
 export async function withWhatsAppSocketOperationTimeout<T>(
   operation: string,
   promise: Promise<T>,
@@ -114,17 +137,19 @@ export function createWhatsAppSocketOperationTimeoutAdapter(
   const operationTimeoutMs = resolveWhatsAppSocketOperationTimeoutMs(timeoutMs);
   return {
     sendMessage: (jid, content, options) => {
-      const send = options
-        ? sock.sendMessage(jid, content, options)
-        : sock.sendMessage(jid, content);
-      return withWhatsAppSocketOperationTimeout(
-        "sendMessage",
-        send,
-        operationTimeoutMs,
-        hooks?.onSendMessageTimeout
-          ? () => hooks.onSendMessageTimeout?.({ jid, promise: send })
-          : undefined,
-      );
+      return runSerializedSocketSendMessage(sock, () => {
+        const send = options
+          ? sock.sendMessage(jid, content, options)
+          : sock.sendMessage(jid, content);
+        return withWhatsAppSocketOperationTimeout(
+          "sendMessage",
+          send,
+          operationTimeoutMs,
+          hooks?.onSendMessageTimeout
+            ? () => hooks.onSendMessageTimeout?.({ jid, promise: send })
+            : undefined,
+        );
+      });
     },
     sendPresenceUpdate: (presence, jid) => {
       const send =