diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index fc02a56e8d19..b574e465703c 100644 --- a/config/assertion-safety-baseline.txt +++ b/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 diff --git a/extensions/line/src/send.response-bounds.test.ts b/extensions/line/src/send.response-bounds.test.ts new file mode 100644 index 000000000000..3f12c242ca6c --- /dev/null +++ b/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(), + 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({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)); + }, + cancel() { + canceled = true; + }, + }); + return { response: new Response(stream, init), wasCanceled: () => canceled }; +} + +async function captureError(run: () => Promise): Promise { + 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({ + 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(); + }); +}); diff --git a/extensions/line/src/send.ts b/extensions/line/src/send.ts index af8c9b085547..2936dd80d483 100644 --- a/extensions/line/src/send.ts +++ b/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 });