fix(streams): preserve stable results when body cleanup fails (#111245)

* fix(proxy): preserve validation after cleanup failure

* fix(msteams): preserve attachment cleanup outcomes

* fix(gateway): preserve loopback size diagnostics

extensions/msteams/src/attachments/bot-framework.test.ts

@@ -535,6 +535,88 @@ describe("downloadMSTeamsBotFrameworkAttachment", () => {
       ]);
     });
 
+    it.each([
+      {
+        name: "attachment info is unavailable",
+        stage: "info",
+        init: { status: 500 },
+        warning: "msteams botFramework attachmentInfo non-ok",
+      },
+      {
+        name: "attachment view is unavailable",
+        stage: "view",
+        init: { status: 500 },
+        warning: "msteams botFramework attachmentView non-ok",
+      },
+      {
+        name: "attachment view content-length is invalid",
+        stage: "view",
+        init: { status: 200, headers: { "content-length": "0x3" } },
+        warning: "msteams botFramework attachmentView invalid content-length",
+      },
+      {
+        name: "attachment view exceeds maxBytes",
+        stage: "view",
+        init: { status: 200, headers: { "content-length": "11" } },
+      },
+    ] as const)("preserves the stable outcome when $name cleanup rejects", async (scenario) => {
+      const unhandledRejections: unknown[] = [];
+      const onUnhandledRejection = (reason: unknown) => {
+        unhandledRejections.push(reason);
+      };
+      const cancel = vi.fn(() => {
+        throw new Error("discarded Teams response cancellation failed");
+      });
+      const body = new ReadableStream<Uint8Array>({ cancel });
+      const discardedResponse = new Response(body, scenario.init);
+      const warn = vi.fn();
+      const fetchFn: typeof fetch = vi.fn(async (input: RequestInfo | URL) => {
+        const url =
+          typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url;
+        if (scenario.stage === "info" || url.endsWith("/views/original")) {
+          return discardedResponse;
+        }
+        return new Response(
+          JSON.stringify({
+            name: "doc.pdf",
+            type: "application/pdf",
+            views: [{ viewId: "original", size: 1 }],
+          }),
+          { status: 200 },
+        );
+      });
+      process.on("unhandledRejection", onUnhandledRejection);
+
+      try {
+        const media = await downloadMSTeamsBotFrameworkAttachment({
+          serviceUrl: "https://smba.trafficmanager.net/amer",
+          attachmentId: "att-1",
+          tokenProvider: buildTokenProvider(),
+          maxBytes: 10,
+          fetchFn,
+          fetchFnSupportsDispatcher: true,
+          resolveFn: resolvePublicHost,
+          logger: { warn },
+        });
+
+        expect(media).toBeUndefined();
+        expect(runtime.saveCalls).toHaveLength(0);
+        expect(cancel).toHaveBeenCalledOnce();
+        if (scenario.warning) {
+          expect(warn).toHaveBeenCalledWith(scenario.warning, expect.any(Object));
+        } else {
+          expect(warn).not.toHaveBeenCalled();
+        }
+        await new Promise<void>((resolve) => {
+          setImmediate(resolve);
+        });
+        expect(unhandledRejections).toStrictEqual([]);
+      } finally {
+        process.off("unhandledRejection", onUnhandledRejection);
+        expect(process.listeners("unhandledRejection")).not.toContain(onUnhandledRejection);
+      }
+    });
+
     it("bounds an unbounded attachmentInfo JSON body and cancels the stream", async () => {
       const state = { canceled: false, enqueued: 0 };
       const chunkBytes = 1024 * 1024;

extensions/msteams/src/attachments/bot-framework.ts

@@ -113,7 +113,7 @@ async function fetchBotFrameworkAttachmentInfo(params: {
     return undefined;
   }
   if (!response.ok) {
-    await response.body?.cancel();
+    await response.body?.cancel().catch(() => undefined);
     params.logger?.warn?.("msteams botFramework attachmentInfo non-ok", {
       status: response.status,
     });
@@ -173,7 +173,7 @@ async function saveBotFrameworkAttachmentView(params: {
     return undefined;
   }
   if (!response.ok) {
-    await response.body?.cancel();
+    await response.body?.cancel().catch(() => undefined);
     params.logger?.warn?.("msteams botFramework attachmentView non-ok", {
       status: response.status,
     });
@@ -183,14 +183,14 @@ async function saveBotFrameworkAttachmentView(params: {
   try {
     contentLength = parseMediaContentLength(response.headers.get("content-length"));
   } catch (err) {
-    await response.body?.cancel();
+    await response.body?.cancel().catch(() => undefined);
     params.logger?.warn?.("msteams botFramework attachmentView invalid content-length", {
       error: err instanceof Error ? err.message : String(err),
     });
     return undefined;
   }
   if (contentLength !== null && contentLength > params.maxBytes) {
-    await response.body?.cancel();
+    await response.body?.cancel().catch(() => undefined);
     return undefined;
   }
   try {

src/gateway/gateway-cli-backend.live-probe-helpers.test.ts

@@ -68,6 +68,7 @@ describe("gateway CLI backend live probe helpers", () => {
 
   afterEach(() => {
     vi.useRealTimers();
+    vi.unstubAllGlobals();
     clearActiveMcpLoopbackRuntimeByOwnerToken(ownerToken);
   });
 
@@ -148,4 +149,61 @@ describe("gateway CLI backend live probe helpers", () => {
       server.close();
     }
   });
+
+  it("preserves the byte-limit diagnostic when body cancellation rejects", async () => {
+    const unhandledRejections: unknown[] = [];
+    const onUnhandledRejection = (reason: unknown) => {
+      unhandledRejections.push(reason);
+    };
+    const cancel = vi.fn(() => {
+      throw new Error("loopback body cancellation failed");
+    });
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new Uint8Array(65));
+      },
+      cancel,
+    });
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async () => new Response(body)),
+    );
+    activateLoopbackRuntime(12345);
+    process.on("unhandledRejection", onUnhandledRejection);
+
+    try {
+      await expect(
+        verifyCliCronMcpLoopbackPreflight(
+          preflightParams({ OPENCLAW_MCP_LOOPBACK_PROBE_MAX_BODY_BYTES: "64" }),
+        ),
+      ).rejects.toThrow("mcp loopback response body exceeded 64 bytes");
+      expect(cancel).toHaveBeenCalledOnce();
+      expect(body.locked).toBe(false);
+      await new Promise<void>((resolve) => {
+        setImmediate(resolve);
+      });
+      expect(unhandledRejections).toStrictEqual([]);
+    } finally {
+      process.off("unhandledRejection", onUnhandledRejection);
+      expect(process.listeners("unhandledRejection")).not.toContain(onUnhandledRejection);
+    }
+  });
+
+  it("still propagates loopback response read errors and releases the reader lock", async () => {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.error(new Error("loopback response read failed"));
+      },
+    });
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async () => new Response(body)),
+    );
+    activateLoopbackRuntime(12345);
+
+    await expect(verifyCliCronMcpLoopbackPreflight(preflightParams())).rejects.toThrow(
+      "loopback response read failed",
+    );
+    expect(body.locked).toBe(false);
+  });
 });

src/gateway/gateway-cli-backend.live-probe-helpers.ts

@@ -4,6 +4,7 @@ import { randomUUID } from "node:crypto";
 import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
 import { renderCatFacePngBase64 } from "../../test/helpers/live-image-probe.js";
 import { isTruthyEnvValue } from "../infra/env.js";
+import { readResponseWithLimit } from "../infra/http-body.js";
 import { parseStrictPositiveInteger } from "../infra/parse-finite-number.js";
 import { sleep } from "../utils/sleep.js";
 import type { GatewayClient } from "./client.js";
@@ -221,7 +222,10 @@ async function callLoopbackJsonRpc(params: {
       body: JSON.stringify(params.body),
       signal: controller.signal,
     });
-    text = await readBoundedResponseText(response, maxBodyBytes);
+    const responseBody = await readResponseWithLimit(response, maxBodyBytes, {
+      onOverflow: () => new Error(`mcp loopback response body exceeded ${maxBodyBytes} bytes`),
+    });
+    text = responseBody.toString("utf8");
   } finally {
     clearTimeout(timer);
   }
@@ -241,28 +245,6 @@ async function callLoopbackJsonRpc(params: {
   return parsed;
 }
 
-async function readBoundedResponseText(response: Response, byteLimit: number): Promise<string> {
-  const reader = response.body?.getReader();
-  if (!reader) {
-    return "";
-  }
-  const chunks: Buffer[] = [];
-  let totalBytes = 0;
-  for (;;) {
-    const { done, value } = await reader.read();
-    if (done) {
-      break;
-    }
-    totalBytes += value.byteLength;
-    if (totalBytes > byteLimit) {
-      await reader.cancel();
-      throw new Error(`mcp loopback response body exceeded ${byteLimit} bytes`);
-    }
-    chunks.push(Buffer.from(value));
-  }
-  return Buffer.concat(chunks, totalBytes).toString("utf8");
-}
-
 export async function verifyCliCronMcpLoopbackPreflight(params: {
   sessionKey: string;
   port: number;

src/infra/net/proxy/proxy-validation.test.ts

@@ -4,12 +4,18 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import os from "node:os";
 import path from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
+import { fetchWithRuntimeDispatcher } from "../runtime-fetch.js";
+import { createHttp1ProxyAgent } from "../undici-runtime.js";
 import { runProxyValidation } from "./proxy-validation.js";
 
+vi.mock("../runtime-fetch.js", () => ({ fetchWithRuntimeDispatcher: vi.fn() }));
+vi.mock("../undici-runtime.js", () => ({ createHttp1ProxyAgent: vi.fn() }));
+
 describe("proxy validation", () => {
   const tempDirs: string[] = [];
 
   afterEach(() => {
+    vi.clearAllMocks();
     for (const dir of tempDirs.splice(0)) {
       rmSync(dir, { recursive: true, force: true });
     }
@@ -91,6 +97,56 @@ describe("proxy validation", () => {
     expect(fetchCheck).toHaveBeenCalled();
   });
 
+  it("preserves the validated response when discarded body cancellation rejects", async () => {
+    const unhandledRejections: unknown[] = [];
+    const onUnhandledRejection = (reason: unknown) => {
+      unhandledRejections.push(reason);
+    };
+    const cancel = vi.fn(() => {
+      throw new Error("proxy response cancellation failed");
+    });
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("validated"));
+      },
+      cancel,
+    });
+    const close = vi.fn(async () => undefined);
+    vi.mocked(createHttp1ProxyAgent).mockReturnValue({ close } as unknown as ReturnType<
+      typeof createHttp1ProxyAgent
+    >);
+    vi.mocked(fetchWithRuntimeDispatcher).mockResolvedValue(
+      new Response(body, {
+        status: 200,
+        headers: { "x-proxy-result": "validated" },
+      }),
+    );
+    process.on("unhandledRejection", onUnhandledRejection);
+
+    try {
+      const result = await runProxyValidation({
+        proxyUrlOverride: "http://proxy.example:3128",
+        allowedUrls: ["https://example.com/"],
+        deniedUrls: [],
+      });
+
+      expect(result).toMatchObject({
+        ok: true,
+        checks: [{ kind: "allowed", ok: true, status: 200 }],
+      });
+      expect(cancel).toHaveBeenCalledOnce();
+      expect(close).toHaveBeenCalledOnce();
+      expect(body.locked).toBe(false);
+      await new Promise<void>((resolve) => {
+        setImmediate(resolve);
+      });
+      expect(unhandledRejections).toStrictEqual([]);
+    } finally {
+      process.off("unhandledRejection", onUnhandledRejection);
+      expect(process.listeners("unhandledRejection")).not.toContain(onUnhandledRejection);
+    }
+  });
+
   it("prefers the configured proxy URL over OPENCLAW_PROXY_URL", async () => {
     const fetchCheck = vi.fn().mockResolvedValue({ ok: true, status: 200 });
 

src/infra/net/proxy/proxy-validation.ts

@@ -230,7 +230,7 @@ async function defaultProxyValidationFetchCheck({
       dispatcher,
       redirect: "manual",
     });
-    void response.body?.cancel();
+    void response.body?.cancel().catch(() => undefined);
     return {
       ok: response.ok,
       status: response.status,