From 1e7694a8a56f2eb8b8bf8a599b83708479be2100 Mon Sep 17 00:00:00 2001 From: ooiuuii Date: Sun, 5 Jul 2026 14:32:36 +0800 Subject: [PATCH] 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 --- docs/concepts/qa-e2e-automation.md | 3 +- .../whatsapp/whatsapp-live.runtime.test.ts | 1 + .../whatsapp/whatsapp-live.runtime.ts | 140 ++++++++++++++---- extensions/whatsapp/src/socket-timing.test.ts | 86 ++++++++++- extensions/whatsapp/src/socket-timing.ts | 47 ++++-- 5 files changed, 238 insertions(+), 39 deletions(-) diff --git a/docs/concepts/qa-e2e-automation.md b/docs/concepts/qa-e2e-automation.md index e39fa939a514..34f9114a5343 100644 --- a/docs/concepts/qa-e2e-automation.md +++ b/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 diff --git a/extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.test.ts b/extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.test.ts index 072f83df187b..3ef618c68dbc 100644 --- a/extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.test.ts +++ b/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", diff --git a/extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.ts b/extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.ts index 1adabdbcc92d..1c978461c912 100644 --- a/extensions/qa-lab/src/live-transports/whatsapp/whatsapp-live.runtime.ts +++ b/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>; -type WhatsAppQaGatewayRuntime = Pick; +type WhatsAppQaGatewayRuntime = Pick & + Partial>; type WhatsAppQaGatewayCallContext = { gateway: Pick; 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( diff --git a/extensions/whatsapp/src/socket-timing.test.ts b/extensions/whatsapp/src/socket-timing.test.ts index 8387bbff0d43..599a4505bc9a 100644 --- a/extensions/whatsapp/src/socket-timing.test.ts +++ b/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 => + 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>() + .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); + }); +}); diff --git a/extensions/whatsapp/src/socket-timing.ts b/extensions/whatsapp/src/socket-timing.ts index 3b4659ebfbc6..620ec490af70 100644 --- a/extensions/whatsapp/src/socket-timing.ts +++ b/extensions/whatsapp/src/socket-timing.ts @@ -27,6 +27,8 @@ type WhatsAppSocketOperationTimeoutHooks = { onSendMessageTimeout?: (params: { jid: string; promise: Promise }) => void; }; +const socketSendMessageQueueTails = new WeakMap>(); + export const DEFAULT_WHATSAPP_SOCKET_TIMING: Required = { 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( + sock: WhatsAppSocketOperationAdapter, + run: () => Promise, +): Promise { + 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( operation: string, promise: Promise, @@ -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 =