fix(telegram): bound stalled diagnostic response reads (#106035)

* fix(telegram): bound stalled diagnostic response reads

* test(telegram): type timeout cancellation mock

* style(telegram): apply oxfmt to diagnostic response timeout files

* fix(telegram): share diagnostic body deadlines

Co-authored-by: Alix-007 <li.long15@xydigit.com>

* chore: leave release notes release-owned

---------

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

docs/.generated/plugin-sdk-api-baseline.sha256

@@ -1,2 +1,2 @@
-0b4e6f1bc9aee2cb849a8645911c96430e0ab212847b5172f93100f34e5d6dbc  plugin-sdk-api-baseline.json
-577a7870668561c977bbe95892db8eeea330a62c6740e5da09ffdd7660dd3f96  plugin-sdk-api-baseline.jsonl
+b54488723b29d94400cf65abba859bab80cd3b1e3bf9f7104956242152470c39  plugin-sdk-api-baseline.json
+d96f25f59d60b3171bc139da39eed20dd5b46d271db0875bf002a546699d2ccb  plugin-sdk-api-baseline.jsonl

docs/plugins/sdk-subpaths.md

@@ -332,7 +332,7 @@ usage endpoint failed or returned no usable usage data.
     | `plugin-sdk/fetch-runtime` | Wrapped fetch, proxy, EnvHttpProxyAgent option, and pinned lookup helpers |
     | `plugin-sdk/runtime-fetch` | Dispatcher-aware runtime fetch without proxy/guarded-fetch imports |
     | `plugin-sdk/inline-image-data-url-runtime` | Inline image data URL sanitizer and signature sniffing helpers without the broad media runtime surface |
-    | `plugin-sdk/response-limit-runtime` | Bounded response-body reader without the broad media runtime surface |
+    | `plugin-sdk/response-limit-runtime` | Byte-, idle-, and deadline-bounded response-body readers without the broad media runtime surface |
     | `plugin-sdk/session-binding-runtime` | Current conversation binding state without configured binding routing or pairing stores |
     | `plugin-sdk/context-visibility-runtime` | Context visibility resolution and supplemental context filtering without broad config/security imports |
     | `plugin-sdk/string-coerce-runtime` | Narrow primitive record/string coercion and normalization helpers without markdown/logging imports |

extensions/telegram/src/audit-membership-runtime.ts

@@ -1,5 +1,6 @@
 // Telegram plugin module implements audit membership runtime behavior.
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
+import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
 import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
 import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
 import { fetchWithTimeout } from "openclaw/plugin-sdk/text-utility-runtime";
@@ -18,6 +19,20 @@ type TelegramGroupMembershipAuditData = Omit<TelegramGroupMembershipAudit, "elap
 const TELEGRAM_BOT_API_MAX_RESPONSE_BYTES = 4 * 1024 * 1024;
 type TelegramChatMemberResult = { status?: string };
 
+async function readTelegramMembershipAuditBody(
+  response: Response,
+  timeoutMs: number,
+): Promise<Buffer> {
+  return await readResponseWithLimit(response, TELEGRAM_BOT_API_MAX_RESPONSE_BYTES, {
+    timeoutMs,
+    chunkTimeoutMs: timeoutMs / 2,
+    onIdleTimeout: ({ chunkTimeoutMs }) =>
+      new Error(`Telegram membership audit response body stalled for ${chunkTimeoutMs}ms`),
+    onTimeout: ({ timeoutMs: resolvedTimeoutMs }) =>
+      new Error(`Telegram membership audit response body timed out after ${resolvedTimeoutMs}ms`),
+  });
+}
+
 export async function auditTelegramGroupMembershipImpl(
   params: AuditTelegramGroupMembershipParams,
 ): Promise<TelegramGroupMembershipAuditData> {
@@ -28,13 +43,29 @@ export async function auditTelegramGroupMembershipImpl(
   const apiBase = resolveTelegramApiBase(params.apiRoot);
   const base = `${apiBase}/bot${params.token}`;
   const groups: TelegramGroupMembershipAuditEntry[] = [];
+  const timeoutMs = resolveTimerTimeoutMs(params.timeoutMs, 1);
+  const deadlineMs = Date.now() + timeoutMs;
 
   for (const chatId of params.groupIds) {
+    const requestTimeoutMs = Math.max(0, deadlineMs - Date.now());
+    if (requestTimeoutMs === 0) {
+      groups.push({
+        chatId,
+        ok: false,
+        status: null,
+        error: `Telegram membership audit timed out after ${timeoutMs}ms`,
+        matchKey: chatId,
+        matchSource: "id",
+      });
+      continue;
+    }
     try {
       const url = `${base}/getChatMember?chat_id=${encodeURIComponent(chatId)}&user_id=${encodeURIComponent(String(params.botId))}`;
-      const res = await fetchWithTimeout(url, {}, params.timeoutMs, fetcher);
+      const res = await fetchWithTimeout(url, {}, requestTimeoutMs, fetcher);
       const json = JSON.parse(
-        (await readResponseWithLimit(res, TELEGRAM_BOT_API_MAX_RESPONSE_BYTES)).toString("utf8"),
+        (await readTelegramMembershipAuditBody(res, Math.max(1, deadlineMs - Date.now()))).toString(
+          "utf8",
+        ),
       ) as TelegramApiOk<TelegramChatMemberResult> | TelegramApiErr;
       if (!res.ok || !isRecord(json) || !json.ok) {
         const desc =

extensions/telegram/src/audit.test.ts

@@ -41,6 +41,21 @@ async function auditSingleGroup() {
   });
 }
 
+function makeStallingJsonResponse(payload: unknown, cancel: (reason?: unknown) => void): Response {
+  return new Response(
+    new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(JSON.stringify(payload)));
+      },
+      cancel,
+    }),
+    {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    },
+  );
+}
+
 describe("telegram audit", () => {
   beforeAll(async () => {
     vi.doMock("./fetch.js", () => ({
@@ -86,4 +101,79 @@ describe("telegram audit", () => {
     expect(res.groups[0]?.ok).toBe(false);
     expect(res.groups[0]?.status).toBe("left");
   });
+
+  it("reports stalled getChatMember response bodies quickly", async () => {
+    const cancel = vi.fn();
+    fetchWithTimeoutMock.mockResolvedValueOnce(
+      makeStallingJsonResponse({ ok: true, result: { status: "member" } }, cancel),
+    );
+
+    vi.useFakeTimers();
+    try {
+      const auditPromise = auditTelegramGroupMembership({
+        token: "t",
+        botId: 123,
+        groupIds: ["-1001"],
+        timeoutMs: 50,
+      });
+      await vi.advanceTimersByTimeAsync(0);
+      await vi.advanceTimersByTimeAsync(60);
+
+      const res = await auditPromise;
+      expect(res.ok).toBe(false);
+      expect(res.groups[0]?.ok).toBe(false);
+      expect(res.groups[0]?.error).toBe("Telegram membership audit response body stalled for 25ms");
+      expect(cancel).toHaveBeenCalledTimes(1);
+      expect(cancel.mock.calls[0]?.[0]).toBeInstanceOf(Error);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("shares one timeout budget across sequential membership checks", async () => {
+    const cancel = vi.fn();
+    fetchWithTimeoutMock
+      .mockImplementationOnce(
+        async () =>
+          await new Promise<Response>((resolve) => {
+            setTimeout(() => {
+              resolve(
+                new Response(JSON.stringify({ ok: true, result: { status: "member" } }), {
+                  status: 200,
+                }),
+              );
+            }, 40);
+          }),
+      )
+      .mockResolvedValueOnce(
+        makeStallingJsonResponse({ ok: true, result: { status: "member" } }, cancel),
+      );
+
+    vi.useFakeTimers();
+    try {
+      const auditPromise = auditTelegramGroupMembership({
+        token: "t",
+        botId: 123,
+        groupIds: ["-1001", "-1002"],
+        timeoutMs: 50,
+      });
+      await vi.advanceTimersByTimeAsync(60);
+
+      const result = await auditPromise;
+      expect(result.groups[0]?.ok).toBe(true);
+      expect(result.groups[1]?.error).toBe(
+        "Telegram membership audit response body stalled for 5ms",
+      );
+      expect(fetchWithTimeoutMock).toHaveBeenNthCalledWith(
+        2,
+        expect.stringContaining("chat_id=-1002"),
+        {},
+        10,
+        fetchWithTimeoutMock,
+      );
+      expect(cancel).toHaveBeenCalledTimes(1);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
 });

extensions/telegram/src/probe.response-body-timeout.integration.test.ts

@@ -0,0 +1,104 @@
+// Real-socket proof for diagnostic response deadlines: production Telegram
+// transport against a local HTTP server, with no fetch or stream mocks.
+import { createServer, type Server } from "node:http";
+import type { AddressInfo, Socket } from "node:net";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { probeTelegram, resetTelegramProbeFetcherCacheForTests } from "./probe.js";
+
+type ResponseMode = "stall" | "trickle";
+
+describe("probeTelegram response body deadlines over real sockets", () => {
+  let server: Server;
+  let apiRoot: string;
+  let responseMode: ResponseMode = "stall";
+  let requestCount = 0;
+  let closedSocketCount = 0;
+  const liveSockets = new Set<Socket>();
+  const activeIntervals = new Set<ReturnType<typeof setInterval>>();
+
+  beforeAll(async () => {
+    for (const name of [
+      "HTTP_PROXY",
+      "HTTPS_PROXY",
+      "ALL_PROXY",
+      "http_proxy",
+      "https_proxy",
+      "all_proxy",
+      "OPENCLAW_PROXY_URL",
+      "OPENCLAW_DEBUG_PROXY_ENABLED",
+      "OPENCLAW_DEBUG_PROXY_URL",
+    ]) {
+      vi.stubEnv(name, "");
+    }
+
+    server = createServer((_req, res) => {
+      requestCount += 1;
+      res.writeHead(200, { "content-type": "application/json" });
+      res.write('{"ok":true,"result":{"id":123');
+      if (responseMode === "stall") {
+        return;
+      }
+
+      const interval = setInterval(() => {
+        res.write(" ");
+      }, 20);
+      activeIntervals.add(interval);
+      res.once("close", () => {
+        clearInterval(interval);
+        activeIntervals.delete(interval);
+      });
+    });
+    server.on("connection", (socket) => {
+      liveSockets.add(socket);
+      socket.once("close", () => {
+        liveSockets.delete(socket);
+        closedSocketCount += 1;
+      });
+    });
+    await new Promise<void>((resolve) => {
+      server.listen(0, "127.0.0.1", resolve);
+    });
+    apiRoot = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+  });
+
+  afterAll(async () => {
+    resetTelegramProbeFetcherCacheForTests();
+    vi.unstubAllEnvs();
+    for (const interval of activeIntervals) {
+      clearInterval(interval);
+    }
+    for (const socket of liveSockets) {
+      socket.destroy();
+    }
+    await new Promise<void>((resolve) => {
+      server.close(() => resolve());
+    });
+  });
+
+  async function expectDeadlineFailure(mode: ResponseMode, expectedError: RegExp) {
+    responseMode = mode;
+    const previousRequestCount = requestCount;
+    const previousClosedSocketCount = closedSocketCount;
+
+    const result = await probeTelegram("placeholder", 200, {
+      apiRoot,
+      includeWebhookInfo: false,
+    });
+
+    expect(result.ok).toBe(false);
+    expect(result.error).toMatch(expectedError);
+    expect(result.elapsedMs).toBeLessThan(1_000);
+    expect(requestCount).toBe(previousRequestCount + 1);
+    await vi.waitFor(() => expect(closedSocketCount).toBeGreaterThan(previousClosedSocketCount), {
+      timeout: 1_000,
+    });
+  }
+
+  it("cancels a socket whose response body stalls", async () => {
+    await expectDeadlineFailure("stall", /response body stalled/i);
+  });
+
+  it("enforces the overall deadline while body bytes keep arriving", async () => {
+    await expectDeadlineFailure("trickle", /response body timed out/i);
+  });
+});

extensions/telegram/src/probe.response-body-timeout.test.ts

@@ -0,0 +1,168 @@
+// Telegram tests cover stalled diagnostic response body handling.
+import { afterEach, describe, expect, it, vi, type Mock } from "vitest";
+import { probeTelegram, resetTelegramProbeFetcherCacheForTests } from "./probe.js";
+
+const resolveTelegramTransport = vi.hoisted(() => vi.fn());
+const makeProxyFetch = vi.hoisted(() => vi.fn());
+
+vi.mock("./fetch.js", () => ({
+  resolveTelegramTransport,
+  resolveTelegramApiBase: (apiRoot?: string) =>
+    apiRoot?.trim()?.replace(/\/+$/, "") || "https://api.telegram.org",
+}));
+
+vi.mock("./proxy.js", () => ({
+  makeProxyFetch,
+}));
+
+function installFetchMock(): Mock {
+  const fetchMock = vi.fn();
+  resolveTelegramTransport.mockImplementation((proxyFetch?: typeof fetch) => ({
+    fetch: proxyFetch ?? (fetchMock as unknown as typeof fetch),
+    sourceFetch: proxyFetch ?? (fetchMock as unknown as typeof fetch),
+    forceFallback: vi.fn().mockReturnValue(true),
+    close: async () => {},
+  }));
+  makeProxyFetch.mockImplementation(() => fetchMock as unknown as typeof fetch);
+  return fetchMock;
+}
+
+function makeJsonResponse(payload: unknown): Response {
+  return new Response(JSON.stringify(payload), {
+    status: 200,
+    headers: { "content-type": "application/json" },
+  });
+}
+
+function makeStallingJsonResponse(payload: unknown, cancel: (reason?: unknown) => void): Response {
+  return new Response(
+    new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(JSON.stringify(payload)));
+      },
+      cancel,
+    }),
+    {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    },
+  );
+}
+
+function makeTricklingJsonResponse(cancel: (reason?: unknown) => void): Response {
+  const encoder = new TextEncoder();
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let cancelled = false;
+  const response = new Response(
+    new ReadableStream<Uint8Array>({
+      start(controller) {
+        let index = 0;
+        const chunks = ["{", '"ok"', ":", "true", ","];
+        const enqueue = () => {
+          if (cancelled) {
+            return;
+          }
+          controller.enqueue(encoder.encode(chunks[index % chunks.length]));
+          index += 1;
+          timer = setTimeout(enqueue, 40);
+        };
+        enqueue();
+      },
+      cancel(reason) {
+        cancelled = true;
+        if (timer) {
+          clearTimeout(timer);
+        }
+        cancel(reason);
+      },
+    }),
+    {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    },
+  );
+  return response;
+}
+
+describe("probeTelegram response body timeouts", () => {
+  afterEach(() => {
+    resetTelegramProbeFetcherCacheForTests();
+    resolveTelegramTransport.mockReset();
+    makeProxyFetch.mockReset();
+    vi.useRealTimers();
+    vi.clearAllMocks();
+  });
+
+  it("fails quickly when getMe returns a stalled response body", async () => {
+    const fetchMock = installFetchMock();
+    const cancel = vi.fn();
+    fetchMock.mockResolvedValueOnce(
+      makeStallingJsonResponse(
+        {
+          ok: true,
+          result: { id: 123, is_bot: true, first_name: "Test", username: "bot" },
+        },
+        cancel,
+      ),
+    );
+
+    vi.useFakeTimers();
+    const probePromise = probeTelegram("placeholder", 50, {
+      includeWebhookInfo: false,
+    });
+    await vi.advanceTimersByTimeAsync(0);
+    await vi.advanceTimersByTimeAsync(60);
+
+    const result = await probePromise;
+    expect(result.ok).toBe(false);
+    expect(result.error).toBe("Telegram diagnostic response body stalled for 25ms");
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(cancel).toHaveBeenCalledTimes(1);
+    expect(cancel.mock.calls[0]?.[0]).toBeInstanceOf(Error);
+  });
+
+  it("fails on the body deadline when getMe trickles without becoming complete JSON", async () => {
+    const fetchMock = installFetchMock();
+    const cancel = vi.fn();
+    vi.useFakeTimers();
+    fetchMock.mockResolvedValueOnce(makeTricklingJsonResponse(cancel));
+
+    const probePromise = probeTelegram("placeholder", 100, {
+      includeWebhookInfo: false,
+    });
+    await vi.advanceTimersByTimeAsync(0);
+    await vi.advanceTimersByTimeAsync(120);
+
+    const result = await probePromise;
+    expect(result.ok).toBe(false);
+    expect(result.error).toBe("Telegram diagnostic response body timed out after 100ms");
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(cancel).toHaveBeenCalledTimes(1);
+    expect(cancel.mock.calls[0]?.[0]).toBeInstanceOf(Error);
+  });
+
+  it("keeps webhook diagnostics best-effort when webhookInfo response body stalls", async () => {
+    const fetchMock = installFetchMock();
+    const cancel = vi.fn();
+    fetchMock.mockResolvedValueOnce(
+      makeJsonResponse({
+        ok: true,
+        result: { id: 123, is_bot: true, first_name: "Test", username: "bot" },
+      }),
+    );
+    fetchMock.mockResolvedValueOnce(
+      makeStallingJsonResponse({ ok: true, result: { url: "https://example.test/hook" } }, cancel),
+    );
+
+    vi.useFakeTimers();
+    const probePromise = probeTelegram("placeholder", 50);
+    await vi.advanceTimersByTimeAsync(0);
+    await vi.advanceTimersByTimeAsync(60);
+
+    const result = await probePromise;
+    expect(result.ok).toBe(true);
+    expect(result.webhook).toBeUndefined();
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(cancel).toHaveBeenCalledTimes(1);
+  });
+});

extensions/telegram/src/probe.ts

@@ -121,6 +121,17 @@ function normalizeBoolean(value: unknown): boolean | null {
   return typeof value === "boolean" ? value : null;
 }
 
+async function readTelegramDiagnosticBody(response: Response, timeoutMs: number): Promise<Buffer> {
+  return await readResponseWithLimit(response, TELEGRAM_BOT_API_MAX_RESPONSE_BYTES, {
+    timeoutMs,
+    chunkTimeoutMs: timeoutMs / 2,
+    onIdleTimeout: ({ chunkTimeoutMs }) =>
+      new Error(`Telegram diagnostic response body stalled for ${chunkTimeoutMs}ms`),
+    onTimeout: ({ timeoutMs: resolvedTimeoutMs }) =>
+      new Error(`Telegram diagnostic response body timed out after ${resolvedTimeoutMs}ms`),
+  });
+}
+
 export async function probeTelegram(
   token: string,
   timeoutMs: number,
@@ -193,7 +204,12 @@ export async function probeTelegram(
     }
 
     const meJson = JSON.parse(
-      (await readResponseWithLimit(meRes, TELEGRAM_BOT_API_MAX_RESPONSE_BYTES)).toString("utf8"),
+      (
+        await readTelegramDiagnosticBody(
+          meRes,
+          Math.min(timeoutBudgetMs, resolveRemainingBudgetMs()),
+        )
+      ).toString("utf8"),
     ) as {
       ok?: boolean;
       description?: string;
@@ -238,9 +254,12 @@ export async function probeTelegram(
             fetcher,
           );
           const webhookJson = JSON.parse(
-            (await readResponseWithLimit(webhookRes, TELEGRAM_BOT_API_MAX_RESPONSE_BYTES)).toString(
-              "utf8",
-            ),
+            (
+              await readTelegramDiagnosticBody(
+                webhookRes,
+                Math.min(timeoutBudgetMs, resolveRemainingBudgetMs()),
+              )
+            ).toString("utf8"),
           ) as {
             ok?: boolean;
             result?: { url?: string; has_custom_certificate?: boolean };

src/infra/http-body.response.test.ts

@@ -34,6 +34,30 @@ function makeStallingStream(initialChunks: Uint8Array[], onCancel?: (reason?: un
   });
 }
 
+function makeTricklingStream(intervalMs: number, onCancel?: (reason?: unknown) => void) {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let cancelled = false;
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      const enqueue = () => {
+        if (cancelled) {
+          return;
+        }
+        controller.enqueue(new Uint8Array([1]));
+        timer = setTimeout(enqueue, intervalMs);
+      };
+      enqueue();
+    },
+    cancel(reason) {
+      cancelled = true;
+      if (timer) {
+        clearTimeout(timer);
+      }
+      onCancel?.(reason);
+    },
+  });
+}
+
 async function expectIdleTimeout(
   createReadPromise: () => Promise<unknown>,
   expectedError: RegExp | string = /stalled/i,
@@ -217,6 +241,75 @@ describe("readResponseWithLimit", () => {
       vi.useRealTimers();
     }
   });
+
+  it("cancels a trickling body when its overall timeout expires", async () => {
+    vi.useFakeTimers();
+    try {
+      const cancel = vi.fn();
+      const response = new Response(makeTricklingStream(40, cancel));
+      const assertion = expect(
+        readResponseWithLimit(response, 1024, {
+          chunkTimeoutMs: 50,
+          timeoutMs: 100,
+          onTimeout: ({ timeoutMs }) => new Error(`custom overall ${timeoutMs}`),
+        }),
+      ).rejects.toThrow("custom overall 100");
+
+      await vi.advanceTimersByTimeAsync(110);
+      await assertion;
+      expect(cancel).toHaveBeenCalledTimes(1);
+      expect(cancel.mock.calls[0]?.[0]).toBeInstanceOf(Error);
+      expect((cancel.mock.calls[0]?.[0] as Error | undefined)?.message).toBe("custom overall 100");
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("cancels a getReader-less body when its overall timeout expires", async () => {
+    vi.useFakeTimers();
+    try {
+      const cancel = vi.fn(async (_reason?: unknown) => undefined);
+      const response = {
+        body: { cancel },
+        arrayBuffer: async () => await new Promise<ArrayBuffer>(() => {}),
+      } as unknown as Response;
+      const assertion = expect(
+        readResponseWithLimit(response, 1024, {
+          timeoutMs: 50,
+          onTimeout: ({ timeoutMs }) => new Error(`fallback overall ${timeoutMs}`),
+        }),
+      ).rejects.toThrow("fallback overall 50");
+
+      await vi.advanceTimersByTimeAsync(50);
+      await assertion;
+      expect(cancel).toHaveBeenCalledTimes(1);
+      expect(cancel.mock.calls[0]?.[0]).toBeInstanceOf(Error);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("clears the overall timeout after a successful read", async () => {
+    vi.useFakeTimers();
+    try {
+      const cancel = vi.fn();
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new Uint8Array([1, 2]));
+          controller.close();
+        },
+        cancel,
+      });
+
+      await expect(
+        readResponseWithLimit(new Response(body), 100, { timeoutMs: 50 }),
+      ).resolves.toEqual(Buffer.from([1, 2]));
+      await vi.advanceTimersByTimeAsync(100);
+      expect(cancel).not.toHaveBeenCalled();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
 });
 
 describe("readResponseTextSnippet", () => {

src/infra/http-body.ts

@@ -2,12 +2,14 @@
 import type { IncomingMessage, ServerResponse } from "node:http";
 import { clearTimeout as clearNodeTimeout, setTimeout as setNodeTimeout } from "node:timers";
 import { decodeTextPrefix } from "@openclaw/normalization-core";
-import { toErrorObject } from "@openclaw/normalization-core/error-coercion";
 import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion";
 import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
 import { formatErrorMessage } from "./errors.js";
+import { readChunkWithIdleTimeout, withResponseBodyTimeout } from "./http-response-body-timeout.js";
 import { parseStrictNonNegativeInteger } from "./parse-finite-number.js";
 
+export { readChunkWithIdleTimeout } from "./http-response-body-timeout.js";
+
 export const DEFAULT_WEBHOOK_MAX_BODY_BYTES = 1024 * 1024;
 export const DEFAULT_WEBHOOK_BODY_TIMEOUT_MS = 30_000;
 
@@ -129,53 +131,6 @@ function advanceRequestBodyChunk(
   };
 }
 
-/** Reads one chunk, rejecting and cancelling the reader after an idle timeout. */
-export async function readChunkWithIdleTimeout(
-  reader: ReadableStreamDefaultReader<Uint8Array>,
-  chunkTimeoutMs: number,
-  onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error,
-): Promise<Awaited<ReturnType<typeof reader.read>>> {
-  let timeoutId: ReturnType<typeof setTimeout> | undefined;
-  let timedOut = false;
-
-  return await new Promise((resolve, reject) => {
-    const clear = () => {
-      if (timeoutId !== undefined) {
-        clearTimeout(timeoutId);
-        timeoutId = undefined;
-      }
-    };
-
-    const resolvedChunkTimeoutMs = resolveTimerTimeoutMs(chunkTimeoutMs, 1);
-    timeoutId = setTimeout(() => {
-      timedOut = true;
-      const error =
-        onIdleTimeout?.({ chunkTimeoutMs: resolvedChunkTimeoutMs }) ??
-        new Error(`Media download stalled: no data received for ${resolvedChunkTimeoutMs}ms`);
-      clear();
-      // Cancel with the timeout error so fetch-backed streams release sockets
-      // and buffers instead of continuing after the caller has failed.
-      void reader.cancel(error).catch(() => undefined);
-      reject(error);
-    }, resolvedChunkTimeoutMs);
-
-    void reader.read().then(
-      (result) => {
-        clear();
-        if (!timedOut) {
-          resolve(result);
-        }
-      },
-      (error: unknown) => {
-        clear();
-        if (!timedOut) {
-          reject(toErrorObject(error, "Non-Error rejection"));
-        }
-      },
-    );
-  });
-}
-
 type ReadResponsePrefixResult = {
   buffer: Buffer;
   size: number;
@@ -185,6 +140,8 @@ type ReadResponsePrefixResult = {
 export type ReadResponseTextPrefixOptions = {
   chunkTimeoutMs?: number;
   onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error;
+  timeoutMs?: number;
+  onTimeout?: (params: { timeoutMs: number }) => Error;
 };
 
 type ReadResponsePrefixOptions = ReadResponseTextPrefixOptions & {
@@ -197,26 +154,11 @@ function validateMaxBytes(maxBytes: number): void {
   }
 }
 
-async function readResponsePrefix(
-  response: Response,
+async function readResponsePrefixFromReader(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
   maxBytes: number,
   options?: ReadResponsePrefixOptions,
 ): Promise<ReadResponsePrefixResult> {
-  validateMaxBytes(maxBytes);
-  const body = response.body;
-  if (!body || typeof body.getReader !== "function") {
-    const fallback = Buffer.from(await response.arrayBuffer());
-    if (fallback.length > maxBytes) {
-      return {
-        buffer: fallback.subarray(0, maxBytes),
-        size: fallback.length,
-        truncated: true,
-      };
-    }
-    return { buffer: fallback, size: fallback.length, truncated: false };
-  }
-
-  const reader = body.getReader();
   const chunks: Uint8Array[] = [];
   let total = 0;
   let size = 0;
@@ -267,6 +209,41 @@ async function readResponsePrefix(
   };
 }
 
+async function readResponsePrefix(
+  response: Response,
+  maxBytes: number,
+  options?: ReadResponsePrefixOptions,
+): Promise<ReadResponsePrefixResult> {
+  validateMaxBytes(maxBytes);
+  const body = response.body;
+  if (!body || typeof body.getReader !== "function") {
+    return await withResponseBodyTimeout({
+      timeoutMs: options?.timeoutMs,
+      onTimeout: options?.onTimeout,
+      cancel: async (error) => await body?.cancel(error),
+      read: async () => {
+        const fallback = Buffer.from(await response.arrayBuffer());
+        if (fallback.length > maxBytes) {
+          return {
+            buffer: fallback.subarray(0, maxBytes),
+            size: fallback.length,
+            truncated: true,
+          };
+        }
+        return { buffer: fallback, size: fallback.length, truncated: false };
+      },
+    });
+  }
+
+  const reader = body.getReader();
+  return await withResponseBodyTimeout({
+    timeoutMs: options?.timeoutMs,
+    onTimeout: options?.onTimeout,
+    cancel: async (error) => await reader.cancel(error),
+    read: async () => await readResponsePrefixFromReader(reader, maxBytes, options),
+  });
+}
+
 export type ReadResponseTextPrefixResult = {
   text: string;
   size: number;
@@ -290,14 +267,12 @@ export async function readResponseTextPrefix(
   };
 }
 
-/** Reads a response body under a byte cap, cancelling the stream on overflow or idle timeout. */
+/** Reads a response body under byte, idle, and overall timeout bounds. */
 export async function readResponseWithLimit(
   response: Response,
   maxBytes: number,
-  options?: {
+  options?: ReadResponseTextPrefixOptions & {
     onOverflow?: (params: { size: number; maxBytes: number; res: Response }) => Error;
-    chunkTimeoutMs?: number;
-    onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error;
   },
 ): Promise<Buffer> {
   const onOverflow =
@@ -307,6 +282,8 @@ export async function readResponseWithLimit(
   const prefix = await readResponsePrefix(response, maxBytes, {
     chunkTimeoutMs: options?.chunkTimeoutMs,
     onIdleTimeout: options?.onIdleTimeout,
+    timeoutMs: options?.timeoutMs,
+    onTimeout: options?.onTimeout,
   });
   if (prefix.truncated) {
     throw onOverflow({ size: prefix.size, maxBytes, res: response });
@@ -317,11 +294,9 @@ export async function readResponseWithLimit(
 /** Reads a small collapsed text prefix from a response body for diagnostics/errors. */
 export async function readResponseTextSnippet(
   response: Response,
-  options?: {
+  options?: ReadResponseTextPrefixOptions & {
     maxBytes?: number;
     maxChars?: number;
-    chunkTimeoutMs?: number;
-    onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error;
   },
 ): Promise<string | undefined> {
   const maxBytes = options?.maxBytes ?? 8 * 1024;
@@ -329,6 +304,8 @@ export async function readResponseTextSnippet(
   const prefix = await readResponseTextPrefix(response, maxBytes, {
     chunkTimeoutMs: options?.chunkTimeoutMs,
     onIdleTimeout: options?.onIdleTimeout,
+    timeoutMs: options?.timeoutMs,
+    onTimeout: options?.onTimeout,
   });
   if (!prefix.text) {
     return undefined;

src/infra/http-response-body-timeout.ts

@@ -0,0 +1,92 @@
+// Applies idle and overall deadlines to fetch response-body reads.
+import { toErrorObject } from "@openclaw/normalization-core/error-coercion";
+import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion";
+
+type TimeoutErrorFactory = (params: { timeoutMs: number }) => Error;
+
+async function withCancellableTimeout<T>(params: {
+  timeoutMs: number;
+  onTimeout: TimeoutErrorFactory;
+  cancel: (error: Error) => Promise<unknown>;
+  read: () => Promise<T>;
+}): Promise<T> {
+  const timeoutMs = resolveTimerTimeoutMs(params.timeoutMs, 1);
+  let timeoutId: ReturnType<typeof setTimeout> | undefined;
+  let timedOut = false;
+
+  return await new Promise<T>((resolve, reject) => {
+    const clear = () => {
+      if (timeoutId !== undefined) {
+        clearTimeout(timeoutId);
+        timeoutId = undefined;
+      }
+    };
+
+    timeoutId = setTimeout(() => {
+      timedOut = true;
+      const error = params.onTimeout({ timeoutMs });
+      clear();
+      void params.cancel(error).catch(() => undefined);
+      reject(error);
+    }, timeoutMs);
+    if (typeof timeoutId === "object" && "unref" in timeoutId) {
+      timeoutId.unref();
+    }
+
+    void Promise.resolve()
+      .then(params.read)
+      .then(
+        (value) => {
+          clear();
+          if (!timedOut) {
+            resolve(value);
+          }
+        },
+        (error: unknown) => {
+          clear();
+          if (!timedOut) {
+            reject(toErrorObject(error, "Non-Error rejection"));
+          }
+        },
+      );
+  });
+}
+
+/** Reads one chunk, rejecting and cancelling the reader after an idle timeout. */
+export async function readChunkWithIdleTimeout(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  chunkTimeoutMs: number,
+  onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error,
+): Promise<Awaited<ReturnType<typeof reader.read>>> {
+  return await withCancellableTimeout({
+    timeoutMs: chunkTimeoutMs,
+    onTimeout: ({ timeoutMs }) =>
+      onIdleTimeout?.({ chunkTimeoutMs: timeoutMs }) ??
+      new Error(`Media download stalled: no data received for ${timeoutMs}ms`),
+    // Cancellation releases fetch sockets and buffers instead of letting the
+    // pending read continue after the caller has failed.
+    cancel: async (error) => await reader.cancel(error),
+    read: async () => await reader.read(),
+  });
+}
+
+export async function withResponseBodyTimeout<T>(params: {
+  timeoutMs: number | undefined;
+  onTimeout: TimeoutErrorFactory | undefined;
+  cancel: (error: Error) => Promise<unknown>;
+  read: () => Promise<T>;
+}): Promise<T> {
+  if (params.timeoutMs === undefined) {
+    return await params.read();
+  }
+  return await withCancellableTimeout({
+    timeoutMs: params.timeoutMs,
+    onTimeout: ({ timeoutMs }) =>
+      params.onTimeout?.({ timeoutMs }) ??
+      new Error(`Response body timed out after ${timeoutMs}ms`),
+    // Fetch resolves at headers. Body cancellation owns socket cleanup when
+    // the separate whole-body deadline wins.
+    cancel: params.cancel,
+    read: params.read,
+  });
+}