fix(qa-lab): bound convex broker response bodies (#98619)

extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.test.ts

@@ -1,4 +1,5 @@
 // Qa Lab tests cover credential lease plugin behavior.
+import { createServer } from "node:http";
 import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import {
@@ -39,6 +40,74 @@ function fetchInit(fetchImpl: FetchMock, index = 0): RequestInit {
   return init;
 }
 
+async function startStreamingFailureBroker(params: {
+  chunkBytes?: number;
+  intervalMs?: number;
+  totalBytes?: number;
+}) {
+  const chunkBytes = params.chunkBytes ?? 64 * 1024;
+  const intervalMs = params.intervalMs ?? 1;
+  const totalBytes = params.totalBytes ?? 4 * 1024 * 1024;
+  let bytesWritten = 0;
+  let requestCount = 0;
+  let resolveClose: () => void = () => {};
+  const closePromise = new Promise<void>((resolve) => {
+    resolveClose = resolve;
+  });
+
+  const server = createServer((_req, res) => {
+    requestCount += 1;
+    res.writeHead(500, { "content-type": "text/plain" });
+    const interval = setInterval(() => {
+      if (bytesWritten >= totalBytes || res.destroyed) {
+        clearInterval(interval);
+        if (!res.destroyed) {
+          res.end();
+        }
+        return;
+      }
+      const nextBytes = Math.min(chunkBytes, totalBytes - bytesWritten);
+      bytesWritten += nextBytes;
+      res.write("x".repeat(nextBytes));
+    }, intervalMs);
+    res.on("close", () => {
+      clearInterval(interval);
+      resolveClose();
+    });
+  });
+
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", () => {
+      server.off("error", reject);
+      resolve();
+    });
+  });
+
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("expected streaming broker address");
+  }
+  return {
+    closePromise,
+    getBytesWritten: () => bytesWritten,
+    getRequestCount: () => requestCount,
+    totalBytes,
+    url: `http://127.0.0.1:${address.port}`,
+    stop: async () => {
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => {
+          if (error) {
+            reject(error);
+            return;
+          }
+          resolve();
+        });
+      });
+    },
+  };
+}
+
 describe("credential lease runtime", () => {
   afterEach(() => {
     vi.restoreAllMocks();
@@ -107,6 +176,60 @@ describe("credential lease runtime", () => {
     expect(headers.authorization).toBe("Bearer maintainer-secret");
   });
 
+  it("bounds oversized convex broker failure bodies before parsing", async () => {
+    const fetchImpl = vi.fn<typeof fetch>().mockResolvedValueOnce(
+      new Response("x".repeat(1_048_577), {
+        status: 500,
+        headers: { "content-type": "text/plain" },
+      }),
+    );
+
+    await expect(
+      acquireQaCredentialLease({
+        kind: "telegram",
+        source: "convex",
+        role: "maintainer",
+        env: {
+          OPENCLAW_QA_CONVEX_SITE_URL: "https://qa-cred.example.convex.site",
+          OPENCLAW_QA_CONVEX_SECRET_MAINTAINER: "maintainer-secret",
+        },
+        fetchImpl,
+        resolveEnvPayload: () => ({ groupId: "-1", driverToken: "unused", sutToken: "unused" }),
+        parsePayload: (payload) =>
+          payload as { groupId: string; driverToken: string; sutToken: string },
+      }),
+    ).rejects.toThrow("Convex credential broker: text response exceeds 1048576 bytes");
+
+    expect(fetchImpl).toHaveBeenCalledTimes(1);
+  });
+
+  it("cancels a streaming convex broker failure body after the response cap", async () => {
+    const broker = await startStreamingFailureBroker({});
+    try {
+      await expect(
+        acquireQaCredentialLease({
+          kind: "telegram",
+          source: "convex",
+          role: "maintainer",
+          env: {
+            OPENCLAW_QA_CONVEX_SITE_URL: broker.url,
+            OPENCLAW_QA_CONVEX_SECRET_MAINTAINER: "maintainer-secret",
+            OPENCLAW_QA_ALLOW_INSECURE_HTTP: "1",
+          },
+          resolveEnvPayload: () => ({ groupId: "-1", driverToken: "unused", sutToken: "unused" }),
+          parsePayload: (payload) =>
+            payload as { groupId: string; driverToken: string; sutToken: string },
+        }),
+      ).rejects.toThrow("Convex credential broker: text response exceeds 1048576 bytes");
+
+      await broker.closePromise;
+      expect(broker.getRequestCount()).toBe(1);
+      expect(broker.getBytesWritten()).toBeLessThan(broker.totalBytes);
+    } finally {
+      await broker.stop();
+    }
+  });
+
   it("hydrates chunked convex credential payloads after acquire", async () => {
     const serialized = JSON.stringify({
       groupId: "-100123",

extensions/qa-lab/src/live-transports/shared/credential-lease.runtime.ts

@@ -2,6 +2,7 @@
 import { randomUUID } from "node:crypto";
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
 import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
+import { readProviderTextResponse } from "openclaw/plugin-sdk/provider-http";
 import { z } from "zod";
 import {
   isQaCredentialTruthyOptIn,
@@ -19,6 +20,7 @@ const DEFAULT_HTTP_TIMEOUT_MS = 15_000;
 const DEFAULT_LEASE_TTL_MS = 20 * 60 * 1_000;
 const DEFAULT_CHUNKED_PAYLOAD_MAX_BYTES = 64 * 1024 * 1024;
 const DEFAULT_CHUNKED_PAYLOAD_MAX_CHUNKS = 4096;
+const CONVEX_BROKER_RESPONSE_MAX_BYTES = 1 * 1024 * 1024;
 const CHUNKED_PAYLOAD_MAX_BYTES_ENV = "OPENCLAW_QA_CREDENTIAL_PAYLOAD_MAX_BYTES";
 const CHUNKED_PAYLOAD_MAX_CHUNKS_ENV = "OPENCLAW_QA_CREDENTIAL_PAYLOAD_MAX_CHUNKS";
 const RETRY_BACKOFF_MS = [500, 1_000, 2_000, 4_000, 5_000] as const;
@@ -282,6 +284,7 @@ async function postConvexBroker(params: {
   authToken: string;
   body: Record<string, unknown>;
   fetchImpl: typeof fetch;
+  maxBytes: number;
   timeoutMs: number;
   url: string;
 }): Promise<unknown> {
@@ -296,7 +299,11 @@ async function postConvexBroker(params: {
     signal: AbortSignal.timeout(timeoutMs),
   });
 
-  const text = await response.text();
+  // Keep ordinary broker responses small, while allowing chunk payloads to use
+  // the larger declared payload ceiling.
+  const text = await readProviderTextResponse(response, "Convex credential broker", {
+    maxBytes: params.maxBytes,
+  });
   const payload: unknown = (() => {
     if (!text.trim()) {
       return undefined;
@@ -341,6 +348,7 @@ async function resolveConvexCredentialPayload(params: {
   for (let index = 0; index < marker.chunkCount; index += 1) {
     const payload = await postConvexBroker({
       fetchImpl: params.fetchImpl,
+      maxBytes: params.config.payloadMaxBytes,
       timeoutMs: params.config.httpTimeoutMs,
       authToken: params.config.authToken,
       url: params.config.payloadChunkUrl,
@@ -437,6 +445,7 @@ export async function acquireQaCredentialLease<TPayload>(
     try {
       const payload = await postConvexBroker({
         fetchImpl,
+        maxBytes: CONVEX_BROKER_RESPONSE_MAX_BYTES,
         timeoutMs: config.httpTimeoutMs,
         authToken: config.authToken,
         url: config.acquireUrl,
@@ -452,6 +461,7 @@ export async function acquireQaCredentialLease<TPayload>(
       const releaseLease = async () => {
         const releasePayload = await postConvexBroker({
           fetchImpl,
+          maxBytes: CONVEX_BROKER_RESPONSE_MAX_BYTES,
           timeoutMs: config.httpTimeoutMs,
           authToken: config.authToken,
           url: config.releaseUrl,
@@ -503,6 +513,7 @@ export async function acquireQaCredentialLease<TPayload>(
         async heartbeat() {
           const heartbeatPayload = await postConvexBroker({
             fetchImpl,
+            maxBytes: CONVEX_BROKER_RESPONSE_MAX_BYTES,
             timeoutMs: config.httpTimeoutMs,
             authToken: config.authToken,
             url: config.heartbeatUrl,