fix(qa-channel): bound stalled bus requests (#108684)

* fix(qa-channel): bound stalled bus requests

* test(qa-channel): cover stalled bus response bodies

* test(qa-channel): make stalled-body timeout deterministic

* style(qa-channel): format timeout regression

---------

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

extensions/qa-channel/src/bus-client.test.ts

@@ -7,6 +7,7 @@ import {
   parseQaTarget,
   pollQaBus,
   resolveQaTargetThread,
+  sendQaBusMessage,
 } from "./bus-client.js";
 
 const guardedFetchCalls = vi.hoisted(
@@ -159,6 +160,7 @@ describe("qa-bus client", () => {
   afterEach(async () => {
     await Promise.all(stops.splice(0).map((stop) => stop()));
     guardedFetchCalls.length = 0;
+    vi.restoreAllMocks();
   });
 
   it("roundtrips explicit group targets", () => {
@@ -293,6 +295,92 @@ describe("qa-bus client", () => {
     }
   });
 
+  it("bounds stalled message requests with a total deadline", async () => {
+    const server = createServer((_req, _res) => {
+      // Accept the request without returning headers so the client deadline owns the outcome.
+    });
+    const port = await listenLoopbackServer(server);
+    stops.push(async () => {
+      server.closeAllConnections?.();
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    });
+
+    const realAbortSignalTimeout = AbortSignal.timeout.bind(AbortSignal);
+    const timeoutSpy = vi
+      .spyOn(AbortSignal, "timeout")
+      .mockImplementationOnce(() => realAbortSignalTimeout(25));
+
+    await expect(
+      sendQaBusMessage({
+        baseUrl: `http://127.0.0.1:${port}`,
+        accountId: "acct-a",
+        to: "dm:alice",
+        text: "hello",
+      }),
+    ).rejects.toMatchObject({ name: "AbortError", cause: { name: "TimeoutError" } });
+    expect(timeoutSpy).toHaveBeenCalledTimes(1);
+    expect(timeoutSpy).toHaveBeenCalledWith(10_000);
+  });
+
+  it("bounds message responses that stall after headers", async () => {
+    let markBodyStarted: () => void = () => {};
+    const bodyStarted = new Promise<void>((resolve) => {
+      markBodyStarted = resolve;
+    });
+    const server = createServer((_req, res) => {
+      res.writeHead(200, {
+        "content-type": "application/json; charset=utf-8",
+        "content-length": "128",
+      });
+      res.write('{"message":', markBodyStarted);
+    });
+    const port = await listenLoopbackServer(server);
+    stops.push(async () => {
+      server.closeAllConnections?.();
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    });
+
+    const timeout = new AbortController();
+    const timeoutSpy = vi.spyOn(AbortSignal, "timeout").mockReturnValueOnce(timeout.signal);
+    const request = sendQaBusMessage({
+      baseUrl: `http://127.0.0.1:${port}`,
+      accountId: "acct-a",
+      to: "dm:alice",
+      text: "hello",
+    });
+    const rejection = expect(
+      withTimeout(request, 1_000, "stalled response body did not settle"),
+    ).rejects.toMatchObject({ name: "AbortError", cause: { name: "TimeoutError" } });
+
+    await withTimeout(bodyStarted, 500, "server did not start the response body");
+    timeout.abort(new DOMException("qa-bus request timed out", "TimeoutError"));
+    await rejection;
+    expect(timeoutSpy).toHaveBeenCalledTimes(1);
+    expect(timeoutSpy).toHaveBeenCalledWith(10_000);
+  });
+
+  it("keeps long polls within the server wait window plus response grace", async () => {
+    const server = await startJsonServer(() => ({
+      body: JSON.stringify({ cursor: 1, events: [] }),
+    }));
+    stops.push(server["stop"]);
+    const timeoutSpy = vi.spyOn(AbortSignal, "timeout");
+
+    await expect(
+      pollQaBus({
+        baseUrl: server.baseUrl,
+        accountId: "acct-a",
+        cursor: 0,
+        timeoutMs: 30_000,
+      }),
+    ).resolves.toEqual({ cursor: 1, events: [] });
+    expect(timeoutSpy).toHaveBeenCalledWith(40_000);
+  });
+
   it("preserves baseUrl path prefixes when composing bus URLs", async () => {
     const server = await startJsonServer((req) => ({
       statusCode: req.url === "/qa-bus/v1/state" ? 200 : 404,

extensions/qa-channel/src/bus-client.ts

@@ -1,6 +1,7 @@
 // Qa Channel plugin module implements bus client behavior.
 import http from "node:http";
 import https from "node:https";
+import { resolvePositiveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
 import { readProviderJsonResponse } from "openclaw/plugin-sdk/provider-http";
 import { parseQaTarget, type QaTargetParts } from "openclaw/plugin-sdk/qa-channel-protocol";
 import { readByteStreamWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
@@ -41,9 +42,16 @@ export type {
 
 type JsonResult<T> = Promise<T>;
 const QA_BUS_JSON_RESPONSE_MAX_BYTES = 16 * 1024 * 1024;
+/** Total deadline for local qa-bus POST requests and long-poll response grace. */
+const QA_BUS_REQUEST_TIMEOUT_MS = 10_000;
 /** Total deadline for local qa-bus state requests. */
 const QA_BUS_STATE_TIMEOUT_MS = 10_000;
 
+type QaBusPostOptions = {
+  signal?: AbortSignal;
+  timeoutMs?: number;
+};
+
 function buildQaBusUrl(baseUrl: string, path: string): URL {
   const normalizedBaseUrl = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/`;
   return new URL(path.replace(/^\/+/, ""), normalizedBaseUrl);
@@ -72,20 +80,16 @@ async function postJson<T>(
   baseUrl: string,
   path: string,
   body: unknown,
-  signal?: AbortSignal,
+  options: QaBusPostOptions = {},
 ): JsonResult<T> {
   const url = buildQaBusUrl(baseUrl, path);
   const payload = JSON.stringify(body);
   const client = url.protocol === "https:" ? https : http;
+  const timeoutMs = resolvePositiveTimerTimeoutMs(options.timeoutMs, QA_BUS_REQUEST_TIMEOUT_MS);
+  const timeoutSignal = AbortSignal.timeout(timeoutMs);
+  const signal = options.signal ? AbortSignal.any([options.signal, timeoutSignal]) : timeoutSignal;
 
   return await new Promise<T>((resolve, reject) => {
-    const abortError = () =>
-      Object.assign(new Error("The operation was aborted"), { name: "AbortError" });
-    if (signal?.aborted) {
-      reject(abortError());
-      return;
-    }
-
     const request = client.request(
       url,
       {
@@ -95,6 +99,7 @@ async function postJson<T>(
           "content-length": Buffer.byteLength(payload),
           connection: "close",
         },
+        signal,
       },
       (response) => {
         const label = `qa-bus ${path}`;
@@ -118,19 +123,7 @@ async function postJson<T>(
       },
     );
 
-    const onAbort = () => {
-      const error = abortError();
-      request.destroy(error);
-      reject(error);
-    };
-    signal?.addEventListener("abort", onAbort, { once: true });
-    request.on("error", (error) => {
-      signal?.removeEventListener("abort", onAbort);
-      reject(error);
-    });
-    request.on("close", () => {
-      signal?.removeEventListener("abort", onAbort);
-    });
+    request.on("error", reject);
     request.end(payload);
   });
 }
@@ -185,7 +178,10 @@ export async function pollQaBus(params: {
       cursor: params.cursor,
       timeoutMs: params.timeoutMs,
     },
-    params.signal,
+    {
+      signal: params.signal,
+      timeoutMs: params.timeoutMs + QA_BUS_REQUEST_TIMEOUT_MS,
+    },
   );
 }