fix(line): retry inbound media download while LINE reports 202 (still preparing) (#108351)

* fix(line): retry inbound media download while LINE reports 202

LINE answers the content endpoint with 202 and an empty body for a
short window while it prepares inbound media. The previous single-shot
download saved that empty body as a 0-byte file, silently dropping the
user's media. Poll the content endpoint with capped exponential
backoff until it stops returning 202, releasing each empty response
before retrying, and throw a visible error if it never becomes ready.

* fix(line): clean up media retry responses

* fix(line): use plugin runtime fetch seam

---------

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

extensions/line/src/download.test.ts

@@ -1,19 +1,27 @@
 // Line tests cover download plugin behavior.
-import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 
-const getMessageContentMock = vi.hoisted(() => vi.fn());
+const fetchMock = vi.hoisted(() => vi.fn());
+const delayMock = vi.hoisted(() => vi.fn());
 const saveMediaStreamMock = vi.hoisted(() => vi.fn());
 
-vi.mock("@line/bot-sdk", () => ({
-  messagingApi: {
-    MessagingApiBlobClient: class {
-      getMessageContent(messageId: string) {
-        return getMessageContentMock(messageId);
-      }
-    },
-  },
+vi.mock("node:timers/promises", () => ({
+  setTimeout: delayMock,
 }));
 
+function responseWithChunks(status: number, parts: Buffer[]): Response {
+  return new Response(Buffer.concat(parts), { status });
+}
+
+function cancellableResponse(status: number): {
+  response: Response;
+  cancel: ReturnType<typeof vi.fn>;
+} {
+  const cancel = vi.fn();
+  const body = new ReadableStream<Uint8Array>({ cancel });
+  return { response: new Response(body, { status }), cancel };
+}
+
 vi.mock("openclaw/plugin-sdk/runtime-env", () => ({
   createSubsystemLogger: () => {
     const logger = {
@@ -34,12 +42,6 @@ vi.mock("openclaw/plugin-sdk/media-store", () => ({
 
 let downloadLineMedia: typeof import("./download.js").downloadLineMedia;
 
-async function* chunks(parts: Buffer[]): AsyncGenerator<Buffer> {
-  for (const part of parts) {
-    yield part;
-  }
-}
-
 function saveMediaStreamCall(): unknown[] {
   const call = saveMediaStreamMock.mock.calls.at(0);
   if (!call) {
@@ -64,15 +66,22 @@ describe("downloadLineMedia", () => {
   });
 
   afterAll(() => {
-    vi.doUnmock("@line/bot-sdk");
+    vi.doUnmock("node:timers/promises");
     vi.doUnmock("openclaw/plugin-sdk/runtime-env");
     vi.doUnmock("openclaw/plugin-sdk/media-store");
+    vi.unstubAllGlobals();
     vi.resetModules();
   });
 
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
   beforeEach(() => {
     vi.restoreAllMocks();
-    getMessageContentMock.mockReset();
+    vi.stubGlobal("fetch", fetchMock);
+    fetchMock.mockReset();
+    delayMock.mockReset().mockResolvedValue(undefined);
     saveMediaStreamMock.mockReset();
     saveMediaStreamMock.mockImplementation(
       async (stream: AsyncIterable<Buffer>, contentType?: string, subdir?: string) => {
@@ -92,10 +101,18 @@ describe("downloadLineMedia", () => {
 
   it("persists inbound media with the shared media store", async () => {
     const jpeg = Buffer.from([0xff, 0xd8, 0xff, 0x00]);
-    getMessageContentMock.mockResolvedValueOnce(chunks([jpeg]));
+    fetchMock.mockResolvedValueOnce(responseWithChunks(200, [jpeg]));
 
     const result = await downloadLineMedia("mid-jpeg", "token");
 
+    expect(fetchMock).toHaveBeenCalledWith(
+      "https://api-data.line.me/v2/bot/message/mid-jpeg/content",
+      expect.objectContaining({
+        headers: { Authorization: "Bearer token" },
+        redirect: "error",
+        signal: expect.any(AbortSignal),
+      }),
+    );
     expect(saveMediaStreamMock).toHaveBeenCalledTimes(1);
     const call = saveMediaStreamCall();
     expect(call[1]).toBeUndefined();
@@ -111,10 +128,13 @@ describe("downloadLineMedia", () => {
   it("does not pass the external messageId to saveMediaStream", async () => {
     const messageId = "a/../../../../etc/passwd";
     const jpeg = Buffer.from([0xff, 0xd8, 0xff, 0x00]);
-    getMessageContentMock.mockResolvedValueOnce(chunks([jpeg]));
+    fetchMock.mockResolvedValueOnce(responseWithChunks(200, [jpeg]));
 
     const result = await downloadLineMedia(messageId, "token");
 
+    expect(fetchMock.mock.calls[0]?.[0]).toBe(
+      "https://api-data.line.me/v2/bot/message/a%2F..%2F..%2F..%2F..%2Fetc%2Fpasswd/content",
+    );
     expect(result.size).toBe(jpeg.length);
     expect(result.contentType).toBe("image/jpeg");
     for (const arg of saveMediaStreamCall()) {
@@ -124,19 +144,21 @@ describe("downloadLineMedia", () => {
     }
   });
 
-  it("delegates oversized media rejection to saveMediaStream", async () => {
-    getMessageContentMock.mockResolvedValueOnce(chunks([Buffer.alloc(4), Buffer.alloc(4)]));
+  it("cancels content when the media store rejects it", async () => {
+    const content = cancellableResponse(200);
+    fetchMock.mockResolvedValueOnce(content.response);
     saveMediaStreamMock.mockRejectedValueOnce(new Error("Media exceeds 0MB limit"));
 
     await expect(downloadLineMedia("mid", "token", 7)).rejects.toThrow(/Media exceeds/i);
     expect(saveMediaStreamMock).toHaveBeenCalledTimes(1);
+    expect(content.cancel).toHaveBeenCalledTimes(1);
   });
 
   it("uses media store content type for M4A media", async () => {
     const m4aHeader = Buffer.from([
       0x00, 0x00, 0x00, 0x1c, 0x66, 0x74, 0x79, 0x70, 0x4d, 0x34, 0x41, 0x20,
     ]);
-    getMessageContentMock.mockResolvedValueOnce(chunks([m4aHeader]));
+    fetchMock.mockResolvedValueOnce(responseWithChunks(200, [m4aHeader]));
 
     const result = await downloadLineMedia("mid-audio", "token");
 
@@ -145,7 +167,7 @@ describe("downloadLineMedia", () => {
   });
 
   it("passes original filenames to the media store for extension fallback", async () => {
-    getMessageContentMock.mockResolvedValueOnce(chunks([Buffer.from("unknown-audio-bytes")]));
+    fetchMock.mockResolvedValueOnce(responseWithChunks(200, [Buffer.from("unknown-audio-bytes")]));
 
     await downloadLineMedia("mid-file-audio", "token", 10 * 1024 * 1024, {
       originalFilename: "voice-note.m4a",
@@ -160,18 +182,88 @@ describe("downloadLineMedia", () => {
     const mp4 = Buffer.from([
       0x00, 0x00, 0x00, 0x1c, 0x66, 0x74, 0x79, 0x70, 0x69, 0x73, 0x6f, 0x6d,
     ]);
-    getMessageContentMock.mockResolvedValueOnce(chunks([mp4]));
+    fetchMock.mockResolvedValueOnce(responseWithChunks(200, [mp4]));
 
     const result = await downloadLineMedia("mid-mp4", "token");
 
     expect(result.contentType).toBe("video/mp4");
   });
 
-  it("propagates media store failures", async () => {
-    const jpeg = Buffer.from([0xff, 0xd8, 0xff, 0x00]);
-    getMessageContentMock.mockResolvedValueOnce(chunks([jpeg]));
-    saveMediaStreamMock.mockRejectedValueOnce(new Error("Media exceeds 0MB limit"));
+  it("retries 202 responses and cancels every discarded body", async () => {
+    const m4aHeader = Buffer.from([
+      0x00, 0x00, 0x00, 0x1c, 0x66, 0x74, 0x79, 0x70, 0x4d, 0x34, 0x41, 0x20,
+    ]);
+    const first = cancellableResponse(202);
+    const second = cancellableResponse(202);
+    fetchMock
+      .mockResolvedValueOnce(first.response)
+      .mockResolvedValueOnce(second.response)
+      .mockResolvedValueOnce(responseWithChunks(200, [m4aHeader]));
 
-    await expect(downloadLineMedia("mid-bad", "token")).rejects.toThrow(/Media exceeds/i);
+    const result = await downloadLineMedia("mid-preparing", "token");
+
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    expect(delayMock).toHaveBeenNthCalledWith(1, 500, undefined, {
+      signal: expect.any(AbortSignal),
+    });
+    expect(delayMock).toHaveBeenNthCalledWith(2, 1000, undefined, {
+      signal: expect.any(AbortSignal),
+    });
+    expect(first.cancel).toHaveBeenCalledTimes(1);
+    expect(second.cancel).toHaveBeenCalledTimes(1);
+    expect(result.contentType).toBe("audio/x-m4a");
+    expect(result.size).toBe(m4aHeader.length);
+  });
+
+  it("cancels every response when content never becomes ready", async () => {
+    const attempts = Array.from({ length: 6 }, () => cancellableResponse(202));
+    for (const attempt of attempts) {
+      fetchMock.mockResolvedValueOnce(attempt.response);
+    }
+
+    await expect(downloadLineMedia("mid-stuck", "token")).rejects.toThrow(/still preparing/i);
+
+    expect(fetchMock).toHaveBeenCalledTimes(6);
+    expect(delayMock).toHaveBeenCalledTimes(5);
+    expect(delayMock.mock.calls.map((call) => call[0])).toEqual([500, 1000, 2000, 4000, 4000]);
+    for (const attempt of attempts) {
+      expect(attempt.cancel).toHaveBeenCalledTimes(1);
+    }
+    expect(saveMediaStreamMock).not.toHaveBeenCalled();
+  });
+
+  it("cancels error responses without retrying", async () => {
+    const response = cancellableResponse(404);
+    fetchMock.mockResolvedValueOnce(response.response);
+
+    await expect(downloadLineMedia("mid-missing", "token")).rejects.toThrow(/HTTP 404/i);
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(response.cancel).toHaveBeenCalledTimes(1);
+    expect(saveMediaStreamMock).not.toHaveBeenCalled();
+  });
+
+  it("aborts a hung content request at the total readiness deadline", async () => {
+    let requestSignal: AbortSignal | undefined;
+    fetchMock.mockImplementation(
+      async (_input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
+        requestSignal = init?.signal ?? undefined;
+        return await new Promise<Response>((_resolve, reject) => {
+          requestSignal?.addEventListener("abort", () => reject(new Error("fetch aborted")), {
+            once: true,
+          });
+        });
+      },
+    );
+
+    vi.useFakeTimers();
+    const pending = downloadLineMedia("mid-hung", "token");
+    const rejection = expect(pending).rejects.toThrow(/did not become ready within 15 seconds/i);
+    await vi.advanceTimersByTimeAsync(15_000);
+    await rejection;
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(requestSignal?.aborted).toBe(true);
+    expect(saveMediaStreamMock).not.toHaveBeenCalled();
   });
 });

extensions/line/src/download.ts

@@ -1,7 +1,11 @@
 // Line plugin module implements download behavior.
-import { messagingApi } from "@line/bot-sdk";
+import { Readable } from "node:stream";
+import { finished } from "node:stream/promises";
+import type { ReadableStream as NodeReadableStream } from "node:stream/web";
+import { setTimeout as delay } from "node:timers/promises";
 import { saveMediaStream } from "openclaw/plugin-sdk/media-store";
 import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
+import { fetchWithRuntimeDispatcherOrMockedGlobal } from "openclaw/plugin-sdk/runtime-fetch";
 
 interface DownloadResult {
   path: string;
@@ -9,24 +13,90 @@ interface DownloadResult {
   size: number;
 }
 
+// LINE prepares inbound media asynchronously. Poll the content endpoint itself
+// because the transcoding-status endpoint does not cover every media type.
+const CONTENT_READY_MAX_ATTEMPTS = 6;
+const CONTENT_READY_BASE_DELAY_MS = 500;
+const CONTENT_READY_MAX_DELAY_MS = 4000;
+const CONTENT_READY_TIMEOUT_MS = 15_000;
+const LINE_CONTENT_BASE_URL = "https://api-data.line.me/v2/bot/message";
+
+function contentBackoffDelayMs(attempt: number): number {
+  return Math.min(CONTENT_READY_BASE_DELAY_MS * 2 ** attempt, CONTENT_READY_MAX_DELAY_MS);
+}
+
+async function fetchLineContentWhenReady(
+  messageId: string,
+  channelAccessToken: string,
+): Promise<Readable> {
+  const controller = new AbortController();
+  const deadline = setTimeout(() => controller.abort(), CONTENT_READY_TIMEOUT_MS);
+  deadline.unref();
+  try {
+    for (let attempt = 0; attempt < CONTENT_READY_MAX_ATTEMPTS; attempt++) {
+      const response = await fetchWithRuntimeDispatcherOrMockedGlobal(
+        `${LINE_CONTENT_BASE_URL}/${encodeURIComponent(messageId)}/content`,
+        {
+          headers: { Authorization: `Bearer ${channelAccessToken}` },
+          redirect: "error",
+          signal: controller.signal,
+        },
+      );
+      if (response.status === 200) {
+        if (!response.body) {
+          throw new Error(`LINE media response for message ${messageId} had no body`);
+        }
+        return Readable.fromWeb(response.body as NodeReadableStream<Uint8Array>);
+      }
+
+      await response.body?.cancel();
+      if (response.status !== 202) {
+        throw new Error(
+          `LINE media download failed for message ${messageId} (HTTP ${response.status})`,
+        );
+      }
+      if (attempt < CONTENT_READY_MAX_ATTEMPTS - 1) {
+        await delay(contentBackoffDelayMs(attempt), undefined, { signal: controller.signal });
+      }
+    }
+  } catch (err) {
+    if (controller.signal.aborted) {
+      throw new Error(
+        `LINE media for message ${messageId} did not become ready within ${CONTENT_READY_TIMEOUT_MS / 1000} seconds`,
+        { cause: err },
+      );
+    }
+    throw err;
+  } finally {
+    clearTimeout(deadline);
+  }
+
+  throw new Error(
+    `LINE media for message ${messageId} was still preparing (HTTP 202) after ${CONTENT_READY_MAX_ATTEMPTS} attempts`,
+  );
+}
+
 export async function downloadLineMedia(
   messageId: string,
   channelAccessToken: string,
   maxBytes = 10 * 1024 * 1024,
   options?: { originalFilename?: string },
 ): Promise<DownloadResult> {
-  const client = new messagingApi.MessagingApiBlobClient({
-    channelAccessToken,
-  });
-
-  const response = await client.getMessageContent(messageId);
-  const saved = await saveMediaStream(
-    response as AsyncIterable<Buffer>,
-    undefined,
-    "inbound",
-    maxBytes,
-    options?.originalFilename,
-  );
+  const content = await fetchLineContentWhenReady(messageId, channelAccessToken);
+  let saved: Awaited<ReturnType<typeof saveMediaStream>>;
+  try {
+    saved = await saveMediaStream(
+      content,
+      undefined,
+      "inbound",
+      maxBytes,
+      options?.originalFilename,
+    );
+  } catch (err) {
+    content.destroy();
+    await finished(content).catch(() => undefined);
+    throw err;
+  }
   logVerbose(`line: persisted media ${messageId} to ${saved.path} (${saved.size} bytes)`);
 
   return {