fix(e2e): bound kitchen sink rpc probe bodies

scripts/e2e/kitchen-sink-rpc-walk.mjs

@@ -28,6 +28,10 @@ const INSTALL_TIMEOUT_MS = readPositiveInt(
 );
 const RPC_TIMEOUT_MS = readPositiveInt(process.env.OPENCLAW_KITCHEN_SINK_RPC_CALL_MS, 60000);
 const FETCH_TIMEOUT_MS = readPositiveInt(process.env.OPENCLAW_KITCHEN_SINK_RPC_FETCH_MS, 10000);
+const FETCH_BODY_MAX_BYTES = readPositiveInt(
+  process.env.OPENCLAW_KITCHEN_SINK_RPC_FETCH_BODY_BYTES,
+  1024 * 1024,
+);
 const MAX_RSS_MIB = readPositiveInt(process.env.OPENCLAW_KITCHEN_SINK_MAX_RSS_MIB, 2048);
 const GATEWAY_TEARDOWN_GRACE_MS = 10000;
 const GATEWAY_TEARDOWN_KILL_GRACE_MS = 2000;
@@ -461,6 +465,7 @@ function isRetryableTransientNetworkError(error, seen = new Set()) {
 export async function fetchJson(url, options = {}) {
   const attempts = Math.max(1, options.attempts ?? 3);
   const timeoutMs = Math.max(1, options.timeoutMs ?? FETCH_TIMEOUT_MS);
+  const maxBodyBytes = Math.max(1, options.maxBodyBytes ?? FETCH_BODY_MAX_BYTES);
   let lastError;
   for (let attempt = 1; attempt <= attempts; attempt += 1) {
     const controller = new AbortController();
@@ -480,7 +485,10 @@ export async function fetchJson(url, options = {}) {
         (options.fetchImpl ?? fetch)(url, { signal: controller.signal }),
         timeoutPromise,
       ]);
-      const text = await Promise.race([response.text(), timeoutPromise]);
+      const text = await Promise.race([
+        readBoundedResponseText(response, maxBodyBytes),
+        timeoutPromise,
+      ]);
       let body = null;
       try {
         body = text ? JSON.parse(text) : null;
@@ -503,6 +511,31 @@ export async function fetchJson(url, options = {}) {
   throw lastError ?? new Error(`fetch ${url} failed`);
 }
 
+export async function readBoundedResponseText(response, byteLimit = FETCH_BODY_MAX_BYTES) {
+  const reader = response.body?.getReader?.();
+  if (!reader) {
+    return await response.text();
+  }
+  const chunks = [];
+  let totalBytes = 0;
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) {
+      break;
+    }
+    const chunk = Buffer.from(value);
+    totalBytes += chunk.byteLength;
+    if (totalBytes > byteLimit) {
+      await reader.cancel().catch(() => undefined);
+      throw Object.assign(new Error(`fetch response body exceeded ${byteLimit} bytes`), {
+        code: "ETOOBIG",
+      });
+    }
+    chunks.push(chunk);
+  }
+  return Buffer.concat(chunks, totalBytes).toString("utf8");
+}
+
 function configureKitchenSink(env, port) {
   const configPath = env.OPENCLAW_CONFIG_PATH;
   const config = fs.existsSync(configPath) ? readJson(configPath) : {};

test/scripts/kitchen-sink-rpc-walk.test.ts

@@ -24,6 +24,7 @@ import {
   hasChildExited,
   makeEnv,
   readPositiveInt,
+  readBoundedResponseText,
   runCommand,
   sampleProcess,
   sampleWindowsProcessByPort,
@@ -620,6 +621,29 @@ describe("kitchen-sink RPC process sampling", () => {
     expect(fetchImpl).toHaveBeenCalledTimes(2);
   });
 
+  it("bounds HTTP probe response bodies", async () => {
+    const fetchImpl = vi
+      .fn()
+      .mockResolvedValue(new Response("x".repeat(1025), { status: 200 }));
+
+    await expect(
+      fetchJson("http://127.0.0.1:19680/healthz", {
+        attempts: 1,
+        fetchImpl,
+        maxBodyBytes: 1024,
+      }),
+    ).rejects.toMatchObject({
+      code: "ETOOBIG",
+      message: "fetch response body exceeded 1024 bytes",
+    });
+  });
+
+  it("reads bounded response streams", async () => {
+    await expect(
+      readBoundedResponseText(new Response('{"status":"live"}'), 1024),
+    ).resolves.toBe('{"status":"live"}');
+  });
+
   it("times out stalled HTTP probe response bodies", async () => {
     vi.useFakeTimers();
     const fetchImpl = vi.fn().mockResolvedValue({