test: stabilize load-sensitive test families that break gates under parallel load (#100420)

* test: stabilize process-heavy fixtures under load

* test: stabilize channel mocks under load

extensions/matrix/src/approval-reactions.test.ts

@@ -10,6 +10,16 @@ import {
 } from "./approval-reactions.js";
 import { setMatrixRuntime } from "./runtime.js";
 
+function createRuntimeLogger(overrides: { warn?: ReturnType<typeof vi.fn> } = {}) {
+  // Runtime state survives no-isolate workers, so expose every logger method later files may call.
+  return {
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: overrides.warn ?? vi.fn(),
+    error: vi.fn(),
+  };
+}
+
 afterEach(() => {
   clearMatrixApprovalReactionTargetsForTest();
   vi.restoreAllMocks();
@@ -124,7 +134,7 @@ describe("matrix approval reactions", () => {
     }));
     setMatrixRuntime({
       state: { openKeyedStore },
-      logging: { getChildLogger: () => ({ warn: vi.fn() }) },
+      logging: { getChildLogger: () => createRuntimeLogger() },
     } as never);
 
     registerMatrixApprovalReactionTarget({
@@ -165,7 +175,7 @@ describe("matrix approval reactions", () => {
           throw new Error("sqlite unavailable");
         }),
       },
-      logging: { getChildLogger: () => ({ warn }) },
+      logging: { getChildLogger: () => createRuntimeLogger({ warn }) },
     } as never);
 
     registerMatrixApprovalReactionTarget({

extensions/whatsapp/src/monitor-inbox.captures-media-path-image-messages.test-support.ts

@@ -8,8 +8,6 @@ import {
   getSock,
   installWebMonitorInboxUnitTestHooks,
   mockLoadConfig,
-  settleInboundWork,
-  waitForMessageCalls,
 } from "./monitor-inbox.test-harness.js";
 let monitorWebInbox: typeof import("./inbound.js").monitorWebInbox;
 const inboundLoggerInfoMock = vi.hoisted(() => vi.fn());
@@ -37,23 +35,45 @@ describe("web monitor inbox", () => {
     monitorWebInbox = getMonitorWebInbox();
   });
 
-  async function openMonitor(onMessage = vi.fn()) {
+  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 runSingleUpsertAndCapture(upsert: unknown) {
     const onMessage = vi.fn();
-    const listener = await openMonitor(onMessage);
+    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;
     sock.ev.emit("messages.upsert", upsert);
-    await waitForMessageCalls(onMessage, 1);
-    await settleInboundWork();
+    await pendingWorkDrained;
     return { onMessage, listener, sock };
   }
 

test/scripts/kitchen-sink-plugin-assertions.test.ts

@@ -16,6 +16,8 @@ import { describe, expect, it } from "vitest";
 const ASSERTIONS_SCRIPT = "scripts/e2e/lib/kitchen-sink-plugin/assertions.mjs";
 const BASH_BIN = process.platform === "win32" ? "bash" : "/bin/bash";
 const SWEEP_SCRIPT = "scripts/e2e/lib/kitchen-sink-plugin/sweep.sh";
+// The shim waits for an explicit log-ready marker; this only bounds a broken fixture process.
+const FIXTURE_READY_WAIT_ATTEMPTS = process.env.CI ? 2_000 : 1_000;
 const REQUIRED_FULL_DIAGNOSTIC_CANARIES = [
   "agent tool result middleware must be a function",
   "trusted tool policy registration requires id, description, and evaluate()",
@@ -843,6 +845,7 @@ exit "$status"
     const fixtureDir = path.join(scratchRoot, "clawhub-fixture");
     const nodeShim = path.join(fakeBin, "node");
     const sleepShim = path.join(fakeBin, "sleep");
+    const fixtureReadyPath = path.join(parent, "fixture-log-ready");
     try {
       mkdirSync(fakeBin, { recursive: true });
       mkdirSync(fixtureDir, { recursive: true });
@@ -853,6 +856,7 @@ exit "$status"
           "printf 'DO_NOT_DUMP_CLAWHUB_PREFIX\\n'",
           "head -c 2048 /dev/zero | tr '\\0' x",
           "printf '\\nFIXTURE_TAIL_MARKER\\n'",
+          ': >"$FIXTURE_READY_PATH"',
           "/bin/sleep 30",
           "",
         ].join("\n"),
@@ -862,8 +866,8 @@ exit "$status"
         sleepShim,
         [
           "#!/usr/bin/env bash",
-          "for _ in $(seq 1 50); do",
-          '  grep -q "FIXTURE_TAIL_MARKER" "$FIXTURE_DIR/clawhub-fixture.log" && exit 0',
+          'for _ in $(seq 1 "$FIXTURE_READY_WAIT_ATTEMPTS"); do',
+          '  [[ -f "$FIXTURE_READY_PATH" ]] && exit 0',
           "  /bin/sleep 0.01",
           "done",
           "exit 1",
@@ -891,6 +895,8 @@ exit "$status"
         {
           FAKE_BIN: fakeBin,
           FIXTURE_DIR: fixtureDir,
+          FIXTURE_READY_PATH: fixtureReadyPath,
+          FIXTURE_READY_WAIT_ATTEMPTS: String(FIXTURE_READY_WAIT_ATTEMPTS),
           SCRATCH_ROOT: scratchRoot,
         },
       );

test/scripts/run-vitest.test.ts

@@ -31,6 +31,8 @@ import {
 } from "../../scripts/run-vitest.mjs";
 
 const posixIt = process.platform === "win32" ? it.skip : it;
+// These bounds only guard broken fixtures; readiness and exit are asserted via process signals.
+const LOAD_SENSITIVE_PROCESS_TIMEOUT_MS = process.env.CI ? 30_000 : 15_000;
 
 describe("scripts/run-vitest", () => {
   it("adds --no-maglev to vitest child processes by default", () => {
@@ -659,8 +661,10 @@ describe("scripts/run-vitest", () => {
     let descendantPid = 0;
 
     try {
-      await waitFor(() => fs.existsSync(childPidPath), 10_000);
-      await waitFor(() => fs.existsSync(descendantPidPath), 10_000);
+      await waitFor(
+        () => fs.existsSync(childPidPath) && fs.existsSync(descendantPidPath),
+        LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
+      );
       childPid = Number(fs.readFileSync(childPidPath, "utf8"));
       descendantPid = Number(fs.readFileSync(descendantPidPath, "utf8"));
       expect(Number.isInteger(childPid)).toBe(true);
@@ -670,11 +674,11 @@ describe("scripts/run-vitest", () => {
 
       expect(runner.pid).toBeGreaterThan(0);
       process.kill(runner.pid!, "SIGTERM");
-      const result = await waitForClose(runner);
+      const result = await waitForClose(runner, LOAD_SENSITIVE_PROCESS_TIMEOUT_MS);
 
       expect(result).toEqual({ code: null, signal: "SIGTERM" });
-      await waitFor(() => !isProcessAlive(childPid), 5_000);
-      await waitFor(() => !isProcessAlive(descendantPid), 5_000);
+      await waitFor(() => !isProcessAlive(childPid), LOAD_SENSITIVE_PROCESS_TIMEOUT_MS);
+      await waitFor(() => !isProcessAlive(descendantPid), LOAD_SENSITIVE_PROCESS_TIMEOUT_MS);
     } finally {
       if (runner.pid && isProcessAlive(runner.pid)) {
         process.kill(runner.pid, "SIGKILL");

test/scripts/upgrade-survivor-probe-gateway.test.ts

@@ -5,12 +5,14 @@ import { createServer as createHttpServer } from "node:http";
 import { createServer as createTcpServer, type Server, type Socket } from "node:net";
 import os from "node:os";
 import path from "node:path";
+import { pathToFileURL } from "node:url";
 import { afterEach, describe, expect, it } from "vitest";
 import { createBoundedChildOutput } from "../helpers/bounded-child-output.js";
 
 const probePath = path.resolve("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs");
 const dockerSurvivorPath = path.resolve("scripts/e2e/upgrade-survivor-docker.sh");
 const tempDirs: string[] = [];
+const LOAD_SENSITIVE_PROCESS_TIMEOUT_MS = process.env.CI ? 30_000 : 15_000;
 
 function makeTempDir(): string {
   const dir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-upgrade-probe-"));
@@ -18,6 +20,12 @@ function makeTempDir(): string {
   return dir;
 }
 
+function writeProbeImport(source: string): string[] {
+  const fixturePath = path.join(makeTempDir(), "probe-import.mjs");
+  fs.writeFileSync(fixturePath, source);
+  return ["--import", pathToFileURL(fixturePath).href];
+}
+
 interface ProbeResult {
   error?: Error;
   signal: NodeJS.Signals | null;
@@ -28,11 +36,12 @@ interface ProbeResult {
 
 function runProbe(
   args: string[],
-  timeout = 5_000,
+  timeout = LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
   env: NodeJS.ProcessEnv = {},
+  nodeArgs: string[] = [],
 ): Promise<ProbeResult> {
   return new Promise((resolve) => {
-    const child = spawn(process.execPath, [probePath, ...args], {
+    const child = spawn(process.execPath, [...nodeArgs, probePath, ...args], {
       env: { ...process.env, ...env },
       stdio: ["ignore", "pipe", "pipe"],
     });
@@ -193,15 +202,26 @@ describe("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs", () => {
   });
 
   it("keeps failed probe retries inside the total timeout", async () => {
-    const server = createHttpServer((_request, response) => {
-      response.writeHead(503, { "content-type": "application/json" });
-      response.end(JSON.stringify({ ready: false, failing: ["gateway"] }));
-    });
-    const baseUrl = await listen(server);
+    const baseUrl = "http://probe.test";
     const out = path.join(makeTempDir(), "ready-timeout.json");
-    const startedAt = Date.now();
-    try {
-      const result = await runProbe([
+    const nodeArgs = writeProbeImport(
+      [
+        "const realSetTimeout = globalThis.setTimeout;",
+        "let now = 0;",
+        "Date.now = () => now;",
+        'globalThis.fetch = async () => new Response(JSON.stringify({ ready: false, failing: ["gateway"] }), { status: 503, headers: { "content-type": "application/json" } });',
+        "globalThis.setTimeout = (callback, delay = 0, ...args) => {",
+        "  if (delay === 50) {",
+        "    now += delay;",
+        "    return realSetTimeout(callback, 0, ...args);",
+        "  }",
+        "  return realSetTimeout(callback, delay, ...args);",
+        "};",
+      ].join("\n"),
+    );
+    // Virtualize fetch and only the retry sleep; CPU scheduling cannot consume the test budget.
+    const result = await runProbe(
+      [
         "--base-url",
         baseUrl,
         "--path",
@@ -214,26 +234,25 @@ describe("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs", () => {
         "50",
         "--attempt-timeout-ms",
         "25",
-      ]);
+      ],
+      LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
+      {},
+      nodeArgs,
+    );
 
-      expect(result.status).not.toBe(0);
-      expect(Date.now() - startedAt).toBeLessThan(300);
-      expect(result.stderr).toContain("probe did not satisfy ready within 50ms");
-      expect(fs.existsSync(out)).toBe(false);
-    } finally {
-      server.close();
-    }
+    expect(result.status).not.toBe(0);
+    expect(result.stderr).toContain("probe did not satisfy ready within 50ms");
+    expect(fs.existsSync(out)).toBe(false);
   });
 
   it("allows degraded ready responses only when degraded readiness is explicit", async () => {
-    const server = createHttpServer((_request, response) => {
-      response.writeHead(503, { "content-type": "application/json" });
-      response.end(JSON.stringify({ ready: false, failing: ["telegram"] }));
-    });
-    const baseUrl = await listen(server);
+    const baseUrl = "http://probe.test";
     const out = path.join(makeTempDir(), "ready-degraded.json");
-    try {
-      const result = await runProbe([
+    const nodeArgs = writeProbeImport(
+      'globalThis.fetch = async () => new Response(JSON.stringify({ ready: false, failing: ["telegram"] }), { status: 503, headers: { "content-type": "application/json" } });',
+    );
+    const result = await runProbe(
+      [
         "--base-url",
         baseUrl,
         "--path",
@@ -247,17 +266,18 @@ describe("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs", () => {
         out,
         "--timeout-ms",
         "300",
-      ]);
+      ],
+      LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
+      {},
+      nodeArgs,
+    );
 
-      expect(result.status).toBe(0);
-      expect(JSON.parse(fs.readFileSync(out, "utf8"))).toMatchObject({
-        body: { failing: ["telegram"], ready: false },
-        path: "/readyz",
-        status: 503,
-      });
-    } finally {
-      server.close();
-    }
+    expect(result.status).toBe(0);
+    expect(JSON.parse(fs.readFileSync(out, "utf8"))).toMatchObject({
+      body: { failing: ["telegram"], ready: false },
+      path: "/readyz",
+      status: 503,
+    });
   });
 
   it("does not let degraded ready mode convert generic server errors into success", async () => {
@@ -293,44 +313,35 @@ describe("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs", () => {
   });
 
   it("rejects declared oversized probe bodies before waiting on the stream", async () => {
-    const server = createHttpServer((_request, response) => {
-      response.writeHead(200, {
-        "content-length": "65",
-        "content-type": "application/json",
-      });
-      response.flushHeaders();
-    });
-    const baseUrl = await listen(server);
+    const baseUrl = "http://probe.test";
     const out = path.join(makeTempDir(), "oversized.json");
-    const startedAt = Date.now();
-    try {
-      const result = await runProbe(
-        [
-          "--base-url",
-          baseUrl,
-          "--path",
-          "/healthz",
-          "--expect",
-          "live",
-          "--out",
-          out,
-          "--timeout-ms",
-          "200",
-          "--attempt-timeout-ms",
-          "100",
-        ],
-        5_000,
-        { OPENCLAW_UPGRADE_SURVIVOR_PROBE_MAX_BODY_BYTES: "64" },
-      );
+    const nodeArgs = writeProbeImport(
+      'globalThis.fetch = async () => new Response(new ReadableStream({ start() {} }), { status: 200, headers: { "content-length": "65", "content-type": "application/json" } });',
+    );
+    const result = await runProbe(
+      [
+        "--base-url",
+        baseUrl,
+        "--path",
+        "/healthz",
+        "--expect",
+        "live",
+        "--out",
+        out,
+        "--timeout-ms",
+        "200",
+        "--attempt-timeout-ms",
+        "100",
+      ],
+      LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
+      { OPENCLAW_UPGRADE_SURVIVOR_PROBE_MAX_BODY_BYTES: "64" },
+      nodeArgs,
+    );
 
-      expect(result.error).toBeUndefined();
-      expect(result.status).not.toBe(0);
-      expect(result.stderr).toContain(`${baseUrl}/healthz probe body exceeded 64 bytes`);
-      expect(fs.existsSync(out)).toBe(false);
-      expect(Date.now() - startedAt).toBeLessThan(750);
-    } finally {
-      server.close();
-    }
+    expect(result.error).toBeUndefined();
+    expect(result.status).not.toBe(0);
+    expect(result.stderr).toContain(`${baseUrl}/healthz probe body exceeded 64 bytes`);
+    expect(fs.existsSync(out)).toBe(false);
   });
 
   it("bounds probes when a server accepts the connection but never responds", async () => {
@@ -417,14 +428,13 @@ describe("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs", () => {
   });
 
   it("caps response bodies before parsing probe JSON", async () => {
-    const server = createHttpServer((_request, response) => {
-      response.writeHead(200, { "content-type": "application/json" });
-      response.end("x".repeat(256));
-    });
-    const baseUrl = await listen(server);
+    const baseUrl = "http://probe.test";
     const out = path.join(makeTempDir(), "oversized.json");
-    try {
-      const result = await runProbe([
+    const nodeArgs = writeProbeImport(
+      'globalThis.fetch = async () => new Response("x".repeat(256), { status: 200, headers: { "content-type": "application/json" } });',
+    );
+    const result = await runProbe(
+      [
         "--base-url",
         baseUrl,
         "--path",
@@ -437,13 +447,14 @@ describe("scripts/e2e/lib/upgrade-survivor/probe-gateway.mjs", () => {
         "300",
         "--max-body-bytes",
         "64",
-      ]);
+      ],
+      LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
+      {},
+      nodeArgs,
+    );
 
-      expect(result.status).not.toBe(0);
-      expect(result.stderr).toContain("probe body exceeded 64 bytes");
-      expect(fs.existsSync(out)).toBe(false);
-    } finally {
-      server.close();
-    }
+    expect(result.status).not.toBe(0);
+    expect(result.stderr).toContain("probe body exceeded 64 bytes");
+    expect(fs.existsSync(out)).toBe(false);
   });
 });

test/scripts/write-cli-startup-metadata.test.ts

@@ -8,6 +8,9 @@ import { resolveWindowsTaskkillPath } from "../../scripts/lib/windows-taskkill.m
 import { __testing, writeCliStartupMetadata } from "../../scripts/write-cli-startup-metadata.ts";
 import { createScriptTestHarness } from "./test-helpers.js";
 
+// These subprocess tests use explicit ready/close signals; timeout only catches broken fixtures.
+const LOAD_SENSITIVE_PROCESS_TIMEOUT_MS = process.env.CI ? 30_000 : 15_000;
+
 function writeFixtureFile(rootDir: string, relativePath: string, contents: string): void {
   const filePath = path.join(rootDir, relativePath);
   mkdirSync(path.dirname(filePath), { recursive: true });
@@ -67,7 +70,10 @@ function expectedTaskkillPath(): string {
   return resolveWindowsTaskkillPath();
 }
 
-async function waitForProcessExit(pid: number, timeoutMs = 1_000): Promise<void> {
+async function waitForProcessExit(
+  pid: number,
+  timeoutMs = LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
+): Promise<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
     if (!processIsAlive(pid)) {
@@ -82,7 +88,7 @@ async function waitForProcessExit(pid: number, timeoutMs = 1_000): Promise<void>
 
 async function waitForChildClose(
   child: ReturnType<typeof spawn>,
-  timeoutMs = 2_000,
+  timeoutMs = LOAD_SENSITIVE_PROCESS_TIMEOUT_MS,
 ): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {
   return await new Promise((resolve, reject) => {
     const timeout = setTimeout(() => {
@@ -292,7 +298,7 @@ describe("write-cli-startup-metadata", () => {
       let grandchildPid = 0;
 
       try {
-        const deadline = Date.now() + 1_000;
+        const deadline = Date.now() + LOAD_SENSITIVE_PROCESS_TIMEOUT_MS;
         while (Date.now() < deadline) {
           try {
             grandchildPid = Number(readFileSync(grandchildPidPath, "utf8"));
@@ -318,7 +324,7 @@ describe("write-cli-startup-metadata", () => {
           code: null,
           signal: "SIGTERM",
         });
-        await waitForProcessExit(grandchildPid, 2_000);
+        await waitForProcessExit(grandchildPid);
       } finally {
         if (runner.pid && processIsAlive(runner.pid)) {
           runner.kill("SIGKILL");