fix(line): prevent unbounded provider response buffering (#119099)

* fix(line): bound provider response bodies

* fix(line): preserve provider status on body errors

* test(line): keep response bounds coverage under lint budget

* test(line): cover bounded retry conflict responses

* chore(line): shrink assertion safety baseline

config/assertion-safety-baseline.txt

@@ -630,7 +630,7 @@ extensions/line/src/monitor-durable.ts	1
 extensions/line/src/monitor.ts	1
 extensions/line/src/outbound.ts	6
 extensions/line/src/rich-menu.ts	1
-extensions/line/src/send.ts	2
+extensions/line/src/send.ts	1
 extensions/line/src/setup-core.ts	1
 extensions/line/src/webhook-spool.ts	4
 extensions/line/src/webhook-utils.ts	1

extensions/line/src/send.response-bounds.test.ts

@@ -0,0 +1,175 @@
+import { HTTPFetchError } from "@line/bot-sdk";
+import { isChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+
+const {
+  lineFetchMock,
+  requireRuntimeConfigMock,
+  resolveLineAccountMock,
+  resolveLineChannelAccessTokenMock,
+  recordChannelActivityMock,
+} = vi.hoisted(() => ({
+  lineFetchMock: vi.fn<typeof fetch>(),
+  requireRuntimeConfigMock: vi.fn((cfg: unknown) => cfg ?? {}),
+  resolveLineAccountMock: vi.fn(() => ({ accountId: "default" })),
+  resolveLineChannelAccessTokenMock: vi.fn(() => "line-token"),
+  recordChannelActivityMock: vi.fn(),
+}));
+
+vi.mock("openclaw/plugin-sdk/plugin-config-runtime", () => ({
+  requireRuntimeConfig: requireRuntimeConfigMock,
+}));
+vi.mock("./accounts.js", () => ({ resolveLineAccount: resolveLineAccountMock }));
+vi.mock("./channel-access-token.js", () => ({
+  resolveLineChannelAccessToken: resolveLineChannelAccessTokenMock,
+}));
+vi.mock("openclaw/plugin-sdk/channel-activity-runtime", () => ({
+  recordChannelActivity: recordChannelActivityMock,
+}));
+
+let sendModule: typeof import("./send.js");
+const LINE_TEST_CFG = { channels: { line: { accounts: { default: {} } } } };
+
+function createTrackedResponse(body: string, init: ResponseInit) {
+  let canceled = false;
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(new TextEncoder().encode(body));
+    },
+    cancel() {
+      canceled = true;
+    },
+  });
+  return { response: new Response(stream, init), wasCanceled: () => canceled };
+}
+
+async function captureError(run: () => Promise<unknown>): Promise<unknown> {
+  try {
+    await run();
+  } catch (error) {
+    return error;
+  }
+  throw new Error("expected LINE send to fail");
+}
+
+describe("LINE bounded provider responses", () => {
+  beforeAll(async () => {
+    sendModule = await import("./send.js");
+  });
+
+  afterAll(() => {
+    vi.doUnmock("openclaw/plugin-sdk/plugin-config-runtime");
+    vi.doUnmock("./accounts.js");
+    vi.doUnmock("./channel-access-token.js");
+    vi.doUnmock("openclaw/plugin-sdk/channel-activity-runtime");
+    vi.resetModules();
+  });
+
+  beforeEach(() => {
+    lineFetchMock.mockReset();
+    requireRuntimeConfigMock.mockClear().mockImplementation((cfg: unknown) => cfg ?? LINE_TEST_CFG);
+    resolveLineAccountMock.mockReset().mockReturnValue({ accountId: "default" });
+    resolveLineChannelAccessTokenMock.mockReset().mockReturnValue("line-token");
+    recordChannelActivityMock.mockReset();
+    vi.stubGlobal("fetch", lineFetchMock);
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  it("preserves partial delivery when an accepted LINE response is oversized", async () => {
+    const tracked = createTrackedResponse("x".repeat(16 * 1024 + 1), {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    });
+    const textSpy = vi.spyOn(tracked.response, "text").mockRejectedValue(new Error("unbounded"));
+    lineFetchMock.mockResolvedValueOnce(tracked.response);
+
+    const caught = await captureError(() =>
+      sendModule.pushMessageLine("U123", "Hello", { cfg: LINE_TEST_CFG }),
+    );
+
+    expect(isChannelPartialDeliveryError(caught)).toBe(true);
+    if (!isChannelPartialDeliveryError(caught)) {
+      throw new Error("expected an accepted LINE delivery without a readable receipt");
+    }
+    expect(caught.deliveryResult).toEqual({ messageIds: [], visibleReplySent: true });
+    expect(textSpy).not.toHaveBeenCalled();
+    expect(tracked.wasCanceled()).toBe(true);
+  });
+
+  it("bounds an accepted retry-key conflict without reclassifying it as rejected", async () => {
+    const tracked = createTrackedResponse("x".repeat(16 * 1024 + 1), {
+      status: 409,
+      statusText: "Conflict",
+      headers: { "content-type": "application/json" },
+    });
+    const textSpy = vi.spyOn(tracked.response, "text").mockRejectedValue(new Error("unbounded"));
+    lineFetchMock.mockResolvedValueOnce(tracked.response);
+
+    const caught = await captureError(() =>
+      sendModule.pushMessageLine("U123", "Hello", { cfg: LINE_TEST_CFG }),
+    );
+
+    expect(isChannelPartialDeliveryError(caught)).toBe(true);
+    expect(caught).not.toBeInstanceOf(HTTPFetchError);
+    expect(lineFetchMock).toHaveBeenCalledOnce();
+    const requestInit = lineFetchMock.mock.calls[0]?.[1];
+    expect(new Headers(requestInit?.headers).get("X-Line-Retry-Key")).toBeTruthy();
+    expect(textSpy).not.toHaveBeenCalled();
+    expect(tracked.wasCanceled()).toBe(true);
+  });
+
+  it("bounds oversized rejected LINE response bodies", async () => {
+    const tracked = createTrackedResponse(`${"line upstream unavailable ".repeat(1024)}tail`, {
+      status: 400,
+      statusText: "Bad Request",
+      headers: { "content-type": "text/plain" },
+    });
+    const textSpy = vi.spyOn(tracked.response, "text").mockRejectedValue(new Error("unbounded"));
+    lineFetchMock.mockResolvedValueOnce(tracked.response);
+
+    const caught = await captureError(() =>
+      sendModule.pushMessageLine("U123", "Hello", { cfg: LINE_TEST_CFG }),
+    );
+
+    expect(caught).toBeInstanceOf(HTTPFetchError);
+    expect(isChannelPartialDeliveryError(caught)).toBe(false);
+    expect(caught).toMatchObject({ status: 400, statusText: "Bad Request" });
+    expect((caught as HTTPFetchError).body).toContain("line upstream unavailable");
+    expect((caught as HTTPFetchError).body).not.toContain("tail");
+    expect(textSpy).not.toHaveBeenCalled();
+    expect(tracked.wasCanceled()).toBe(true);
+  });
+
+  it("preserves reply rejection status when the LINE error body cannot be read", async () => {
+    const response = new Response(
+      new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.error(new Error("provider response body failed"));
+        },
+      }),
+      { status: 503, statusText: "Service Unavailable", headers: { "content-type": "text/plain" } },
+    );
+    const textSpy = vi.spyOn(response, "text").mockRejectedValue(new Error("unbounded"));
+    lineFetchMock.mockResolvedValueOnce(response);
+
+    const caught = await captureError(() =>
+      sendModule.sendMessageLine("U123", "Hello", {
+        cfg: LINE_TEST_CFG,
+        replyToken: "reply-token",
+      }),
+    );
+
+    expect(caught).toBeInstanceOf(HTTPFetchError);
+    expect(isChannelPartialDeliveryError(caught)).toBe(false);
+    expect(caught).toMatchObject({
+      status: 503,
+      statusText: "Service Unavailable",
+      body: "",
+    });
+    expect(lineFetchMock).toHaveBeenCalledOnce();
+    expect(textSpy).not.toHaveBeenCalled();
+  });
+});

extensions/line/src/send.ts

@@ -7,6 +7,10 @@ import { createChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-i
 import { pruneMapToMaxSize } from "openclaw/plugin-sdk/collection-runtime";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import { requireRuntimeConfig } from "openclaw/plugin-sdk/plugin-config-runtime";
+import {
+  readProviderJsonResponse,
+  readResponseTextLimited,
+} from "openclaw/plugin-sdk/provider-http";
 import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
 import { fetchWithRuntimeDispatcherOrMockedGlobal } from "openclaw/plugin-sdk/runtime-fetch";
 import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime";
@@ -38,6 +42,9 @@ const PROFILE_CACHE_TTL_MS = 5 * 60 * 1000;
 const PROFILE_CACHE_MAX_ENTRIES = 1000;
 const LINE_FLEX_ALT_TEXT_LIMIT = 1500;
 const LINE_LOCATION_LABEL_LIMIT = 100;
+// This cap bounds receipts and diagnostics: overflow after acceptance becomes no-retry partial
+// delivery, while rejected responses keep their status with prefix-only diagnostics.
+const LINE_PROVIDER_RESPONSE_MAX_BYTES = 16 * 1024;
 
 function cacheUserProfile(
   userId: string,
@@ -197,19 +204,23 @@ async function sendLineProviderMessages(
   const acceptedRetryConflict = retryKey !== undefined && response.status === 409;
 
   if (!response.ok && !acceptedRetryConflict) {
+    const body = await readResponseTextLimited(response, LINE_PROVIDER_RESPONSE_MAX_BYTES).catch(
+      () => "",
+    );
     throw new HTTPFetchError(`${response.status} - ${response.statusText}`, {
       status: response.status,
       statusText: response.statusText,
       headers: response.headers,
-      body: await response.text(),
+      body,
     });
   }
 
   try {
-    const text = await response.text();
-    return (text ? JSON.parse(text) : null) as
-      | messagingApi.PushMessageResponse
-      | messagingApi.ReplyMessageResponse;
+    return await readProviderJsonResponse<
+      messagingApi.PushMessageResponse | messagingApi.ReplyMessageResponse
+    >(response, `LINE ${operation} response`, {
+      maxBytes: LINE_PROVIDER_RESPONSE_MAX_BYTES,
+    });
   } catch (error) {
     // LINE accepted this exact request before its receipt became unreadable; retrying duplicates it.
     throw createChannelPartialDeliveryError(error, { messageIds: [], visibleReplySent: true });