test(whatsapp): fix order-dependent pairing-prompt flake with event-driven inbound drain (#131109)

* test(whatsapp): give pairing-prompt waits the saturation budget

The monitor-inbox.policy pairing test waited on vi.waitFor's 1s default
while sibling pairing tests use the suite-standard 5s/5ms saturation
budget. Under full-project no-isolate runs the worker can stall ~1.5s
mid-flow (sync module fetches against the saturated shared transform
queue), starving waitFor's interval timers so the prompt lands just
after the deadline. Probes showed every mock firing correctly on the
right sock — not the shared-worker mock-defeat class.

Centralize waitForPairingPromptSent in the monitor-inbox harness, use
it at all three call sites (dropping the duplicated inline waits), and
align inbound.media's 2s delivery wait to the same 5s budget.

* test(whatsapp): event-driven inbound drain for pairing waits

Replace the wall-clock pairing wait with waitForInboundWorkDrained: the
harness now tracks each listener's onPendingWorkChanged signal and
resolves when pending inbound work returns to zero, so no saturation
stall can outlast a deadline. settleInboundWork keeps its yield-ticks
semantics for tests that observe intermediate states (held handlers,
parked debounce batches).

Route the policy and media-and-session monitor helpers through the
harness startInboxMonitor so their listeners are drain-tracked, collapse
media-and-session's bespoke pending-work machinery into the shared
helper, upgrade the policy negative assertions to drain-backed (non-
vacuous), and drop the now-unused getMonitorWebInbox export.

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<typeof vi.fn>) {
+  // 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];
 }

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();
   });

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<Parameters<typeof monitorWebInbox>[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<void>((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 };
   }
 

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<string, unknown>;
   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<typeof vi.fn>) {
@@ -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) {

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<InboundWorkTracker>();
+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<void>((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({