diff --git a/extensions/whatsapp/src/inbound.media.test.ts b/extensions/whatsapp/src/inbound.media.test.ts index 0def2b23e6cb..49f7bf383f4d 100644 --- a/extensions/whatsapp/src/inbound.media.test.ts +++ b/extensions/whatsapp/src/inbound.media.test.ts @@ -263,9 +263,11 @@ const QUOTED_SYNTHETIC_API_KEY = "synthetic-quoted-api-key-never-real"; const DEPLOYMENT_REDACTION_SENTINEL = "deployment-secret-never-real"; async function waitForMessage(onMessage: ReturnType) { + // Saturated no-isolate suite runs can stall the worker (sync module fetches + // against the shared transform queue) well past a 2s delivery budget. await vi.waitFor(() => expect(onMessage).toHaveBeenCalledTimes(1), { interval: 1, - timeout: 2_000, + timeout: 5_000, }); return onMessage.mock.calls[0]?.[0]; } diff --git a/extensions/whatsapp/src/monitor-inbox.access-and-echo.test.ts b/extensions/whatsapp/src/monitor-inbox.access-and-echo.test.ts index 84171f448786..892d0d4751c0 100644 --- a/extensions/whatsapp/src/monitor-inbox.access-and-echo.test.ts +++ b/extensions/whatsapp/src/monitor-inbox.access-and-echo.test.ts @@ -4,7 +4,6 @@ import { describe, expect, it, vi } from "vitest"; import { isRecentOutboundMessage } from "./inbound/dedupe.js"; import { buildNotifyMessageUpsert, - expectPairingPromptSent, getRecordChannelActivityMock, installWebMonitorInboxUnitTestHooks, mockLoadConfig, @@ -12,6 +11,7 @@ import { startInboxMonitor, upsertPairingRequestMock, waitForMessageCalls, + waitForPairingPromptSent, } from "./monitor-inbox.test-harness.js"; const nowSeconds = (offsetMs = 0) => Math.floor((Date.now() + offsetMs) / 1000); @@ -227,14 +227,8 @@ describe("web monitor inbox", () => { }); sock.ev.emit("messages.upsert", upsertBlocked); - await vi.waitFor( - () => { - expect(sock.sendMessage).toHaveBeenCalledTimes(1); - }, - { timeout: 5_000, interval: 5 }, - ); + await waitForPairingPromptSent(sock, "999@s.whatsapp.net", "+999"); expect(onMessage).not.toHaveBeenCalled(); - expectPairingPromptSent(sock, "999@s.whatsapp.net", "+999"); const upsertBlockedAgain = buildNotifyMessageUpsert({ id: "no-config-1b", @@ -293,15 +287,9 @@ describe("web monitor inbox", () => { }); sock.ev.emit("messages.upsert", upsertBlocked); - await vi.waitFor( - () => { - expect(sock.sendMessage).toHaveBeenCalledTimes(1); - }, - { timeout: 5_000, interval: 5 }, - ); + await waitForPairingPromptSent(sock, "999@s.whatsapp.net", "+999"); expect(onMessage).not.toHaveBeenCalled(); - expectPairingPromptSent(sock, "999@s.whatsapp.net", "+999"); await listener.close(); }); diff --git a/extensions/whatsapp/src/monitor-inbox.media-and-session.test.ts b/extensions/whatsapp/src/monitor-inbox.media-and-session.test.ts index 942994eb7e60..bd62e0ecea2d 100644 --- a/extensions/whatsapp/src/monitor-inbox.media-and-session.test.ts +++ b/extensions/whatsapp/src/monitor-inbox.media-and-session.test.ts @@ -1,14 +1,12 @@ // WhatsApp monitor inbox media and session behavior. import { beforeEach, describe, expect, it, vi } from "vitest"; import { - DEFAULT_ACCOUNT_ID, - getAuthDir, - getMonitorWebInbox, getSock, installWebMonitorInboxUnitTestHooks, mockLoadConfig, + startInboxMonitor, + waitForInboundWorkDrained, } from "./monitor-inbox.test-harness.js"; -let monitorWebInbox: typeof import("./inbound.js").monitorWebInbox; const inboundLoggerInfoMock = vi.hoisted(() => vi.fn()); vi.mock("openclaw/plugin-sdk/logging-core", async () => { @@ -31,48 +29,19 @@ describe("web monitor inbox", () => { beforeEach(() => { inboundLoggerInfoMock.mockReset(); - monitorWebInbox = getMonitorWebInbox(); }); - async function openMonitor( - onMessage = vi.fn(), - extraOptions: Partial[0]> = {}, - ) { - return await monitorWebInbox({ - cfg: mockLoadConfig() as never, - verbose: false, - accountId: DEFAULT_ACCOUNT_ID, - authDir: getAuthDir(), - onMessage, - ...extraOptions, - }); + async function openMonitor(onMessage = vi.fn()) { + const { listener } = await startInboxMonitor(onMessage); + return listener; } async function runSingleUpsertAndCapture(upsert: unknown) { const onMessage = vi.fn(); - let armed = false; - let observedPendingWork = false; - let resolvePendingWorkDrained!: () => void; - const pendingWorkDrained = new Promise((resolve) => { - resolvePendingWorkDrained = resolve; - }); - const listener = await openMonitor(onMessage, { - onPendingWorkChanged: (pendingWorkCount) => { - if (!armed) { - return; - } - if (pendingWorkCount > 0) { - observedPendingWork = true; - } else if (observedPendingWork) { - resolvePendingWorkDrained(); - } - }, - }); - const sock = getSock(); - // The monitor owns async media and delivery work; wait for its drain signal instead of polling. - armed = true; + const { listener, sock } = await startInboxMonitor(onMessage); sock.ev.emit("messages.upsert", upsert); - await pendingWorkDrained; + // The monitor owns async media and delivery work; wait for its drain instead of polling. + await waitForInboundWorkDrained(); return { onMessage, listener, sock }; } diff --git a/extensions/whatsapp/src/monitor-inbox.policy.test.ts b/extensions/whatsapp/src/monitor-inbox.policy.test.ts index e0d1c12fd004..5690e0556206 100644 --- a/extensions/whatsapp/src/monitor-inbox.policy.test.ts +++ b/extensions/whatsapp/src/monitor-inbox.policy.test.ts @@ -2,15 +2,12 @@ import { describe, expect, it, vi } from "vitest"; import type { WebInboundMessage } from "./inbound/types.js"; import { - DEFAULT_ACCOUNT_ID, - expectPairingPromptSent, - getAuthDir, - getMonitorWebInbox, - getSock, installWebMonitorInboxUnitTestHooks, mockLoadConfig, - settleInboundWork, + startInboxMonitor, + waitForInboundWorkDrained, waitForMessageCalls, + waitForPairingPromptSent, } from "./monitor-inbox.test-harness.js"; const nowSeconds = (offsetMs = 0) => Math.floor((Date.now() + offsetMs) / 1000); @@ -59,28 +56,16 @@ async function startWebInboxMonitor(params: { loadConfig?: () => Record; sendReadReceipts?: boolean; }) { - const monitorWebInbox = getMonitorWebInbox(); if (params.config) { mockLoadConfig.mockReturnValue(params.config); } const onMessage = vi.fn(); - const base = { + const { listener, sock } = await startInboxMonitor(onMessage, { cfg: (params.config ?? mockLoadConfig()) as never, ...(params.loadConfig ? { loadConfig: params.loadConfig as never } : {}), - verbose: false, - accountId: DEFAULT_ACCOUNT_ID, - authDir: getAuthDir(), - onMessage, - }; - const listener = await monitorWebInbox( - params.sendReadReceipts === undefined - ? base - : { - ...base, - sendReadReceipts: params.sendReadReceipts, - }, - ); - return { onMessage, listener, sock: getSock() }; + ...(params.sendReadReceipts === undefined ? {} : { sendReadReceipts: params.sendReadReceipts }), + }); + return { onMessage, listener, sock }; } function firstInboundPayload(onMessage: ReturnType) { @@ -122,7 +107,7 @@ describe("web monitor inbox", () => { }), ), ); - await vi.waitFor(() => expectPairingPromptSent(sock, "999@s.whatsapp.net", "+999")); + await waitForPairingPromptSent(sock, "999@s.whatsapp.net", "+999"); // Should NOT call onMessage for unauthorized senders expect(onMessage).not.toHaveBeenCalled(); @@ -169,7 +154,7 @@ describe("web monitor inbox", () => { }), ), ); - await settleInboundWork(); + await waitForInboundWorkDrained(); expect(onMessage).not.toHaveBeenCalled(); expect(sock.sendMessage).not.toHaveBeenCalled(); @@ -367,7 +352,7 @@ describe("web monitor inbox", () => { }), ), ); - await settleInboundWork(); + await waitForInboundWorkDrained(); expect(onMessage).toHaveBeenCalledTimes(expectedCalls); if (expectedCalls === 1) { diff --git a/extensions/whatsapp/src/monitor-inbox.test-harness.ts b/extensions/whatsapp/src/monitor-inbox.test-harness.ts index bae205e9cf6c..11733a5b54f1 100644 --- a/extensions/whatsapp/src/monitor-inbox.test-harness.ts +++ b/extensions/whatsapp/src/monitor-inbox.test-harness.ts @@ -276,13 +276,10 @@ function expectInboxPairingReplyText( return resolvedCode; } -export function getMonitorWebInbox(): MonitorWebInbox { - if (!monitorWebInbox) { - throw new Error("monitorWebInbox not initialized"); - } - return monitorWebInbox; -} - +// Yields two macrotask ticks so already-scheduled inbound continuations run. +// This deliberately does NOT wait for pending inbound work to finish — tests +// observing intermediate states (held handlers, parked debounce batches) rely +// on that. For final-state assertions use waitForInboundWorkDrained(). export async function settleInboundWork() { await new Promise((resolve) => { setImmediate(resolve); @@ -292,6 +289,56 @@ export async function settleInboundWork() { }); } +type InboundWorkTracker = { pending: number }; +const inboundWorkTrackers = new Set(); +let inboundDrainWaiters: Array<() => void> = []; + +function releaseInboundDrainWaitersIfIdle() { + if (inboundDrainWaiters.length === 0) { + return; + } + for (const tracker of inboundWorkTrackers) { + if (tracker.pending > 0) { + return; + } + } + const waiters = inboundDrainWaiters; + inboundDrainWaiters = []; + for (const release of waiters) { + release(); + } +} + +function hasPendingInboundWork(): boolean { + for (const tracker of inboundWorkTrackers) { + if (tracker.pending > 0) { + return true; + } + } + return false; +} + +// Event-driven drain: resolves when every harness-started listener reports zero +// pending inbound work. Unlike a vi.waitFor deadline, this cannot fail spuriously +// when a saturated no-isolate worker stalls mid-flow (sync module fetches against +// the shared transform queue starve waitFor's interval timers). Do not call it +// while a handler or debounced batch is intentionally held open — it would wait +// for that work too; use settleInboundWork/waitForMessageCalls there. +export async function waitForInboundWorkDrained() { + if (inboundWorkTrackers.size === 0) { + throw new Error("waitForInboundWorkDrained requires a listener started via startInboxMonitor"); + } + if (hasPendingInboundWork()) { + await new Promise((resolve) => { + inboundDrainWaiters.push(resolve); + }); + } + // One extra tick lets drained-callback continuations (delivery bookkeeping) run. + await new Promise((resolve) => { + setImmediate(resolve); + }); +} + export function resetWebInboundDedupeForTests() { if (!resetWebInboundDedupe) { throw new Error("resetWebInboundDedupe not initialized"); @@ -316,13 +363,24 @@ export async function startInboxMonitor( if (!monitorWebInbox) { ({ monitorWebInbox } = await import("./inbound.js")); } - const listener = await monitorWebInbox({ + const merged = { cfg: mockLoadConfig() as never, verbose: false, onMessage, accountId: DEFAULT_ACCOUNT_ID, authDir: getAuthDir(), ...extraOptions, + }; + const tracker: InboundWorkTracker = { pending: 0 }; + inboundWorkTrackers.add(tracker); + const callerOnPendingWorkChanged = merged.onPendingWorkChanged; + const listener = await monitorWebInbox({ + ...merged, + onPendingWorkChanged: (pendingWorkCount: number, at?: number) => { + tracker.pending = pendingWorkCount; + releaseInboundDrainWaitersIfIdle(); + callerOnPendingWorkChanged?.(pendingWorkCount, at); + }, }); return { listener, sock: getSock() }; } @@ -353,7 +411,7 @@ export function buildNotifyMessageUpsert(params: { }; } -export function expectPairingPromptSent(sock: MockSock, jid: string, senderE164: string) { +function expectPairingPromptSent(sock: MockSock, jid: string, senderE164: string) { expect(sock.sendMessage).toHaveBeenCalledTimes(1); const sendCall = sock.sendMessage.mock.calls.at(0); expect(sendCall?.[0]).toBe(jid); @@ -364,6 +422,14 @@ export function expectPairingPromptSent(sock: MockSock, jid: string, senderE164: }); } +// The pairing reply is sent before the inbound handler completes, so a full +// drain guarantees the prompt is observable — and makes the callers' negative +// assertions (onMessage/readMessages never called) non-vacuous. +export async function waitForPairingPromptSent(sock: MockSock, jid: string, senderE164: string) { + await waitForInboundWorkDrained(); + expectPairingPromptSent(sock, jid, senderE164); +} + let authDir: string | undefined; export function getAuthDir(): string { @@ -379,6 +445,8 @@ export function installWebMonitorInboxUnitTestHooks(opts?: { authDir?: boolean } beforeEach(async () => { vi.useRealTimers(); vi.clearAllMocks(); + inboundWorkTrackers.clear(); + inboundDrainWaiters = []; channelActivityMocks.recordChannelActivity.mockClear(); pluginRuntimeMocks.reset(); setWhatsAppRuntime({