fix(telegram): recover when media downloads stall (#103020)

* fix(telegram): add request timeout to media file downloads * fix(telegram): bound inbound media startup Co-authored-by: llagy009 <0668001470@xydigit.com> Co-authored-by: Alix-007 <li.long15@xydigit.com> * docs(changelog): note Telegram media recovery * fix(net): normalize preflight abort errors * test(media): accept nullable request signals * fix(telegram): abort webhook media on shutdown * fix(telegram): own webhook media cancellation * fix(telegram): retry aborted media groups * fix(telegram): cancel isolated media work * fix(telegram): stop draining ended polling cycles * fix(telegram): separate cycle media cancellation * fix(telegram): preserve live polling media fallbacks * chore: move release note to PR body --------- Co-authored-by: llagy009 <0668001470@xydigit.com> Co-authored-by: Peter Steinberger <steipete@gmail.com>
2026-07-20 20:21:40 +00:00 · 2026-07-10 21:31:16 +08:00
parent 53058123fc
commit 26aef8c48b
17 changed files with 1145 additions and 60 deletions
--- a/extensions/telegram/src/bot-handlers.runtime.ts
+++ b/extensions/telegram/src/bot-handlers.runtime.ts
@@ -233,9 +233,13 @@ export const registerTelegramHandlers = ({
    token: opts.token,
    transport: telegramTransport,
  });
+  const mediaAbortSignal =
+    opts.mediaAbortSignal && opts.fetchAbortSignal
+      ? AbortSignal.any([opts.mediaAbortSignal, opts.fetchAbortSignal])
+      : (opts.mediaAbortSignal ?? opts.fetchAbortSignal);
  const mediaRuntimeWithAbort = {
    ...mediaRuntimeOptions,
-    abortSignal: opts.fetchAbortSignal,
+    abortSignal: mediaAbortSignal,
  };
  const DEFAULT_TEXT_FRAGMENT_MAX_GAP_MS = 1500;
  const TELEGRAM_TEXT_FRAGMENT_START_THRESHOLD_CHARS = 4000;
@@ -670,6 +674,7 @@ export const registerTelegramHandlers = ({
              ...spooledReplayOptions(spooledReplayParticipants),
            },
            dispatchDedupeKeys: last.dispatchDedupeKeys,
+            durableMediaReplay: spooledReplayParticipants.length > 0,
          });
          settleSpooledReplayParticipants(spooledReplayParticipants, result);
          return;
@@ -719,6 +724,7 @@ export const registerTelegramHandlers = ({
          dispatchDedupeKeys: mergeDispatchDedupeKeys(
            ...entries.map((entry) => entry.dispatchDedupeKeys),
          ),
+          durableMediaReplay: spooledReplayParticipants.length > 0,
        });
        settleSpooledReplayParticipants(spooledReplayParticipants, result);
      } catch (err) {
@@ -1080,6 +1086,14 @@ export const registerTelegramHandlers = ({
            ...mediaRuntimeWithAbort,
          });
        } catch (mediaErr) {
+          // Only durable ingress can replay an aborted album. Classic polling keeps
+          // its best-effort partial delivery so Telegram does not acknowledge a drop.
+          if (
+            mediaRuntimeWithAbort.abortSignal?.aborted &&
+            entry.spooledReplayParticipants.length > 0
+          ) {
+            throw mediaErr;
+          }
          if (!isRecoverableMediaGroupError(mediaErr)) {
            throw mediaErr;
          }
@@ -1138,6 +1152,7 @@ export const registerTelegramHandlers = ({
          ...spooledReplayOptions(entry.spooledReplayParticipants),
        },
        dispatchDedupeKeys: entry.dispatchDedupeKeys,
+        durableMediaReplay: entry.spooledReplayParticipants.length > 0,
      });
      settleSpooledReplayParticipants(entry.spooledReplayParticipants, result);
    } catch (err) {
@@ -1197,6 +1212,7 @@ export const registerTelegramHandlers = ({
          ...spooledReplayOptions(entry.spooledReplayParticipants),
        },
        dispatchDedupeKeys: entry.dispatchDedupeKeys,
+        durableMediaReplay: entry.spooledReplayParticipants.length > 0,
      });
      settleSpooledReplayParticipants(entry.spooledReplayParticipants, result);
    } catch (err) {
@@ -1424,6 +1440,7 @@ export const registerTelegramHandlers = ({
    ctx: TelegramContext,
    chain: TelegramCachedMessageNode[],
    shouldHydrateMedia: (node: TelegramCachedMessageNode, index: number) => Promise<boolean>,
+    durableMediaReplay: boolean,
  ): Promise<{ replyMedia: TelegramMediaRef[]; replyChain: TelegramReplyChainEntry[] }> => {
    const replyMedia: TelegramMediaRef[] = [];
    const replyChain: TelegramReplyChainEntry[] = [];
@@ -1440,7 +1457,7 @@ export const registerTelegramHandlers = ({
            ctx: {
              message: node.sourceMessage,
              me: ctx.me,
-              getFile: async () => await bot.api.getFile(replyFileId),
+              getFile: async (signal) => await bot.api.getFile(replyFileId, signal),
            },
            maxBytes: mediaMaxBytes,
            ...mediaRuntimeWithAbort,
@@ -1453,6 +1470,12 @@ export const registerTelegramHandlers = ({
              }
            : undefined;
        } catch (err) {
+          // Only durable ingress can replay a reply-media abort. Live polling must
+          // preserve the current text instead of acknowledging it without dispatch.
+          if (mediaRuntimeWithAbort.abortSignal?.aborted && durableMediaReplay) {
+            recordTelegramMessageProcessingResult({ kind: "failed-retryable", error: err });
+            throw err;
+          }
          logger.warn(
            { chatId: ctx.message.chat.id, error: String(err) },
            "reply media fetch failed",
@@ -1475,11 +1498,15 @@ export const registerTelegramHandlers = ({
    storeAllowFrom: string[];
    options?: TelegramMessageContextOptions;
    dispatchDedupeKeys?: string[];
+    durableMediaReplay?: boolean;
  }): Promise<TelegramMessageProcessingResult> => {
    let dispatchDedupeCommitted = false;
    let dispatchDedupeRollbackAttempted = false;
-    const spooledReplay =
-      params.options?.spooledReplay === true || isTelegramSpooledReplayUpdate(params.ctx.update);
+    // Callback-submit retries also set options.spooledReplay without durable ingress.
+    // Media aborts retry only when the update frame or a buffered participant owns replay.
+    const durableMediaReplay =
+      params.durableMediaReplay === true || isTelegramSpooledReplayUpdate(params.ctx.update);
+    const spooledReplay = params.options?.spooledReplay === true || durableMediaReplay;
    const forgetCommittedDispatchDedupeKeys = async () => {
      dispatchDedupeRollbackAttempted = true;
      await forgetDispatchDedupeKeys(params.dispatchDedupeKeys ?? []);
@@ -1547,6 +1574,7 @@ export const registerTelegramHandlers = ({
        params.ctx,
        replyChainNodes,
        shouldHydrateReplyMedia,
+        durableMediaReplay,
      );
      const promptContextMediaByMessageId = new Map<string, TelegramMediaRef>();
      const currentMessageId =
--- a/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts
+++ b/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts
@@ -67,8 +67,11 @@ const {
 } = harness;
 const { createTelegramBotCore: createTelegramBotBase, setTelegramBotRuntimeForTest } =
  await import("./bot-core.js");
-const { runWithTelegramUpdateProcessingFrame, withTelegramSpooledReplayUpdate } =
-  await import("./bot-processing-outcome.js");
+const {
+  runWithTelegramSpooledReplayUpdate,
+  runWithTelegramUpdateProcessingFrame,
+  withTelegramSpooledReplayUpdate,
+} = await import("./bot-processing-outcome.js");
 const { MediaFetchError } = await import("./telegram-media.runtime.js");

 let createTelegramBot: (
@@ -870,6 +873,123 @@ describe("createTelegramBot channel_post media", () => {
    }
  });

+  it("durably retries every spooled album update when shutdown aborts a download", async () => {
+    setOpenChannelPostConfig();
+    sendMessageSpy.mockClear();
+    replySpy.mockClear();
+
+    const shutdown = new AbortController();
+    saveRemoteMedia.mockImplementationOnce(async () => {
+      shutdown.abort();
+      throw Object.assign(new Error("aborted"), { name: "AbortError" });
+    });
+
+    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    try {
+      createTelegramBot({
+        token: "tok",
+        testTimings: TELEGRAM_TEST_TIMINGS,
+        fetchAbortSignal: shutdown.signal,
+      });
+      const handler = getOnHandler("channel_post") as (
+        ctx: Record<string, unknown>,
+      ) => Promise<void>;
+      const firstUpdate = { update_id: 98079 };
+      const secondUpdate = { update_id: 98080 };
+      const first = await runWithTelegramSpooledReplayUpdate(firstUpdate, () =>
+        handler({
+          ...createChannelPostContext({
+            messageId: 98079,
+            caption: "shutdown album",
+            date: 1736380800,
+            mediaGroupId: "shutdown-album-1",
+            photoFileId: "p1",
+          }),
+          update: firstUpdate,
+        }),
+      );
+      const second = await runWithTelegramSpooledReplayUpdate(secondUpdate, () =>
+        handler({
+          ...createChannelPostContext({
+            messageId: 98080,
+            date: 1736380801,
+            mediaGroupId: "shutdown-album-1",
+            photoFileId: "p2",
+          }),
+          update: secondUpdate,
+        }),
+      );
+
+      expect(first.deferredWork).toBeDefined();
+      expect(second.deferredWork).toBeDefined();
+      if (!first.deferredWork || !second.deferredWork) {
+        throw new Error("Expected both album updates to register durable replay work");
+      }
+      await flushChannelPostMediaGroup(setTimeoutSpy);
+
+      const [firstResult, secondResult] = await Promise.all([
+        first.deferredWork.task,
+        second.deferredWork.task,
+      ]);
+      expect(firstResult).toEqual({
+        kind: "failed-retryable",
+        error: expect.any(MediaFetchError),
+      });
+      expect(secondResult).toEqual({
+        kind: "failed-retryable",
+        error: expect.any(MediaFetchError),
+      });
+      expect(sendMessageSpy).not.toHaveBeenCalled();
+      expect(replySpy).not.toHaveBeenCalled();
+    } finally {
+      setTimeoutSpy.mockRestore();
+    }
+  });
+
+  it("keeps live album delivery when classic polling aborts a download", async () => {
+    setOpenChannelPostConfig();
+    sendMessageSpy.mockClear();
+    replySpy.mockClear();
+
+    const shutdown = new AbortController();
+    saveRemoteMedia
+      .mockImplementationOnce(async () => ({
+        path: "/tmp/classic-restart-first.jpg",
+        contentType: "image/jpeg",
+      }))
+      .mockImplementationOnce(async () => {
+        shutdown.abort();
+        throw Object.assign(new Error("aborted"), { name: "AbortError" });
+      });
+    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    try {
+      createTelegramBot({
+        token: "tok",
+        testTimings: TELEGRAM_TEST_TIMINGS,
+        fetchAbortSignal: shutdown.signal,
+      });
+      const handler = getOnHandler("channel_post") as (
+        ctx: Record<string, unknown>,
+      ) => Promise<void>;
+      await queueChannelPostAlbum(handler, {
+        caption: "classic restart album",
+        mediaGroupId: "classic-restart-album-1",
+        firstMessageId: 98081,
+        secondMessageId: 98082,
+      });
+      await flushChannelPostMediaGroup(setTimeoutSpy);
+      await waitForMockCalls(replySpy, 1);
+
+      expect(replySpy).toHaveBeenCalledTimes(1);
+      expect(replyPayload()).toMatchObject({
+        Body: expect.stringContaining("classic restart album"),
+        MediaPaths: ["/tmp/classic-restart-first.jpg"],
+      });
+    } finally {
+      setTimeoutSpy.mockRestore();
+    }
+  });
+
  it("drops the media group when a non-recoverable media error occurs", async () => {
    replySpy.mockReset();
    setOpenChannelPostConfig();
--- a/extensions/telegram/src/bot.test.ts
+++ b/extensions/telegram/src/bot.test.ts
@@ -50,6 +50,8 @@ const {
  wasSentByBot,
 } = await import("./bot.create-telegram-bot.test-harness.js");
 const { recordOutboundMessageForPromptContext } = await import("./outbound-message-context.js");
+const { runWithTelegramUpdateProcessingFrame, withTelegramSpooledReplayUpdate } =
+  await import("./bot-processing-outcome.js");

 let createTelegramBotBase: typeof import("./bot-core.js").createTelegramBotCore;
 let setTelegramBotRuntimeForTest: typeof import("./bot-core.js").setTelegramBotRuntimeForTest;
@@ -2989,6 +2991,9 @@ describe("createTelegramBot", () => {
    onSpy.mockClear();
    replySpy.mockClear();
    getFileSpy.mockClear();
+    const botShutdown = new AbortController();
+    const mediaAbort = new AbortController();
+    let replyGetFileSignal: AbortSignal | undefined;
    loadWebMedia.mockResolvedValueOnce({ path: "/tmp/reply-photo.png", contentType: "image/png" });

    const mediaFetch = vi.fn(
@@ -3003,6 +3008,8 @@ describe("createTelegramBot", () => {
    try {
      createTelegramBot({
        token: "tok",
+        fetchAbortSignal: botShutdown.signal,
+        mediaAbortSignal: mediaAbort.signal,
        telegramTransport: {
          fetch: mediaFetch as typeof fetch,
          sourceFetch: mediaFetch as typeof fetch,
@@ -3025,7 +3032,15 @@ describe("createTelegramBot", () => {
        me: { username: "openclaw_bot" },
        getFile: async () => ({}),
      });
+      replyGetFileSignal = mockArg(
+        getFileSpy as unknown as MockCallSource,
+        0,
+        1,
+        "reply getFile signal",
+      ) as AbortSignal;
+      expect(replyGetFileSignal.aborted).toBe(false);
    } finally {
+      mediaAbort.abort();
      ssrfMock.mockRestore();
    }

@@ -3041,11 +3056,147 @@ describe("createTelegramBot", () => {
      ReplyToBody?: string;
    };
    expect(payload.ReplyToBody).toBe("<media:image>");
-    expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1");
+    expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal));
+    expect(replyGetFileSignal?.aborted).toBe(true);
+    expect(botShutdown.signal.aborted).toBe(false);
+    botShutdown.abort();
    expect(loadWebMedia).not.toHaveBeenCalled();
    expect(mediaFetch).toHaveBeenCalledTimes(1);
  });

+  it("dispatches the current text when best-effort reply media times out", async () => {
+    onSpy.mockClear();
+    replySpy.mockClear();
+    getFileSpy.mockClear();
+    const timeout = Object.assign(new Error("media response headers timed out"), {
+      name: "TimeoutError",
+    });
+    const mediaFetch = vi.fn(async () => {
+      throw timeout;
+    });
+    const ssrfMock = mockPinnedHostnameResolution();
+
+    try {
+      createTelegramBot({
+        token: "tok",
+        telegramTransport: {
+          fetch: mediaFetch as typeof fetch,
+          sourceFetch: mediaFetch as typeof fetch,
+          close: async () => {},
+        },
+      });
+      const handler = getOnHandler("message") as (ctx: Record<string, unknown>) => Promise<void>;
+
+      await handler({
+        message: {
+          chat: { id: 7, type: "private" },
+          text: "continue without the old image",
+          date: 1736380800,
+          reply_to_message: {
+            message_id: 9001,
+            photo: [{ file_id: "reply-photo-1" }],
+            from: { first_name: "Ada" },
+          },
+        },
+        me: { username: "openclaw_bot" },
+        getFile: async () => ({}),
+      });
+    } finally {
+      ssrfMock.mockRestore();
+    }
+
+    expect(mediaFetch).toHaveBeenCalledTimes(1);
+    expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal));
+    expect(replySpy).toHaveBeenCalledTimes(1);
+    const payload = mockMsgContextArg(replySpy as unknown as MockCallSource, 0, 0, "replySpy call");
+    expect(payload.Body).toContain("continue without the old image");
+  });
+
+  it("dispatches the current text when classic polling aborts reply media", async () => {
+    onSpy.mockClear();
+    replySpy.mockClear();
+    getFileSpy.mockClear();
+    const botShutdown = new AbortController();
+    getFileSpy.mockImplementationOnce(async () => {
+      botShutdown.abort();
+      throw Object.assign(new Error("aborted"), { name: "AbortError" });
+    });
+
+    createTelegramBot({ token: "tok", fetchAbortSignal: botShutdown.signal });
+    const handler = getOnHandler("message") as (ctx: Record<string, unknown>) => Promise<void>;
+
+    const { result } = await runWithTelegramUpdateProcessingFrame(() =>
+      handler({
+        message: {
+          chat: { id: 7, type: "private" },
+          text: "continue after polling restart",
+          date: 1736380800,
+          reply_to_message: {
+            message_id: 9001,
+            photo: [{ file_id: "reply-photo-1" }],
+            from: { first_name: "Ada" },
+          },
+        },
+        me: { username: "openclaw_bot" },
+        getFile: async () => ({}),
+      }),
+    );
+
+    expect(result).toEqual({ kind: "completed" });
+    expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal));
+    expect(replySpy).toHaveBeenCalledTimes(1);
+    const payload = mockMsgContextArg(replySpy as unknown as MockCallSource, 0, 0, "replySpy call");
+    expect(payload.Body).toContain("continue after polling restart");
+  });
+
+  it("durably retries a spooled reply when shutdown aborts reply media", async () => {
+    onSpy.mockClear();
+    replySpy.mockClear();
+    getFileSpy.mockClear();
+    const botShutdown = new AbortController();
+    const mediaAbort = new AbortController();
+    getFileSpy.mockImplementationOnce(async () => {
+      botShutdown.abort();
+      throw new Error("Bad Request: file is too big");
+    });
+
+    createTelegramBot({
+      token: "tok",
+      fetchAbortSignal: botShutdown.signal,
+      mediaAbortSignal: mediaAbort.signal,
+    });
+    const handler = getOnHandler("message") as (ctx: Record<string, unknown>) => Promise<void>;
+    const update = {
+      update_id: 98081,
+      message: {
+        chat: { id: 7, type: "private" },
+        text: "keep the old image",
+        date: 1736380800,
+        reply_to_message: {
+          message_id: 9001,
+          photo: [{ file_id: "reply-photo-1" }],
+          from: { first_name: "Ada" },
+        },
+      },
+    };
+
+    const { result } = await runWithTelegramUpdateProcessingFrame(() =>
+      withTelegramSpooledReplayUpdate(update, () =>
+        handler({
+          update,
+          message: update.message,
+          me: { username: "openclaw_bot" },
+          getFile: async () => ({}),
+        }),
+      ),
+    );
+
+    expect(result).toEqual({ kind: "failed-retryable", error: expect.any(Error) });
+    expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal));
+    expect(replySpy).not.toHaveBeenCalled();
+    expect(mediaAbort.signal.aborted).toBe(false);
+  });
+
  it("hydrates reply chains from cached Telegram messages", async () => {
    onSpy.mockClear();
    replySpy.mockClear();
@@ -3164,7 +3315,7 @@ describe("createTelegramBot", () => {
    expect(messagesById.get("9000")?.media_path).toMatch(/^media:\/\/inbound\//);
    expect(messagesById.get("9000")?.media_path).not.toBe(payload.ReplyChain?.[1]?.mediaPath);
    expect(messagesById.get("9000")?.media_ref).toBeUndefined();
-    expect(getFileSpy).toHaveBeenCalledWith("root-photo-1");
+    expect(getFileSpy).toHaveBeenCalledWith("root-photo-1", expect.any(AbortSignal));
    expect(mediaFetch).toHaveBeenCalledTimes(1);
  });

@@ -3322,7 +3473,7 @@ describe("createTelegramBot", () => {
      is_reply_target: true,
    });
    if (expectHydrated) {
-      expect(getFileSpy).toHaveBeenCalledWith("generated-photo-1");
+      expect(getFileSpy).toHaveBeenCalledWith("generated-photo-1", expect.any(AbortSignal));
      expect(mediaFetch).toHaveBeenCalledTimes(1);
    } else {
      expect(getFileSpy).not.toHaveBeenCalled();
@@ -3582,7 +3733,7 @@ describe("createTelegramBot", () => {
      if (expectHydrated) {
        expect(replyMessage?.media_path).toMatch(/^media:\/\/inbound\//);
        expect(replyMessage?.media_ref).toBeUndefined();
-        expect(getFileSpy).toHaveBeenCalledWith("allowed-photo-1");
+        expect(getFileSpy).toHaveBeenCalledWith("allowed-photo-1", expect.any(AbortSignal));
        expect(mediaFetch).toHaveBeenCalledTimes(1);
      } else {
        expect(replyMessage?.media_path).toBeUndefined();
@@ -3729,7 +3880,7 @@ describe("createTelegramBot", () => {
      await flushTimer?.();
      await replyDelivered;

-      expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1");
+      expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal));
      expect(mediaFetch).toHaveBeenCalled();
    } finally {
      setTimeoutSpy.mockRestore();
--- a/extensions/telegram/src/bot.types.ts
+++ b/extensions/telegram/src/bot.types.ts
@@ -20,6 +20,8 @@ export type TelegramBotOptions = {
  botInfo?: TelegramBotInfo;
  /** Signal to abort in-flight Telegram API fetch requests (e.g. getUpdates) on shutdown. */
  fetchAbortSignal?: AbortSignal;
+  /** Signal to abort inbound media resolution without cancelling adopted-turn Bot API calls. */
+  mediaAbortSignal?: AbortSignal;
  /** Minimum grammY client timeout when timeoutSeconds is configured on long-polling bots. */
  minimumClientTimeoutSeconds?: number;
  updateOffset?: {
--- a/extensions/telegram/src/bot/delivery.resolve-media-retry.test.ts
+++ b/extensions/telegram/src/bot/delivery.resolve-media-retry.test.ts
@@ -594,8 +594,11 @@ describe("resolveMedia getFile retry", () => {
      fetchImpl: callerFetch,
      dispatcherAttempts,
      trustExplicitProxyDns: true,
+      responseHeaderTimeoutMs: 120_000,
      readIdleTimeoutMs: 30_000,
    });
+    expect(params.timeoutMs).toBeUndefined();
+    expect(params.retry).toBeUndefined();
    expect(typeof params.shouldRetryFetchError).toBe("function");
    expectFetchSsrfPolicyFields({
      allowRfc2544BenchmarkRange: true,
@@ -625,6 +628,49 @@ describe("resolveMedia getFile retry", () => {
    expectReadRemoteMediaBufferFields({ fetchImpl: callerFetch });
  });

+  it.each([
+    { mediaField: "document" as const, filePath: "documents/file_42.pdf" },
+    { mediaField: "sticker" as const, filePath: "stickers/file_0.webp" },
+  ])("keeps the session abort signal attached to $mediaField downloads", async (scenario) => {
+    const shutdown = new AbortController();
+    const getFile = vi.fn().mockResolvedValue({ file_path: scenario.filePath });
+    readRemoteMediaBuffer.mockResolvedValueOnce({
+      buffer: Buffer.from("media"),
+      contentType: scenario.mediaField === "sticker" ? "image/webp" : "application/pdf",
+      fileName: scenario.filePath.split("/").at(-1),
+    });
+    saveMediaBuffer.mockResolvedValueOnce({
+      path: `/tmp/${scenario.filePath.split("/").at(-1)}`,
+      contentType: scenario.mediaField === "sticker" ? "image/webp" : "application/pdf",
+    });
+
+    await resolveMediaWithDefaults(makeCtx(scenario.mediaField, getFile), {
+      abortSignal: shutdown.signal,
+    });
+
+    expect(requireReadRemoteMediaBufferParams()).toMatchObject({
+      requestInit: { signal: shutdown.signal },
+      responseHeaderTimeoutMs: 120_000,
+      readIdleTimeoutMs: 30_000,
+    });
+  });
+
+  it("omits nested download retries so callers own failure handling", async () => {
+    const timeout = Object.assign(new Error("request timed out"), { name: "TimeoutError" });
+    const fetchError = Object.assign(new Error("failed to fetch media", { cause: timeout }), {
+      name: "MediaFetchError",
+      code: "fetch_failed",
+    });
+    const getFile = vi.fn().mockResolvedValue({ file_path: "documents/first.pdf" });
+    readRemoteMediaBuffer.mockRejectedValueOnce(fetchError);
+
+    await expect(resolveMediaWithDefaults(makeCtx("document", getFile))).rejects.toBe(fetchError);
+
+    expect(readRemoteMediaBuffer).toHaveBeenCalledTimes(1);
+    expect(saveRemoteMedia).toHaveBeenCalledTimes(1);
+    expect(requireReadRemoteMediaBufferParams().retry).toBeUndefined();
+  });
+
  it("allows an explicit Telegram apiRoot host without broadening the default SSRF allowlist", async () => {
    const getFile = vi.fn().mockResolvedValue({ file_path: "documents/file_42.pdf" });
    readRemoteMediaBuffer.mockResolvedValueOnce({
--- a/extensions/telegram/src/bot/delivery.resolve-media.ts
+++ b/extensions/telegram/src/bot/delivery.resolve-media.ts
@@ -184,6 +184,8 @@ function resolveRequiredTelegramTransport(transport?: TelegramTransport): Telegr

 /** Default idle timeout for Telegram media downloads (30 seconds). */
 const TELEGRAM_DOWNLOAD_IDLE_TIMEOUT_MS = 30_000;
+/** Maximum wait for Telegram media response headers (120 seconds). */
+const TELEGRAM_DOWNLOAD_RESPONSE_HEADER_TIMEOUT_MS = 120_000;

 function usesTrustedTelegramExplicitProxy(transport: TelegramTransport): boolean {
  return (
@@ -274,6 +276,7 @@ async function downloadAndSaveTelegramFile(params: {
  apiRoot?: string;
  trustedLocalFileRoots?: readonly string[];
  dangerouslyAllowPrivateNetwork?: boolean;
+  abortSignal?: AbortSignal;
 }) {
  const trustedLocalFile = resolveTrustedLocalTelegramRoot(
    params.filePath,
@@ -345,17 +348,12 @@ async function downloadAndSaveTelegramFile(params: {
    dispatcherAttempts: transport.dispatcherAttempts,
    trustExplicitProxyDns: usesTrustedTelegramExplicitProxy(transport),
    shouldRetryFetchError: shouldRetryTelegramTransportFallback,
-    retry: {
-      attempts: 3,
-      minDelayMs: 1000,
-      maxDelayMs: 4000,
-      jitter: 0.2,
-      label: "telegram:media-download",
-      onRetry: ({ attempt, maxAttempts }) =>
-        logVerbose(`telegram: media download retry ${attempt}/${maxAttempts}`),
-    },
+    // The update spool and best-effort album/reply callers own failure handling.
+    // Nested retries would multiply this header deadline before those owners act.
+    ...(params.abortSignal ? { requestInit: { signal: params.abortSignal } } : {}),
    filePathHint: params.filePath,
    maxBytes: params.maxBytes,
+    responseHeaderTimeoutMs: TELEGRAM_DOWNLOAD_RESPONSE_HEADER_TIMEOUT_MS,
    readIdleTimeoutMs: TELEGRAM_DOWNLOAD_IDLE_TIMEOUT_MS,
    ssrfPolicy: buildTelegramMediaSsrfPolicy(params.apiRoot, params.dangerouslyAllowPrivateNetwork),
    fallbackContentType: params.mimeType,
@@ -409,6 +407,7 @@ async function resolveStickerMedia(params: {
    apiRoot: params.apiRoot,
    trustedLocalFileRoots: params.trustedLocalFileRoots,
    dangerouslyAllowPrivateNetwork: params.dangerouslyAllowPrivateNetwork,
+    abortSignal,
  });

  // Check sticker cache for existing description
@@ -516,6 +515,7 @@ export async function resolveMedia(params: {
    apiRoot,
    trustedLocalFileRoots,
    dangerouslyAllowPrivateNetwork,
+    abortSignal,
  });
  const placeholder = saved.contentType?.startsWith("audio/")
    ? "<media:audio>"
--- a/extensions/telegram/src/polling-session.test.ts
+++ b/extensions/telegram/src/polling-session.test.ts
@@ -3510,6 +3510,34 @@ describe("TelegramPollingSession", () => {

      const runPromise = session.runUntilAbort();
      await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(2));
+      const firstFetchSignal = mockObjectArg(
+        createTelegramBotMock,
+        "first createTelegramBot",
+        0,
+      ).fetchAbortSignal;
+      const firstMediaSignal = mockObjectArg(
+        createTelegramBotMock,
+        "first createTelegramBot",
+        0,
+      ).mediaAbortSignal;
+      const secondFetchSignal = mockObjectArg(
+        createTelegramBotMock,
+        "second createTelegramBot",
+        1,
+      ).fetchAbortSignal;
+      const secondMediaSignal = mockObjectArg(
+        createTelegramBotMock,
+        "second createTelegramBot",
+        1,
+      ).mediaAbortSignal;
+      expect(firstFetchSignal).toBeInstanceOf(AbortSignal);
+      expect(firstMediaSignal).toBeInstanceOf(AbortSignal);
+      expect(secondFetchSignal).toBeInstanceOf(AbortSignal);
+      expect(secondMediaSignal).toBeInstanceOf(AbortSignal);
+      expect((firstFetchSignal as AbortSignal).aborted).toBe(false);
+      expect((firstMediaSignal as AbortSignal).aborted).toBe(true);
+      expect((secondFetchSignal as AbortSignal).aborted).toBe(false);
+      expect((secondMediaSignal as AbortSignal).aborted).toBe(false);
      expectLogIncludes(log, "isolated polling ingress failed: worker crashed");
      expect(
        statusPatches(setStatus).some(
@@ -3520,12 +3548,239 @@ describe("TelegramPollingSession", () => {
      abort.abort();
      await vi.advanceTimersByTimeAsync(20_000);
      await runPromise;
+      expect((firstFetchSignal as AbortSignal).aborted).toBe(true);
+      expect((secondFetchSignal as AbortSignal).aborted).toBe(true);
+      expect((secondMediaSignal as AbortSignal).aborted).toBe(true);
    } finally {
      vi.useRealTimers();
      await fs.rm(tempDir, { recursive: true, force: true });
    }
  });

+  it("waits for a fresh bot before draining updates after an isolated worker crash", async () => {
+    vi.useFakeTimers({ shouldAdvanceTime: true });
+    const abort = new AbortController();
+    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
+    let releaseBackoff: (() => void) | undefined;
+    const backoff = new Promise<void>((resolve) => {
+      releaseBackoff = resolve;
+    });
+    sleepWithAbortMock.mockImplementationOnce(async () => {
+      await backoff;
+      return undefined;
+    });
+
+    let firstMediaSignal: AbortSignal | undefined;
+    let rejectFirstWorker: ((err: Error) => void) | undefined;
+    const firstWorkerDone = new Promise<void>((_resolve, reject) => {
+      rejectFirstWorker = reject;
+    });
+    const firstHandleUpdate = vi.fn(async () => {
+      rejectFirstWorker?.(new Error("worker crashed"));
+      if (!firstMediaSignal) {
+        throw new Error("Expected the first polling cycle signal");
+      }
+      await waitForAbortSignal(firstMediaSignal);
+      throw Object.assign(new Error("aborted"), { name: "AbortError" });
+    });
+    const secondHandleUpdate = vi.fn(async () => undefined);
+    const createBot = (handleUpdate: (update: { update_id?: number }) => Promise<unknown>) => ({
+      api: {
+        deleteWebhook: vi.fn(async () => true),
+        config: { use: vi.fn() },
+      },
+      init: vi.fn(async () => undefined),
+      handleUpdate,
+      stop: vi.fn(async () => undefined),
+    });
+    createTelegramBotMock
+      .mockImplementationOnce((opts: { mediaAbortSignal?: AbortSignal }) => {
+        firstMediaSignal = opts.mediaAbortSignal;
+        return createBot(firstHandleUpdate);
+      })
+      .mockReturnValueOnce(createBot(secondHandleUpdate));
+
+    let workerIndex = 0;
+    let stopSecondWorker: (() => void) | undefined;
+    const secondWorkerDone = new Promise<void>((resolve) => {
+      stopSecondWorker = resolve;
+    });
+    const createWorker = vi.fn(() => {
+      workerIndex += 1;
+      if (workerIndex === 1) {
+        return {
+          onMessage: vi.fn(() => () => undefined),
+          stop: vi.fn(async () => undefined),
+          task: vi.fn(async () => await firstWorkerDone),
+        };
+      }
+      return {
+        onMessage: vi.fn(() => () => undefined),
+        stop: vi.fn(async () => {
+          stopSecondWorker?.();
+        }),
+        task: vi.fn(async () => await secondWorkerDone),
+      };
+    });
+
+    try {
+      const session = createPollingSession({
+        abortSignal: abort.signal,
+        isolatedIngress: {
+          enabled: true,
+          spoolDir: tempDir,
+          createWorker,
+          drainIntervalMs: 10,
+        },
+      });
+      const runPromise = session.runUntilAbort();
+      await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(1));
+
+      await writeSpooledTestUpdates(tempDir, [
+        topicUpdate(42, 10, "crash the old bot"),
+        topicUpdate(43, 11, "wait for the fresh bot"),
+      ]);
+      await vi.advanceTimersByTimeAsync(50);
+      await vi.waitFor(() => expect(firstHandleUpdate).toHaveBeenCalledTimes(1));
+      await vi.waitFor(() => expect(sleepWithAbortMock).toHaveBeenCalledTimes(1));
+      await vi.advanceTimersByTimeAsync(5_000);
+      expect(firstHandleUpdate).toHaveBeenCalledTimes(1);
+      await vi.waitFor(async () => {
+        const unstarted = (
+          await listTelegramSpooledUpdates({ spoolDir: tempDir, limit: "all" })
+        ).find((update) => update.updateId === 43);
+        expect(unstarted).toMatchObject({ attempts: 0 });
+        expect(unstarted?.lastAttemptAt).toBeUndefined();
+        expect(unstarted?.lastError).toBeUndefined();
+      });
+
+      releaseBackoff?.();
+      await vi.advanceTimersByTimeAsync(2_000);
+      await vi.waitFor(() => expect(secondHandleUpdate).toHaveBeenCalledTimes(2));
+      abort.abort();
+      await vi.advanceTimersByTimeAsync(20_000);
+      await runPromise;
+
+      expect(createWorker).toHaveBeenCalledTimes(2);
+      expect(firstHandleUpdate).toHaveBeenCalledTimes(1);
+      expect(await pendingUpdateIds(tempDir, "all")).toEqual([]);
+    } finally {
+      releaseBackoff?.();
+      abort.abort();
+      stopSecondWorker?.();
+      vi.useRealTimers();
+      await fs.rm(tempDir, { recursive: true, force: true });
+    }
+  });
+
+  it("keeps adopted-turn Bot API delivery alive when an isolated worker crashes", async () => {
+    vi.useFakeTimers({ shouldAdvanceTime: true });
+    const abort = new AbortController();
+    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
+    let releaseBackoff: (() => void) | undefined;
+    const backoff = new Promise<void>((resolve) => {
+      releaseBackoff = resolve;
+    });
+    sleepWithAbortMock.mockImplementationOnce(async () => {
+      await backoff;
+      return undefined;
+    });
+
+    let adopted: (() => void) | undefined;
+    const adoptedTurn = new Promise<void>((resolve) => {
+      adopted = resolve;
+    });
+    let deliveryResolve: (() => void) | undefined;
+    let deliveryReject: ((err: unknown) => void) | undefined;
+    const delivery = new Promise<void>((resolve, reject) => {
+      deliveryResolve = resolve;
+      deliveryReject = reject;
+    });
+    let firstFetchSignal: AbortSignal | undefined;
+    let firstMediaSignal: AbortSignal | undefined;
+    const sendMessage = vi.fn(async () => {
+      if (firstFetchSignal?.aborted) {
+        throw new Error("adopted-turn Bot API client was aborted");
+      }
+    });
+    createTelegramBotMock.mockImplementationOnce(
+      (opts: { fetchAbortSignal?: AbortSignal; mediaAbortSignal?: AbortSignal }) => {
+        firstFetchSignal = opts.fetchAbortSignal;
+        firstMediaSignal = opts.mediaAbortSignal;
+        const api = {
+          deleteWebhook: vi.fn(async () => true),
+          sendMessage,
+          config: { use: vi.fn() },
+        };
+        return {
+          api,
+          init: vi.fn(async () => undefined),
+          handleUpdate: vi.fn(async (update: { update_id?: number }) => {
+            const participant = createTelegramSpooledReplayDeferredParticipant(
+              `test-adopted-delivery:${update.update_id}`,
+            );
+            if (!participant || !firstMediaSignal) {
+              throw new Error("Expected a spooled participant and media signal");
+            }
+            participant.settle({ kind: "completed" });
+            adopted?.();
+            void (async () => {
+              try {
+                await waitForAbortSignal(firstMediaSignal);
+                // Streaming, edits, and native-quote fallbacks retain this adopted bot client.
+                await api.sendMessage();
+                deliveryResolve?.();
+              } catch (err) {
+                deliveryReject?.(err);
+              }
+            })();
+          }),
+          stop: vi.fn(async () => undefined),
+        };
+      },
+    );
+
+    const createWorker = vi.fn(() => ({
+      onMessage: vi.fn(() => () => undefined),
+      stop: vi.fn(async () => undefined),
+      task: vi.fn(async () => {
+        await adoptedTurn;
+        throw new Error("worker crashed after turn adoption");
+      }),
+    }));
+
+    try {
+      await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "finish after worker crash")]);
+      const session = createPollingSession({
+        abortSignal: abort.signal,
+        isolatedIngress: {
+          enabled: true,
+          spoolDir: tempDir,
+          createWorker,
+          drainIntervalMs: 10,
+        },
+      });
+      const runPromise = session.runUntilAbort();
+
+      await delivery;
+      expect(sendMessage).toHaveBeenCalledTimes(1);
+      expect(firstMediaSignal?.aborted).toBe(true);
+      expect(firstFetchSignal?.aborted).toBe(false);
+      await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
+
+      abort.abort();
+      releaseBackoff?.();
+      await vi.advanceTimersByTimeAsync(20_000);
+      await runPromise;
+      expect(firstFetchSignal?.aborted).toBe(true);
+    } finally {
+      releaseBackoff?.();
+      abort.abort();
+      vi.useRealTimers();
+      await fs.rm(tempDir, { recursive: true, force: true });
+    }
+  });
+
  it("treats isolated ingress worker rejection after abort as clean shutdown", async () => {
    vi.useFakeTimers({ shouldAdvanceTime: true });
    const abort = new AbortController();
--- a/extensions/telegram/src/polling-session.ts
+++ b/extensions/telegram/src/polling-session.ts
@@ -37,6 +37,7 @@ import {
  TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS,
 } from "./spooled-update-retry-policy.js";
 import {
+  abandonTelegramSpooledUpdateClaim,
  claimNextTelegramSpooledUpdate,
  completeTelegramSpooledUpdate,
  failTelegramSpooledUpdateClaim,
@@ -328,7 +329,7 @@ export class TelegramPollingSession {
  #webhookCleared = false;
  #forceRestarted = false;
  #activeRunner: ReturnType<typeof run> | undefined;
-  #activeFetchAbort: AbortController | undefined;
+  #activeCycleAbort: AbortController | undefined;
  #spooledUpdateHandlerKeys = new Set<string>();
  #deferredSpooledUpdateClaimKeys = new Set<string>();
  #transportState: TelegramPollingTransportState;
@@ -372,7 +373,7 @@ export class TelegramPollingSession {
  }

  abortActiveFetch() {
-    this.#activeFetchAbort?.abort();
+    this.#activeCycleAbort?.abort();
  }

  async runUntilAbort(): Promise<void> {
@@ -493,8 +494,16 @@ export class TelegramPollingSession {
  }

  async #createPollingBot(): Promise<TelegramBot | undefined> {
-    const fetchAbortController = new AbortController();
-    this.#activeFetchAbort = fetchAbortController;
+    const cycleAbortController = new AbortController();
+    this.#activeCycleAbort = cycleAbortController;
+    const cycleAbortSignal = this.opts.abortSignal
+      ? AbortSignal.any([this.opts.abortSignal, cycleAbortController.signal])
+      : cycleAbortController.signal;
+    // Isolated turns can outlive their polling worker after adoption. Keep their
+    // Bot API client session-owned while media remains cycle-owned and retryable.
+    const botApiAbortSignal = this.opts.isolatedIngress?.enabled
+      ? this.opts.abortSignal
+      : cycleAbortSignal;
    const telegramTransport = this.#transportState.acquireForNextCycle();
    const persistedLastUpdateId = this.opts.getLastUpdateId();
    const lastUpdateId = this.opts.isolatedIngress?.enabled ? null : persistedLastUpdateId;
@@ -511,15 +520,16 @@ export class TelegramPollingSession {
        config: this.opts.config,
        accountId: this.opts.accountId,
        botInfo: this.opts.botInfo,
-        fetchAbortSignal: fetchAbortController.signal,
+        ...(botApiAbortSignal ? { fetchAbortSignal: botApiAbortSignal } : {}),
+        mediaAbortSignal: cycleAbortSignal,
        minimumClientTimeoutSeconds: TELEGRAM_POLLING_CLIENT_TIMEOUT_FLOOR_SECONDS,
        ...(updateOffset ? { updateOffset } : {}),
        telegramTransport,
      });
    } catch (err) {
      await this.#waitBeforeRetryOnRecoverableSetupError(err, "Telegram setup network error");
-      if (this.#activeFetchAbort === fetchAbortController) {
-        this.#activeFetchAbort = undefined;
+      if (this.#activeCycleAbort === cycleAbortController) {
+        this.#activeCycleAbort = undefined;
      }
      return undefined;
    }
@@ -900,6 +910,7 @@ export class TelegramPollingSession {
  async #drainSpooledUpdates(params: {
    bot: TelegramBot;
    isDrainHealthy: () => boolean;
+    shouldStop: () => boolean;
    spoolDir: string;
  }): Promise<SpooledUpdateDrainResult> {
    const activeLaneKeys = this.#activeSpooledUpdateLaneKeysForSpool(params.spoolDir);
@@ -948,7 +959,7 @@ export class TelegramPollingSession {
    ]);
    let started = 0;
    while (started < TELEGRAM_SPOOLED_DRAIN_START_LIMIT) {
-      if (this.opts.abortSignal?.aborted) {
+      if (params.shouldStop() || this.opts.abortSignal?.aborted) {
        break;
      }
      const claimedUpdate = await this.#claimNextSpooledUpdate({
@@ -959,13 +970,21 @@ export class TelegramPollingSession {
      if (!claimedUpdate) {
        break;
      }
+      if (params.shouldStop() || this.opts.abortSignal?.aborted) {
+        try {
+          await abandonTelegramSpooledUpdateClaim(claimedUpdate);
+        } catch (err) {
+          this.opts.log(
+            `[telegram][diag] spooled update ${claimedUpdate.updateId} could not be requeued after its polling cycle ended: ${formatErrorMessage(err)}`,
+          );
+        }
+        break;
+      }
      const laneKey = this.#spooledUpdateLaneKey(claimedUpdate);
      const handlerKey = buildSpooledUpdateHandlerKey({ spoolDir: params.spoolDir, laneKey });
      if (activeSpooledUpdateHandlersByLane.has(handlerKey)) {
        blockedByLane.add(handlerKey);
-        await releaseTelegramSpooledUpdateClaim(claimedUpdate, {
-          lastError: "active Telegram spool handler already owns this lane",
-        });
+        await abandonTelegramSpooledUpdateClaim(claimedUpdate);
        blockedLaneKeys.add(laneKey);
        continue;
      }
@@ -1141,9 +1160,17 @@ export class TelegramPollingSession {
    if (!ingress?.enabled) {
      return this.#runPollingCycle(bot);
    }
+    const cycleAbortController = this.#activeCycleAbort;
+    const abortMedia = () => {
+      cycleAbortController?.abort();
+    };
    try {
      await bot.init();
    } catch (err) {
+      abortMedia();
+      if (this.#activeCycleAbort === cycleAbortController) {
+        this.#activeCycleAbort = undefined;
+      }
      const shouldRetry = await this.#waitBeforeRetryOnRecoverableSetupError(
        err,
        "Telegram bot init failed",
@@ -1196,6 +1223,11 @@ export class TelegramPollingSession {
    const stalledBacklogKeys = new Set<string>();
    let requestImmediateDrain: () => void = () => undefined;
    let drainRequested = false;
+    let cycleEnding = false;
+    const endCycle = () => {
+      cycleEnding = true;
+      abortMedia();
+    };
    const unsubscribe = worker.onMessage((message) => {
      const ackSpooledUpdate = (
        requestId: string,
@@ -1268,6 +1300,7 @@ export class TelegramPollingSession {
      }
    });
    const stopOnAbort = () => {
+      endCycle();
      void stopWorker();
    };
    this.opts.abortSignal?.addEventListener("abort", stopOnAbort, { once: true });
@@ -1295,6 +1328,7 @@ export class TelegramPollingSession {
        return;
      }
      restartRequested = true;
+      endCycle();
      void stopWorker();
      if (!forceCycleTimer) {
        forceCycleTimer = setTimeout(() => {
@@ -1310,7 +1344,7 @@ export class TelegramPollingSession {
      }
    };
    const drainOnce = async () => {
-      if (restartRequested || this.opts.abortSignal?.aborted) {
+      if (cycleEnding || restartRequested || this.opts.abortSignal?.aborted) {
        return;
      }
      if (drainActive) {
@@ -1324,6 +1358,7 @@ export class TelegramPollingSession {
        const drain = await this.#drainSpooledUpdates({
          bot,
          isDrainHealthy,
+          shouldStop: () => cycleEnding,
          spoolDir,
        });
        consecutiveDrainFailures = 0;
@@ -1373,7 +1408,12 @@ export class TelegramPollingSession {
          drainHealth.lastCompletedAt = Date.now();
        }
        drainActive = false;
-        if (drainRequested && !restartRequested && !this.opts.abortSignal?.aborted) {
+        if (
+          drainRequested &&
+          !cycleEnding &&
+          !restartRequested &&
+          !this.opts.abortSignal?.aborted
+        ) {
          drainRequested = false;
          // Handler finalizers clear active lane guards in microtasks; redrain
          // after them so newly unblocked same-lane rows can claim immediately.
@@ -1410,10 +1450,12 @@ export class TelegramPollingSession {
      try {
        await Promise.race([worker.task(), forceCyclePromise]);
        clearForceCycleTimer();
+        endCycle();
      } catch (err) {
        if (this.opts.abortSignal?.aborted) {
          return "exit";
        }
+        endCycle();
        // The worker only issues getUpdates, so a 409 is always a duplicate
        // poller (or stale webhook) conflict. Mirror the classic polling
        // cycle: re-clear the webhook, rotate the transport (#69787), and
@@ -1469,12 +1511,17 @@ export class TelegramPollingSession {
      clearForceCycleTimer();
      unsubscribe();
      this.opts.abortSignal?.removeEventListener("abort", stopOnAbort);
+      // End media work before waiting for durable handlers so every interrupted claim can retry.
+      endCycle();
      await stopWorker();
      if (!restartRequested) {
        await drainOnce();
        await waitForGracefulStop(() => this.#waitForSpooledUpdateHandlers());
      }
      await waitForGracefulStop(stopBot);
+      if (this.#activeCycleAbort === cycleAbortController) {
+        this.#activeCycleAbort = undefined;
+      }
    }
  }

@@ -1508,7 +1555,7 @@ export class TelegramPollingSession {
    const runner = run(bot, this.opts.runnerOptions);
    this.opts.log(`[telegram][diag] polling cycle started ${liveness.formatDiagnosticFields()}`);
    this.#activeRunner = runner;
-    const fetchAbortController = this.#activeFetchAbort;
+    const fetchAbortController = this.#activeCycleAbort;
    const abortFetch = () => {
      fetchAbortController?.abort();
    };
@@ -1657,8 +1704,8 @@ export class TelegramPollingSession {
      await waitForGracefulStop(stopRunner);
      await waitForGracefulStop(stopBot);
      this.#activeRunner = undefined;
-      if (this.#activeFetchAbort === fetchAbortController) {
-        this.#activeFetchAbort = undefined;
+      if (this.#activeCycleAbort === fetchAbortController) {
+        this.#activeCycleAbort = undefined;
      }
    }
  }
--- a/extensions/telegram/src/telegram-ingress-spool.ts
+++ b/extensions/telegram/src/telegram-ingress-spool.ts
@@ -341,6 +341,15 @@ export async function releaseTelegramSpooledUpdateClaim(
  );
 }

+export async function abandonTelegramSpooledUpdateClaim(
+  update: ClaimedTelegramSpooledUpdate,
+): Promise<void> {
+  await createTelegramIngressQueue(path.dirname(update.pendingPath)).release(
+    queueMutationTarget(update),
+    { recordAttempt: false },
+  );
+}
+
 export async function refreshTelegramSpooledUpdateClaim(
  update: ClaimedTelegramSpooledUpdate,
  options?: { refreshedAt?: number },
--- a/extensions/telegram/src/webhook.test.ts
+++ b/extensions/telegram/src/webhook.test.ts
@@ -575,6 +575,41 @@ describe("startTelegramWebhook", () => {
    );
  });

+  it("aborts bot media fetches when the webhook stops", async () => {
+    const callerAbort = new AbortController();
+    const started = await startTelegramWebhook({
+      token: TELEGRAM_TOKEN,
+      secret: TELEGRAM_SECRET,
+      port: 0,
+      abortSignal: callerAbort.signal,
+      path: TELEGRAM_WEBHOOK_PATH,
+      spoolDir: requireWebhookSpoolDir(),
+    });
+
+    try {
+      const botParams = requireRecord(
+        requireMockCall(createTelegramBotSpy, 0, "createTelegramBot")[0],
+        "createTelegramBot params",
+      );
+      const fetchAbortSignal = botParams.fetchAbortSignal;
+      expect(fetchAbortSignal).toBeInstanceOf(AbortSignal);
+      if (!(fetchAbortSignal instanceof AbortSignal)) {
+        throw new Error("expected bot fetch abort signal");
+      }
+      const aborted = new Promise<void>((resolve) => {
+        fetchAbortSignal.addEventListener("abort", () => resolve(), { once: true });
+      });
+
+      await started.stop();
+
+      await expect(aborted).resolves.toBeUndefined();
+      expect(callerAbort.signal.aborted).toBe(false);
+    } finally {
+      await started.stop();
+      callerAbort.abort();
+    }
+  });
+
  it("keeps local listener alive and retries when setWebhook has a recoverable startup failure", async () => {
    const runtimeLog = vi.fn();
    const runtimeError = vi.fn();
--- a/extensions/telegram/src/webhook.ts
+++ b/extensions/telegram/src/webhook.ts
@@ -700,10 +700,15 @@ export async function startTelegramWebhook(opts: {
    closeTransportPromise ??= telegramTransport.close();
    return closeTransportPromise;
  };
+  const botAbortController = new AbortController();
+  const botFetchAbortSignal = opts.abortSignal
+    ? AbortSignal.any([opts.abortSignal, botAbortController.signal])
+    : botAbortController.signal;
  const bot = createTelegramBot({
    token: opts.token,
    runtime,
    proxyFetch: opts.fetch,
+    fetchAbortSignal: botFetchAbortSignal,
    config: opts.config,
    accountId: opts.accountId,
    telegramTransport,
@@ -716,6 +721,7 @@ export async function startTelegramWebhook(opts: {
      retryPolicy: webhookRegistrationRetryPolicy,
    });
  } catch (err) {
+    botAbortController.abort();
    await bot.stop();
    await closeTransportOnce();
    throw err;
@@ -976,6 +982,7 @@ export async function startTelegramWebhook(opts: {
    if (shutDown) {
      return;
    }
+    botAbortController.abort();
    shutDown = true;
    if (drainTimer) {
      clearInterval(drainTimer);
--- a/src/channels/message/ingress-queue.test.ts
+++ b/src/channels/message/ingress-queue.test.ts
@@ -248,6 +248,18 @@ describe("channel ingress queue", () => {
        lastAttemptAt: 20,
        lastError: "retry",
      });
+
+      const reclaimed = await queue.claim("b", { ownerId: "replacement" });
+      if (!reclaimed) {
+        throw new Error("Expected the released ingress event to be claimable");
+      }
+      expect(await queue.release(reclaimed, { recordAttempt: false, releasedAt: 30 })).toBe(true);
+      expect((await queue.listPending()).find((record) => record.id === "b")).toMatchObject({
+        attempts: 1,
+        lastAttemptAt: 20,
+        lastError: "retry",
+        updatedAt: 30,
+      });
    });
  });

--- a/src/channels/message/ingress-queue.ts
+++ b/src/channels/message/ingress-queue.ts
@@ -156,7 +156,7 @@ export type ChannelIngressQueue<TPayload, TMetadata = unknown, TCompletedMetadat
  ): Promise<boolean>;
  release(
    idOrClaim: string | ChannelIngressQueueClaimRef,
-    options?: { lastError?: string; releasedAt?: number },
+    options?: { lastError?: string; releasedAt?: number; recordAttempt?: boolean },
  ): Promise<boolean>;
  fail(
    idOrClaim: string | ChannelIngressQueueClaimRef,
@@ -769,8 +769,14 @@ export function createChannelIngressQueue<
            claim_token: null,
            claim_owner: null,
            claimed_at: null,
-            attempts: eb("attempts", "+", 1),
-            last_attempt_at: releasedAt,
+            // A claim can lose its owner before processing starts. Returning it
+            // must not consume retry budget or erase the previous real failure.
+            ...(releaseOptions?.recordAttempt === false
+              ? {}
+              : {
+                  attempts: eb("attempts", "+", 1),
+                  last_attempt_at: releasedAt,
+                }),
            ...(releaseOptions?.lastError === undefined
              ? {}
              : { last_error: releaseOptions.lastError }),
--- a/src/infra/net/fetch-guard.ssrf.test.ts
+++ b/src/infra/net/fetch-guard.ssrf.test.ts
@@ -208,6 +208,16 @@ describe("fetchWithSsrFGuard hardening", () => {

  const createPublicLookup = (): LookupFn =>
    vi.fn(async () => [{ address: "93.184.216.34", family: 4 }]) as unknown as LookupFn;
+  const createStalledLookup = () => {
+    let release: (() => void) | undefined;
+    const lookupFn = vi.fn(
+      async () =>
+        await new Promise<Array<{ address: string; family: 4 }>>((resolve) => {
+          release = () => resolve([{ address: "93.184.216.34", family: 4 }]);
+        }),
+    ) as unknown as LookupFn;
+    return { lookupFn, release: () => release?.() };
+  };
  const createLoopbackLookup = (): LookupFn =>
    vi.fn(async () => [{ address: "127.0.0.1", family: 4 }]) as unknown as LookupFn;
  const createIpv6LoopbackLookup = (): LookupFn =>
@@ -1902,6 +1912,48 @@ describe("fetchWithSsrFGuard hardening", () => {
    expect(outcome).toBe("TimeoutError");
  });

+  it("aborts a stalled DNS preflight from the caller signal without dispatching", async () => {
+    const stalledLookup = createStalledLookup();
+    const fetchImpl = vi.fn(async () => okResponse());
+    const controller = new AbortController();
+    const abortError = Object.assign(new Error("gateway shutdown"), { name: "AbortError" });
+    const fetchPromise = fetchWithSsrFGuard({
+      url: "https://public.example/resource",
+      fetchImpl,
+      lookupFn: stalledLookup.lookupFn,
+      signal: controller.signal,
+    }).catch((error: unknown) => error);
+
+    await vi.waitFor(() => expect(stalledLookup.lookupFn).toHaveBeenCalledOnce());
+    controller.abort(abortError);
+    const outcome = await raceWithTimeoutResult(fetchPromise, 250, new Error("hung"));
+
+    expect(outcome).toBe(abortError);
+    expect(fetchImpl).not.toHaveBeenCalled();
+    stalledLookup.release();
+    await Promise.resolve();
+    expect(fetchImpl).not.toHaveBeenCalled();
+  });
+
+  it("applies timeoutMs while DNS preflight is stalled", async () => {
+    const stalledLookup = createStalledLookup();
+    const fetchImpl = vi.fn(async () => okResponse());
+    const fetchPromise = fetchWithSsrFGuard({
+      url: "https://public.example/resource",
+      fetchImpl,
+      lookupFn: stalledLookup.lookupFn,
+      timeoutMs: 1,
+    }).catch((error: unknown) => error);
+
+    const outcome = await raceWithTimeoutResult(fetchPromise, 250, new Error("hung"));
+
+    expect(outcome).toMatchObject({ name: "TimeoutError" });
+    expect(fetchImpl).not.toHaveBeenCalled();
+    stalledLookup.release();
+    await Promise.resolve();
+    expect(fetchImpl).not.toHaveBeenCalled();
+  });
+
  it("inherits the configured global stream timeout for guarded direct dispatchers", async () => {
    try {
      ensureGlobalUndiciStreamTimeouts({ timeoutMs: 1_900_000 });
--- a/src/infra/net/fetch-guard.ts
+++ b/src/infra/net/fetch-guard.ts
@@ -3,6 +3,8 @@
 import type { Dispatcher } from "undici";
 import { logWarn } from "../../logger.js";
 import { buildTimeoutAbortSignal } from "../../utils/fetch-timeout.js";
+import { createAbortError } from "../abort-signal.js";
+import { toErrorObject } from "../errors.js";
 import {
  normalizeHeadersInitForFetch,
  normalizeRequestInitHeadersForFetch,
@@ -121,6 +123,44 @@ type GuardedFetchPresetOptions = Omit<
 const DEFAULT_MAX_REDIRECTS = 3;
 const OPENCLAW_DEBUG_PROXY_ENABLED = "OPENCLAW_DEBUG_PROXY_ENABLED";

+async function runAbortablePreflight<T>(run: () => Promise<T>, signal?: AbortSignal): Promise<T> {
+  if (!signal) {
+    return await run();
+  }
+  if (signal.aborted) {
+    throw signal.reason ?? createAbortError("Guarded fetch aborted during network preflight");
+  }
+  return await new Promise<T>((resolve, reject) => {
+    let settled = false;
+    const settle = (complete: () => void) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      signal.removeEventListener("abort", onAbort);
+      complete();
+    };
+    const onAbort = () =>
+      settle(() =>
+        reject(
+          toErrorObject(
+            signal.reason ?? createAbortError("Guarded fetch aborted during network preflight"),
+            "Guarded fetch aborted during network preflight",
+          ),
+        ),
+      );
+    signal.addEventListener("abort", onAbort, { once: true });
+    if (signal.aborted) {
+      onAbort();
+      return;
+    }
+    void run().then(
+      (value) => settle(() => resolve(value)),
+      (error: unknown) => settle(() => reject(toErrorObject(error, "Network preflight failed"))),
+    );
+  });
+}
+
 function getRedirectVisitKey(url: string, init: RequestInit | undefined): string {
  return `${init?.method?.toUpperCase() ?? "GET"} ${url}`;
 }
@@ -216,6 +256,7 @@ async function assertExplicitProxyAllowed(
  dispatcherPolicy: PinnedDispatcherPolicy | undefined,
  lookupFn: LookupFn | undefined,
  policy: SsrFPolicy | undefined,
+  signal: AbortSignal | undefined,
 ): Promise<void> {
  // Explicit proxies are operator-configured, but the proxy host still needs
  // basic URL and private-network validation before target validation proceeds.
@@ -242,10 +283,14 @@ async function assertExplicitProxyAllowed(
          ...(dispatcherPolicy.allowPrivateProxy === true ? { allowPrivateNetwork: true } : {}),
        }
      : undefined;
-  await resolvePinnedHostnameWithPolicy(parsedProxyUrl.hostname, {
-    lookupFn,
-    policy: proxyPolicy,
-  });
+  await runAbortablePreflight(
+    async () =>
+      await resolvePinnedHostnameWithPolicy(parsedProxyUrl.hostname, {
+        lookupFn,
+        policy: proxyPolicy,
+      }),
+    signal,
+  );
 }

 function isRedirectStatus(status: number): boolean {
@@ -493,6 +538,15 @@ async function fetchWithSsrFGuardInternal(
    // Resolve inside the redirect loop so exact-origin trust never carries across origins.
    const policyForUrl = resolveSsrFPolicyForUrl(parsedUrl, params.policy);
    const dispatcherPolicy = params.resolveDispatcherPolicy?.(parsedUrl) ?? params.dispatcherPolicy;
+    const resolvePinnedHostname = async () =>
+      await runAbortablePreflight(
+        async () =>
+          await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, {
+            lookupFn: params.lookupFn,
+            policy: policyForUrl,
+          }),
+        signal,
+      );
    try {
      const usesTrustedExplicitProxyMode =
        mode === GUARDED_FETCH_MODE.TRUSTED_EXPLICIT_PROXY &&
@@ -502,7 +556,7 @@ async function fetchWithSsrFGuardInternal(
        dispatcherPolicy,
        usesTrustedExplicitProxyMode ? false : params.pinDns,
      );
-      await assertExplicitProxyAllowed(dispatcherPolicy, params.lookupFn, params.policy);
+      await assertExplicitProxyAllowed(dispatcherPolicy, params.lookupFn, params.policy, signal);
      const isStrictManagedProxyActive =
        mode === GUARDED_FETCH_MODE.STRICT && isManagedProxyActive();
      const shouldCheckManagedProxyBypass =
@@ -538,10 +592,7 @@ async function fetchWithSsrFGuardInternal(
        dispatcher = createHttp1EnvHttpProxyAgent(undefined, timeoutMs);
      } else if (canUseManagedProxy) {
        if (shouldCheckManagedProxyBypass) {
-          const pinned = await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, {
-            lookupFn: params.lookupFn,
-            policy: policyForUrl,
-          });
+          const pinned = await resolvePinnedHostname();
          dispatcher = shouldUseConfiguredLocalOriginManagedProxyBypass({
            url: parsedUrl,
            managedProxyBypass: params.managedProxyBypass,
@@ -562,16 +613,10 @@ async function fetchWithSsrFGuardInternal(
        // real fetches continue through pinned DNS below.
        assertHostnameAllowedWithPolicy(parsedUrl.hostname, policyForUrl);
      } else if (params.pinDns === false) {
-        await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, {
-          lookupFn: params.lookupFn,
-          policy: policyForUrl,
-        });
+        await resolvePinnedHostname();
        dispatcher = createPolicyDispatcherWithoutPinnedDns(dispatcherPolicy, timeoutMs);
      } else {
-        const pinned = await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, {
-          lookupFn: params.lookupFn,
-          policy: policyForUrl,
-        });
+        const pinned = await resolvePinnedHostname();
        dispatcher = createPinnedDispatcher(pinned, dispatcherPolicy, policyForUrl, timeoutMs);
      }

--- a/src/media/fetch.test.ts
+++ b/src/media/fetch.test.ts
@@ -69,6 +69,12 @@ function makeLookupFn(): LookupFn {
  return vi.fn(async () => ({ address: "149.154.167.220", family: 4 })) as unknown as LookupFn;
 }

+function abortReasonError(signal?: AbortSignal | null): Error {
+  return signal?.reason instanceof Error
+    ? signal.reason
+    : new Error("request aborted", { cause: signal?.reason });
+}
+
 function requireFetchGuardRequest(): unknown {
  const [call] = fetchWithSsrFGuardMock.mock.calls;
  if (!call) {
@@ -233,6 +239,7 @@ describe("readRemoteMediaBuffer", () => {
        url: string;
        fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
        init?: RequestInit;
+        signal?: AbortSignal;
      };
      if (params.url.startsWith("http://127.0.0.1/")) {
        throw new Error("Blocked hostname or private/internal/special-use IP address");
@@ -242,7 +249,10 @@ describe("readRemoteMediaBuffer", () => {
        throw new Error("fetch is not available");
      }
      return {
-        response: await fetcher(params.url, params.init),
+        response: await fetcher(params.url, {
+          ...params.init,
+          ...(params.signal ? { signal: params.signal } : {}),
+        }),
        finalUrl: params.url,
        release: async () => {},
      };
@@ -392,6 +402,155 @@ describe("readRemoteMediaBuffer", () => {
    });
  });

+  it("aborts when response headers exceed their deadline", async () => {
+    vi.useFakeTimers();
+    try {
+      const fetchImpl = vi.fn(
+        async (_input: RequestInfo | URL, init?: RequestInit) =>
+          await new Promise<Response>((_resolve, reject) => {
+            const signal = init?.signal;
+            const rejectForAbort = () => reject(abortReasonError(signal));
+            if (signal?.aborted) {
+              rejectForAbort();
+              return;
+            }
+            signal?.addEventListener("abort", rejectForAbort, { once: true });
+          }),
+      );
+
+      const result = readRemoteMediaBuffer({
+        url: "https://example.com/file.bin",
+        fetchImpl,
+        lookupFn: makeLookupFn(),
+        maxBytes: 1024,
+        responseHeaderTimeoutMs: 20,
+      }).catch((error: unknown) => error);
+
+      await vi.advanceTimersByTimeAsync(25);
+
+      await expect(result).resolves.toMatchObject({
+        name: "MediaFetchError",
+        code: "fetch_failed",
+        cause: { name: "TimeoutError" },
+      });
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("clears the response-header deadline while a healthy body keeps progressing", async () => {
+    vi.useFakeTimers();
+    try {
+      const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => {
+        const signal = init?.signal;
+        return new Response(
+          new ReadableStream<Uint8Array>({
+            start(controller) {
+              const failForAbort = () => controller.error(signal?.reason);
+              if (signal?.aborted) {
+                failForAbort();
+                return;
+              }
+              signal?.addEventListener("abort", failForAbort, { once: true });
+              setTimeout(() => controller.enqueue(new Uint8Array([1])), 25);
+              setTimeout(() => controller.enqueue(new Uint8Array([2])), 50);
+              setTimeout(() => {
+                signal?.removeEventListener("abort", failForAbort);
+                controller.close();
+              }, 75);
+            },
+          }),
+          { status: 200 },
+        );
+      });
+
+      const result = readRemoteMediaBuffer({
+        url: "https://example.com/file.bin",
+        fetchImpl,
+        lookupFn: makeLookupFn(),
+        maxBytes: 1024,
+        responseHeaderTimeoutMs: 10,
+        readIdleTimeoutMs: 30,
+      });
+
+      await vi.advanceTimersByTimeAsync(80);
+
+      await expect(result).resolves.toMatchObject({ buffer: Buffer.from([1, 2]) });
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("propagates a parent abort while waiting for response headers", async () => {
+    const parent = new AbortController();
+    const fetchImpl = vi.fn(
+      async (_input: RequestInfo | URL, init?: RequestInit) =>
+        await new Promise<Response>((_resolve, reject) => {
+          const signal = init?.signal;
+          const rejectForAbort = () => reject(abortReasonError(signal));
+          if (signal?.aborted) {
+            rejectForAbort();
+            return;
+          }
+          signal?.addEventListener("abort", rejectForAbort, { once: true });
+        }),
+    );
+    const result = readRemoteMediaBuffer({
+      url: "https://example.com/file.bin",
+      fetchImpl,
+      requestInit: { signal: parent.signal },
+      lookupFn: makeLookupFn(),
+      maxBytes: 1024,
+      responseHeaderTimeoutMs: 60_000,
+    }).catch((error: unknown) => error);
+
+    parent.abort();
+
+    await expect(result).resolves.toMatchObject({
+      name: "MediaFetchError",
+      code: "fetch_failed",
+      cause: { name: "AbortError" },
+    });
+  });
+
+  it("keeps the parent abort active while reading the response body", async () => {
+    const parent = new AbortController();
+    const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => {
+      const signal = init?.signal;
+      return new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new Uint8Array([1]));
+            const failForAbort = () => controller.error(signal?.reason);
+            if (signal?.aborted) {
+              failForAbort();
+              return;
+            }
+            signal?.addEventListener("abort", failForAbort, { once: true });
+          },
+        }),
+        { status: 200 },
+      );
+    });
+    const result = readRemoteMediaBuffer({
+      url: "https://example.com/file.bin",
+      fetchImpl,
+      requestInit: { signal: parent.signal },
+      lookupFn: makeLookupFn(),
+      maxBytes: 1024,
+      responseHeaderTimeoutMs: 60_000,
+    }).catch((error: unknown) => error);
+
+    await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1));
+    parent.abort();
+
+    await expect(result).resolves.toMatchObject({
+      name: "MediaFetchError",
+      code: "fetch_failed",
+      cause: { name: "AbortError" },
+    });
+  });
+
  it("retries transient fetch failures when retry is enabled", async () => {
    const transientError = Object.assign(new Error("socket hang up"), { code: "ECONNRESET" });
    const fetchImpl = vi
@@ -614,10 +773,12 @@ describe("readRemoteMediaBuffer", () => {

  it("passes request timeout through the guarded fetch path", async () => {
    const fetchImpl = vi.fn(async () => new Response("ok", { status: 200 }));
+    const parent = new AbortController();

    await readRemoteMediaBuffer({
      url: "https://example.com/file.bin",
      fetchImpl,
+      requestInit: { signal: parent.signal },
      lookupFn: makeLookupFn(),
      maxBytes: 1024,
      timeoutMs: 1234,
@@ -627,6 +788,7 @@ describe("readRemoteMediaBuffer", () => {
    expect(requireFetchGuardRequest()).toMatchObject({
      url: "https://example.com/file.bin",
      timeoutMs: 1234,
+      signal: parent.signal,
    });
  });

@@ -656,6 +818,92 @@ describe("readRemoteMediaBuffer", () => {
    await expect(fs.readFile(saved.path)).resolves.toStrictEqual(Buffer.from([1, 2, 3, 4]));
  });

+  it("keeps saving a healthy streaming body after the response-header deadline", async () => {
+    vi.useFakeTimers();
+    try {
+      const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => {
+        const signal = init?.signal;
+        return new Response(
+          new ReadableStream<Uint8Array>({
+            start(controller) {
+              const failForAbort = () => controller.error(signal?.reason);
+              if (signal?.aborted) {
+                failForAbort();
+                return;
+              }
+              signal?.addEventListener("abort", failForAbort, { once: true });
+              setTimeout(() => controller.enqueue(new Uint8Array([1])), 25);
+              setTimeout(() => controller.enqueue(new Uint8Array([2])), 50);
+              setTimeout(() => {
+                signal?.removeEventListener("abort", failForAbort);
+                controller.close();
+              }, 75);
+            },
+          }),
+          { status: 200, headers: { "content-type": "application/octet-stream" } },
+        );
+      });
+      const result = saveRemoteMedia({
+        url: "https://example.com/download",
+        fetchImpl,
+        lookupFn: makeLookupFn(),
+        maxBytes: 8,
+        responseHeaderTimeoutMs: 10,
+        readIdleTimeoutMs: 30,
+      });
+
+      await vi.advanceTimersByTimeAsync(80);
+
+      const saved = await result;
+      await expect(fs.readFile(saved.path)).resolves.toStrictEqual(Buffer.from([1, 2]));
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("keeps the parent abort active while saving the response body", async () => {
+    const parent = new AbortController();
+    let bodyStarted!: () => void;
+    const bodyReady = new Promise<void>((resolve) => {
+      bodyStarted = resolve;
+    });
+    const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => {
+      const signal = init?.signal;
+      return new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new Uint8Array([1]));
+            const failForAbort = () => controller.error(signal?.reason);
+            if (signal?.aborted) {
+              failForAbort();
+              return;
+            }
+            signal?.addEventListener("abort", failForAbort, { once: true });
+            bodyStarted();
+          },
+        }),
+        { status: 200 },
+      );
+    });
+    const result = saveRemoteMedia({
+      url: "https://example.com/download",
+      fetchImpl,
+      requestInit: { signal: parent.signal },
+      lookupFn: makeLookupFn(),
+      maxBytes: 8,
+      responseHeaderTimeoutMs: 60_000,
+    }).catch((error: unknown) => error);
+
+    await bodyReady;
+    parent.abort();
+
+    await expect(result).resolves.toMatchObject({
+      name: "MediaFetchError",
+      code: "fetch_failed",
+      cause: { name: "AbortError" },
+    });
+  });
+
  it("clamps oversized saved-response idle timeout timers", async () => {
    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
    try {
--- a/src/media/fetch.ts
+++ b/src/media/fetch.ts
@@ -3,7 +3,7 @@ import { MAX_DOCUMENT_BYTES } from "@openclaw/media-core/constants";
 import { parseMediaContentLength } from "@openclaw/media-core/content-length";
 import { basenameFromAnyPath, extnameFromAnyPath } from "@openclaw/media-core/file-name";
 import { detectMime, extensionForMime } from "@openclaw/media-core/mime";
-import { isAbortError } from "../infra/abort-signal.js";
+import { isAbortError, mergeAbortSignals } from "../infra/abort-signal.js";
 import { formatErrorMessage } from "../infra/errors.js";
 import {
  readChunkWithIdleTimeout,
@@ -19,6 +19,7 @@ import type { LookupFn, PinnedDispatcherPolicy, SsrFPolicy } from "../infra/net/
 import { retryAsync, type RetryOptions } from "../infra/retry.js";
 import { isTransientNetworkError } from "../infra/unhandled-rejections.js";
 import { redactSensitiveText } from "../logging/redact.js";
+import { buildTimeoutAbortSignal } from "../utils/fetch-timeout.js";
 import { saveMediaBuffer, saveMediaStream, type SavedMedia } from "./store.js";

 /** Default remote media fetch cap shared by buffer reads and store writes. */
@@ -75,8 +76,10 @@ type FetchMediaOptions = {
  filePathHint?: string;
  maxBytes?: number;
  maxRedirects?: number;
-  /** Abort the guarded fetch request if it has not completed by this deadline (ms). */
+  /** Abort the complete guarded fetch and body operation after this deadline (ms). */
  timeoutMs?: number;
+  /** Abort if final response headers have not arrived by this deadline (ms). */
+  responseHeaderTimeoutMs?: number;
  /** Abort if the response body stops yielding data for this long (ms). */
  readIdleTimeoutMs?: number;
  ssrfPolicy?: SsrFPolicy;
@@ -189,6 +192,7 @@ async function fetchGuardedMediaResponse(
    requestInit,
    maxRedirects,
    timeoutMs,
+    responseHeaderTimeoutMs,
    ssrfPolicy,
    lookupFn,
    dispatcherPolicy,
@@ -203,6 +207,15 @@ async function fetchGuardedMediaResponse(
    dispatcherAttempts && dispatcherAttempts.length > 0
      ? dispatcherAttempts
      : [{ dispatcherPolicy, lookupFn }];
+  const responseHeaderDeadline = buildTimeoutAbortSignal({
+    timeoutMs: responseHeaderTimeoutMs,
+    operation: "media response headers",
+    url,
+  });
+  const requestSignal = mergeAbortSignals([
+    requestInit?.signal ?? undefined,
+    responseHeaderDeadline.signal,
+  ]);
  const runGuardedFetch = async (attempt: FetchDispatcherAttempt) =>
    await fetchWithSsrFGuard(
      (trustExplicitProxyDns && attempt.dispatcherPolicy?.mode === "explicit-proxy"
@@ -213,6 +226,7 @@ async function fetchGuardedMediaResponse(
        init: requestInit,
        maxRedirects,
        ...(timeoutMs !== undefined ? { timeoutMs } : {}),
+        ...(requestSignal.signal ? { signal: requestSignal.signal } : {}),
        policy: ssrfPolicy,
        lookupFn: attempt.lookupFn ?? lookupFn,
        dispatcherPolicy: attempt.dispatcherPolicy,
@@ -253,13 +267,21 @@ async function fetchGuardedMediaResponse(
        attemptErrors.push(err);
      }
    }
+    // Clear only the header timer. The merged parent signal stays attached until
+    // release so shutdown can still interrupt a response body read.
+    responseHeaderDeadline.cleanup();
    return {
      response: result.response,
      finalUrl: result.finalUrl,
-      release: result.release,
+      release: async () => {
+        requestSignal.dispose();
+        await result.release();
+      },
      sourceUrl,
    };
  } catch (err) {
+    responseHeaderDeadline.cleanup();
+    requestSignal.dispose();
    throw new MediaFetchError(
      "fetch_failed",
      `Failed to fetch media from ${sourceUrl}: ${formatErrorMessage(err)}`,