diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 6f363bb6ac30..6c3e77bb461e 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -6e7f42fbac908f7d80fa6af6ae1d65c6550f1235df19e2170223dd469536a76b plugin-sdk-api-baseline.json -19c0e0458c782ce08f7405a789b99c28f93161b1667650890028feef3848d40b plugin-sdk-api-baseline.jsonl +14ff327d9f8d7d823d2391b857441a7b2e7092db4ae07b4c3b93f5972969ee53 plugin-sdk-api-baseline.json +533532f5fc00832ca36e708031923ad0a9ec246e178e0027ab5eb235960ff2a8 plugin-sdk-api-baseline.jsonl diff --git a/extensions/sms/src/twilio.test.ts b/extensions/sms/src/twilio.test.ts index 2bc05c4104c3..b6b2b0632019 100644 --- a/extensions/sms/src/twilio.test.ts +++ b/extensions/sms/src/twilio.test.ts @@ -494,9 +494,9 @@ describe("Twilio SMS helpers", () => { expect(release).toHaveBeenCalledTimes(1); }); - it("bounds and cancels oversized guarded Twilio error bodies", async () => { + it("bounds guarded Twilio errors on complete UTF-8 characters and cancels overflow", async () => { const release = vi.fn(async () => {}); - const tracked = cancelTrackedTextResponse(`${"upstream unavailable ".repeat(512)}tail`, { + const tracked = cancelTrackedTextResponse(`${"x".repeat(8 * 1024 - 2)}😀tail`, { status: 503, }); fetchWithSsrFGuardMock.mockResolvedValue({ @@ -515,8 +515,9 @@ describe("Twilio SMS helpers", () => { caught = error as Error; } - expect(caught?.message).toContain("Twilio SMS send failed (503): upstream unavailable"); + expect(caught?.message).toContain("Twilio SMS send failed (503): "); expect(caught?.message).toContain("... [truncated]"); + expect(caught?.message).not.toContain("�"); expect(caught?.message).not.toContain("tail"); expect(caught?.message.length).toBeLessThan(8_300); expect(tracked.wasCanceled()).toBe(true); @@ -556,7 +557,11 @@ describe("Twilio SMS helpers", () => { it("rejects malformed JSON from Twilio Messaging Service lookup", async () => { const fetchImpl = vi.fn( - async () => new Response("NOT JSON {{{", { status: 200, headers: { "content-type": "application/json" } }), + async () => + new Response("NOT JSON {{{", { + status: 200, + headers: { "content-type": "application/json" }, + }), ); await expect( @@ -570,7 +575,11 @@ describe("Twilio SMS helpers", () => { it("returns empty list on malformed JSON from Twilio incoming phone number list", async () => { const fetchImpl = vi.fn( - async () => new Response("NOT JSON {{{", { status: 200, headers: { "content-type": "application/json" } }), + async () => + new Response("NOT JSON {{{", { + status: 200, + headers: { "content-type": "application/json" }, + }), ); const result = await listTwilioIncomingPhoneNumbers({ diff --git a/extensions/sms/src/twilio.ts b/extensions/sms/src/twilio.ts index ecedbebf801b..4fe2b7bf3871 100644 --- a/extensions/sms/src/twilio.ts +++ b/extensions/sms/src/twilio.ts @@ -2,6 +2,10 @@ import { createHmac, timingSafeEqual } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import * as querystring from "node:querystring"; +import { + readResponseTextPrefix, + readResponseWithLimit, +} from "openclaw/plugin-sdk/response-limit-runtime"; import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime"; import { readRequestBodyWithLimit } from "openclaw/plugin-sdk/webhook-ingress"; import type { ResolvedSmsAccount, SmsInboundMessage, SmsSendResult } from "./types.js"; @@ -273,54 +277,19 @@ function appendTruncatedResponseSuffix(text: string): string { } async function readTwilioApiResponseText(response: Response): Promise { - if (!response.body) { - return ""; - } - const maxBytes = response.ok ? TWILIO_API_SUCCESS_BODY_LIMIT_BYTES : TWILIO_API_ERROR_BODY_LIMIT_BYTES; - const truncateOnLimit = !response.ok; - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let totalBytes = 0; - let text = ""; - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) { - return text + decoder.decode(); - } - if (!value?.byteLength) { - continue; - } - - const remainingBytes = maxBytes - totalBytes; - if (value.byteLength > remainingBytes) { - const clipped = remainingBytes > 0 ? value.slice(0, remainingBytes) : undefined; - if (truncateOnLimit) { - if (clipped) { - text += decoder.decode(clipped, { stream: true }); - } - await reader.cancel().catch(() => undefined); - return appendTruncatedResponseSuffix(text + decoder.decode()); - } - await reader.cancel().catch(() => undefined); - throw new Error( - `Twilio SMS API response body too large: ${totalBytes + value.byteLength} bytes ` + - `(limit: ${maxBytes} bytes)`, - ); - } - - text += decoder.decode(value, { stream: true }); - totalBytes += value.byteLength; - } - } finally { - try { - reader.releaseLock(); - } catch {} + if (!response.ok) { + const prefix = await readResponseTextPrefix(response, maxBytes); + return prefix.truncated ? appendTruncatedResponseSuffix(prefix.text) : prefix.text; } + + const body = await readResponseWithLimit(response, maxBytes, { + onOverflow: ({ size, maxBytes: limit }) => + new Error(`Twilio SMS API response body too large: ${size} bytes (limit: ${limit} bytes)`), + }); + return new TextDecoder().decode(body); } function normalizeRequestHeaders(headers: HeadersInit | undefined): Record { diff --git a/extensions/voice-call/src/providers/shared/guarded-json-api.test.ts b/extensions/voice-call/src/providers/shared/guarded-json-api.test.ts index 33559cf12bc0..9e16fe839950 100644 --- a/extensions/voice-call/src/providers/shared/guarded-json-api.test.ts +++ b/extensions/voice-call/src/providers/shared/guarded-json-api.test.ts @@ -131,9 +131,11 @@ describe("guardedJsonApiRequest", () => { expect(release).toHaveBeenCalledTimes(1); }); - it("bounds provider error bodies and cancels unread overflow", async () => { + it("bounds provider error bodies on complete UTF-8 characters and cancels overflow", async () => { const release = vi.fn(async () => {}); - const tracked = cancelTrackedTextResponse("x".repeat(9 * 1024), { status: 500 }); + const tracked = cancelTrackedTextResponse(`${"x".repeat(8 * 1024 - 2)}😀tail`, { + status: 500, + }); fetchWithSsrFGuardMock.mockResolvedValue({ response: tracked.response, release, @@ -155,6 +157,8 @@ describe("guardedJsonApiRequest", () => { expect(caught?.message).toContain("provider error: 500 "); expect(caught?.message).toContain("... [truncated]"); + expect(caught?.message).not.toContain("�"); + expect(caught?.message).not.toContain("tail"); expect(caught?.message.length).toBeLessThan(8_300); expect(tracked.wasCanceled()).toBe(true); expect(release).toHaveBeenCalledTimes(1); diff --git a/extensions/voice-call/src/providers/shared/response-body.ts b/extensions/voice-call/src/providers/shared/response-body.ts index 0d7b91d2b5fe..7c8bb5f1c7f9 100644 --- a/extensions/voice-call/src/providers/shared/response-body.ts +++ b/extensions/voice-call/src/providers/shared/response-body.ts @@ -1,4 +1,8 @@ // Voice Call provider HTTP clients share bounded response body readers. +import { + readResponseTextPrefix, + readResponseWithLimit, +} from "openclaw/plugin-sdk/response-limit-runtime"; const PROVIDER_JSON_RESPONSE_MAX_BYTES = 1 * 1024 * 1024; const PROVIDER_ERROR_RESPONSE_MAX_BYTES = 8 * 1024; @@ -21,50 +25,16 @@ function appendTruncatedSuffix(text: string): string { async function readProviderResponseTextWithLimit( params: ReadProviderResponseTextParams, ): Promise { - if (!params.response.body) { - return ""; + if (params.truncateOnLimit) { + const prefix = await readResponseTextPrefix(params.response, params.maxBytes); + return prefix.truncated ? appendTruncatedSuffix(prefix.text) : prefix.text; } - const reader = params.response.body.getReader(); - const decoder = new TextDecoder(); - let totalBytes = 0; - let text = ""; - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) { - return text + decoder.decode(); - } - if (!value?.byteLength) { - continue; - } - - const remainingBytes = params.maxBytes - totalBytes; - if (value.byteLength > remainingBytes) { - if (params.truncateOnLimit) { - const clipped = remainingBytes > 0 ? value.slice(0, remainingBytes) : undefined; - if (clipped) { - text += decoder.decode(clipped, { stream: true }); - } - await reader.cancel().catch(() => undefined); - return appendTruncatedSuffix(text + decoder.decode()); - } - await reader.cancel().catch(() => undefined); - throw new Error( - `provider response body too large: ${totalBytes + value.byteLength} bytes ` + - `(limit: ${params.maxBytes} bytes)`, - ); - } - - text += decoder.decode(value, { stream: true }); - totalBytes += value.byteLength; - } - } finally { - try { - reader.releaseLock(); - } catch {} - } + const body = await readResponseWithLimit(params.response, params.maxBytes, { + onOverflow: ({ size, maxBytes }) => + new Error(`provider response body too large: ${size} bytes (limit: ${maxBytes} bytes)`), + }); + return new TextDecoder().decode(body); } export async function readProviderJsonResponseText(response: Response): Promise { diff --git a/packages/ai/src/providers/openai-chatgpt-responses.test.ts b/packages/ai/src/providers/openai-chatgpt-responses.test.ts index 2c9fe4442326..20974b189b8a 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.test.ts @@ -898,11 +898,12 @@ describe("streamOpenAICodexResponses transport", () => { }); it("bounds non-OK ChatGPT response bodies before formatting API errors", async () => { - const chunkSize = 1024 * 1024; + const byteLimit = 16 * 1024; const totalChunks = 32; - const chunk = new TextEncoder() - .encode("usage limit ".repeat(Math.ceil(chunkSize / "usage limit ".length))) - .subarray(0, chunkSize); + const prefix = "usage limit "; + const chunk = new TextEncoder().encode( + `${prefix}${"x".repeat(byteLimit - prefix.length - 2)}😀tail`, + ); let pullCount = 0; let canceled = false; const overflowing = new ReadableStream({ @@ -939,6 +940,8 @@ describe("streamOpenAICodexResponses transport", () => { expect(result.stopReason).toBe("error"); expect(result.errorMessage).toContain("usage limit"); + expect(result.errorMessage).not.toContain("�"); + expect(result.errorMessage).not.toContain("tail"); expect(result.errorMessage?.length).toBeLessThanOrEqual(16 * 1024); expect(canceled).toBe(true); expect(pullCount).toBeGreaterThanOrEqual(1); diff --git a/packages/ai/src/providers/openai-chatgpt-responses.ts b/packages/ai/src/providers/openai-chatgpt-responses.ts index 0bc727ec69e4..e973257b4cee 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.ts @@ -1682,7 +1682,11 @@ async function readChatGptResponsesErrorTextLimited(response: Response): Promise break; } } - text += decoder.decode(); + // A capped prefix may end mid-sequence. Flushing only after EOF avoids + // inventing a replacement character while preserving malformed full bodies. + if (!reachedLimit) { + text += decoder.decode(); + } } finally { if (reachedLimit) { // This provider module is browser-safe, so keep error-body capping on Web APIs. diff --git a/packages/memory-host-sdk/src/host/response-snippet.test.ts b/packages/memory-host-sdk/src/host/response-snippet.test.ts index 9c8db2f365da..e7f7f89e8485 100644 --- a/packages/memory-host-sdk/src/host/response-snippet.test.ts +++ b/packages/memory-host-sdk/src/host/response-snippet.test.ts @@ -41,6 +41,19 @@ describe("readResponseTextSnippet", () => { ); }); + it("drops partial UTF-8 characters when byte-capped snippets truncate a stream", async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("ab" + String.fromCodePoint(0x1f600) + "cd")); + }, + cancel() {}, + }); + + await expect( + readResponseTextSnippet(new Response(stream), { maxBytes: 3, maxChars: 100 }), + ).resolves.toBe("ab... [truncated]"); + }); + it("cancels snippet body reads when the caller signal aborts", async () => { let canceled = false; const response = stallingResponse(() => { diff --git a/packages/memory-host-sdk/src/host/response-snippet.ts b/packages/memory-host-sdk/src/host/response-snippet.ts index 95fda9a0de3e..12172e4b5d25 100644 --- a/packages/memory-host-sdk/src/host/response-snippet.ts +++ b/packages/memory-host-sdk/src/host/response-snippet.ts @@ -1,4 +1,5 @@ // Memory Host SDK module implements response snippet behavior. +import { decodeTextPrefix } from "@openclaw/normalization-core"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; const DEFAULT_ERROR_BODY_MAX_BYTES = 8 * 1024; @@ -37,7 +38,9 @@ export async function readResponseTextSnippet( return ""; } - const text = new TextDecoder().decode(joinChunks(prefix.bytes, prefix.length)); + const text = decodeTextPrefix(joinChunks(prefix.bytes, prefix.length), { + truncated: prefix.truncated, + }); const collapsed = text.replace(/\s+/g, " ").trim(); if (!collapsed) { return ""; diff --git a/packages/normalization-core/src/index.ts b/packages/normalization-core/src/index.ts index e0e122eae808..2e5bae9bcea9 100644 --- a/packages/normalization-core/src/index.ts +++ b/packages/normalization-core/src/index.ts @@ -8,4 +8,5 @@ export * from "./number-coercion.js"; export * from "./record-coerce.js"; export * from "./string-coerce.js"; export * from "./string-normalization.js"; +export * from "./text-decoding.js"; export * from "./utf16-slice.js"; diff --git a/packages/normalization-core/src/text-decoding.test.ts b/packages/normalization-core/src/text-decoding.test.ts new file mode 100644 index 000000000000..7eaa2b3577d3 --- /dev/null +++ b/packages/normalization-core/src/text-decoding.test.ts @@ -0,0 +1,18 @@ +import { describe, expect, it } from "vitest"; +import { decodeTextPrefix } from "./text-decoding.js"; + +describe("decodeTextPrefix", () => { + const encoded = new TextEncoder().encode("ab😀cd"); + + it("decodes complete text normally", () => { + expect(decodeTextPrefix(encoded)).toBe("ab😀cd"); + }); + + it("drops an incomplete trailing sequence from a truncated prefix", () => { + expect(decodeTextPrefix(encoded.subarray(0, 3), { truncated: true })).toBe("ab"); + }); + + it("preserves normal replacement behavior for a complete malformed body", () => { + expect(decodeTextPrefix(encoded.subarray(0, 3))).toBe("ab�"); + }); +}); diff --git a/packages/normalization-core/src/text-decoding.ts b/packages/normalization-core/src/text-decoding.ts new file mode 100644 index 000000000000..85c128e632a1 --- /dev/null +++ b/packages/normalization-core/src/text-decoding.ts @@ -0,0 +1,12 @@ +export type DecodeTextPrefixOptions = { + encoding?: string; + truncated?: boolean; +}; + +/** Decodes a byte prefix without inventing a replacement character for a cut trailing sequence. */ +export function decodeTextPrefix(bytes: Uint8Array, options: DecodeTextPrefixOptions = {}): string { + const decoder = new TextDecoder(options.encoding); + // Streaming mode retains an incomplete tail; discarding this one-shot decoder + // drops only that cut sequence while complete bodies still flush normally. + return decoder.decode(bytes, options.truncated ? { stream: true } : undefined); +} diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 56441fdd2685..dc98fb4b4846 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -195,12 +195,12 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { ), publicExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", - 10468, + 10471, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", - 5224, + 5225, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/agents/provider-http-errors.test.ts b/src/agents/provider-http-errors.test.ts index a7b17919ff17..6cb79f456802 100644 --- a/src/agents/provider-http-errors.test.ts +++ b/src/agents/provider-http-errors.test.ts @@ -216,6 +216,12 @@ describe("provider error utils", () => { expect(releaseLock).toHaveBeenCalledTimes(1); }); + it("drops partial UTF-8 characters when provider error body reads truncate", async () => { + const response = new Response(new Blob([new TextEncoder().encode("ab😀cd")]).stream()); + + await expect(readResponseTextLimited(response, 3)).resolves.toBe("ab"); + }); + it("attaches structured provider error metadata", async () => { // API-key-like substrings must be redacted from stored error bodies. const response = new Response( diff --git a/src/agents/provider-http-errors.ts b/src/agents/provider-http-errors.ts index 3d0ba6473521..0fa813d519da 100644 --- a/src/agents/provider-http-errors.ts +++ b/src/agents/provider-http-errors.ts @@ -7,7 +7,7 @@ import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; export { asFiniteNumber } from "../../packages/normalization-core/src/number-coercion.js"; import { normalizeOptionalString as trimToUndefined } from "../../packages/normalization-core/src/string-coerce.js"; -import { readResponseWithLimit } from "../infra/http-body.js"; +import { readResponseTextPrefix, readResponseWithLimit } from "../infra/http-body.js"; import { redactSensitiveText } from "../logging/redact.js"; export { asBoolean } from "../utils/boolean.js"; export { normalizeOptionalString as trimToUndefined } from "../../packages/normalization-core/src/string-coerce.js"; @@ -42,53 +42,7 @@ export async function readResponseTextLimited( if (limitBytes <= 0) { return ""; } - const reader = response.body?.getReader(); - if (!reader) { - return ""; - } - - const decoder = new TextDecoder(); - let total = 0; - let text = ""; - let reachedLimit = false; - - try { - while (true) { - const { value, done } = await reader.read(); - if (done) { - break; - } - if (!value || value.byteLength === 0) { - continue; - } - const remaining = limitBytes - total; - if (remaining <= 0) { - reachedLimit = true; - break; - } - const chunk = value.byteLength > remaining ? value.subarray(0, remaining) : value; - total += chunk.byteLength; - text += decoder.decode(chunk, { stream: true }); - if (total >= limitBytes) { - reachedLimit = true; - break; - } - } - text += decoder.decode(); - } finally { - if (reachedLimit) { - // Stop the upstream body once the diagnostic budget is full. - await reader.cancel().catch(() => {}); - } - try { - reader.releaseLock(); - } catch { - // Error-body reads are diagnostic best effort; release failures must not - // hide the bounded provider error text already captured. - } - } - - return text; + return (await readResponseTextPrefix(response, limitBytes)).text; } /** Reads a successful provider text response under a byte cap. */ diff --git a/src/agents/tools/web-shared.test.ts b/src/agents/tools/web-shared.test.ts index e893c37d7090..f6866975a20e 100644 --- a/src/agents/tools/web-shared.test.ts +++ b/src/agents/tools/web-shared.test.ts @@ -23,6 +23,7 @@ function responseFromReader(params: { chunks: string[]; cancel: () => Promise; releaseLock: () => void; + contentType?: string; readError?: Error; }): Response { const chunks: Array> = params.chunks.map((chunk) => ({ @@ -49,7 +50,7 @@ function responseFromReader(params: { return { body: { getReader: () => reader }, - headers: new Headers({ "content-type": "text/plain; charset=utf-8" }), + headers: new Headers({ "content-type": params.contentType ?? "text/plain; charset=utf-8" }), } as Response; } @@ -159,6 +160,24 @@ describe("readResponseText", () => { expect(releaseLock).toHaveBeenCalledTimes(1); }); + it("drops partial UTF-8 characters when bounded response reads truncate a stream", async () => { + const cancel = vi.fn(async () => undefined); + const releaseLock = vi.fn(); + const response = responseFromReader({ + chunks: ["ab" + String.fromCodePoint(0x1f600) + "cd"], + cancel, + releaseLock, + }); + + await expect(readResponseText(response, { maxBytes: 3 })).resolves.toEqual({ + text: "ab", + truncated: true, + bytesRead: 3, + }); + expect(cancel).toHaveBeenCalledTimes(1); + expect(releaseLock).toHaveBeenCalledTimes(1); + }); + it("marks bounded response readers truncated after stream errors", async () => { const cancel = vi.fn(async () => undefined); const releaseLock = vi.fn(); @@ -232,6 +251,30 @@ describe("readResponseText", () => { expect(releaseLock).toHaveBeenCalledTimes(1); }); + it("keeps truncated fallback charset decoding isolated between responses", async () => { + const firstResponse = responseFromReader({ + chunks: ["ab😀cd"], + cancel: vi.fn(async () => undefined), + releaseLock: vi.fn(), + contentType: "text/plain; charset=x-unsupported-test", + }); + await expect(readResponseText(firstResponse, { maxBytes: 3 })).resolves.toMatchObject({ + text: "ab", + truncated: true, + }); + + const secondResponse = responseFromReader({ + chunks: ["cd"], + cancel: vi.fn(async () => undefined), + releaseLock: vi.fn(), + contentType: "text/plain; charset=x-unsupported-test", + }); + await expect(readResponseText(secondResponse, { maxBytes: 64 })).resolves.toMatchObject({ + text: "cd", + truncated: false, + }); + }); + it("does not mark exact-limit responses as truncated when followed by zero-byte chunks", async () => { const cancel = vi.fn(async () => undefined); const releaseLock = vi.fn(); diff --git a/src/agents/tools/web-shared.ts b/src/agents/tools/web-shared.ts index 582cf0a0d1d8..de7fc2d76982 100644 --- a/src/agents/tools/web-shared.ts +++ b/src/agents/tools/web-shared.ts @@ -3,6 +3,7 @@ * * Keeps web_fetch and web_search providers aligned on bounded IO and cache semantics. */ +import { decodeTextPrefix } from "@openclaw/normalization-core"; import { asDateTimestampMs, MAX_TIMER_TIMEOUT_SECONDS, @@ -119,7 +120,6 @@ export type ReadResponseTextResult = { const RESPONSE_CHARSET_SCAN_BYTES = 4096; const latin1Decoder = new TextDecoder("latin1"); -const utf8Decoder = new TextDecoder("utf-8"); function normalizeCharset(value: string | undefined): string | undefined { const normalized = value?.trim().replace(/^["']|["']$/g, "") ?? ""; @@ -215,13 +215,13 @@ function responseContentType(res: Response): string | null { return typeof headers?.get === "function" ? headers.get("content-type") : null; } -function decodeResponseBytes(res: Response, bytes: Uint8Array): string { +function decodeResponseBytes(res: Response, bytes: Uint8Array, truncated = false): string { const contentType = responseContentType(res); const charset = readCharsetParam(contentType) ?? sniffCharset(contentType, bytes); try { - return new TextDecoder(charset ?? "utf-8").decode(bytes); + return decodeTextPrefix(bytes, { encoding: charset ?? "utf-8", truncated }); } catch { - return utf8Decoder.decode(bytes); + return decodeTextPrefix(bytes, { encoding: "utf-8", truncated }); } } @@ -312,7 +312,7 @@ export async function readResponseText( } const bytes = concatBytes(parts, bytesRead); - return { text: decodeResponseBytes(res, bytes), truncated, bytesRead }; + return { text: decodeResponseBytes(res, bytes, truncated), truncated, bytesRead }; } if (maxBytes) { diff --git a/src/infra/http-body.response.test.ts b/src/infra/http-body.response.test.ts index 33fe5e344933..6e0dc25d5eec 100644 --- a/src/infra/http-body.response.test.ts +++ b/src/infra/http-body.response.test.ts @@ -1,7 +1,11 @@ // Tests bounded HTTP response reads and cleanup behavior. import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import { readResponseTextSnippet, readResponseWithLimit } from "./http-body.js"; +import { + readResponseTextPrefix, + readResponseTextSnippet, + readResponseWithLimit, +} from "./http-body.js"; function makeStream(chunks: Uint8Array[], delayMs?: number) { return new ReadableStream({ @@ -235,6 +239,12 @@ describe("readResponseTextSnippet", () => { options: { maxBytes: 7, maxChars: 50 }, expected: "1234567…", }, + { + name: "drops partial UTF-8 characters when snippets truncate at a byte boundary", + response: new Response(makeStream([new TextEncoder().encode("ab😀cd")])), + options: { maxBytes: 3, maxChars: 50 }, + expected: "ab…", + }, { name: "keeps character-limited snippets UTF-16 well-formed", response: new Response(makeStream([new TextEncoder().encode("ab🚀tail")])), @@ -253,6 +263,18 @@ describe("readResponseTextSnippet", () => { ).rejects.toThrow(/maxBytes must be a non-negative finite number/); }); + it("cancels immediately when a diagnostic prefix fills the byte budget", async () => { + const cancel = vi.fn(); + const response = new Response(makeStallingStream([new TextEncoder().encode("exact")], cancel)); + + await expect(readResponseTextPrefix(response, 5)).resolves.toEqual({ + text: "exact", + size: 5, + truncated: true, + }); + expect(cancel).toHaveBeenCalledTimes(1); + }); + it.each([ { name: "applies the idle timeout while reading snippets", diff --git a/src/infra/http-body.ts b/src/infra/http-body.ts index 72df7f924219..1899f1533211 100644 --- a/src/infra/http-body.ts +++ b/src/infra/http-body.ts @@ -1,6 +1,7 @@ // Reads HTTP request and response bodies with timeout and byte limits. import type { IncomingMessage, ServerResponse } from "node:http"; import { clearTimeout as clearNodeTimeout, setTimeout as setNodeTimeout } from "node:timers"; +import { decodeTextPrefix } from "@openclaw/normalization-core"; import { toErrorObject } from "@openclaw/normalization-core/error-coercion"; import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; @@ -181,6 +182,15 @@ type ReadResponsePrefixResult = { truncated: boolean; }; +export type ReadResponseTextPrefixOptions = { + chunkTimeoutMs?: number; + onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error; +}; + +type ReadResponsePrefixOptions = ReadResponseTextPrefixOptions & { + stopAtLimit?: boolean; +}; + function validateMaxBytes(maxBytes: number): void { if (!Number.isFinite(maxBytes) || maxBytes < 0) { throw new RangeError(`maxBytes must be a non-negative finite number: ${maxBytes}`); @@ -190,10 +200,7 @@ function validateMaxBytes(maxBytes: number): void { async function readResponsePrefix( response: Response, maxBytes: number, - options?: { - chunkTimeoutMs?: number; - onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error; - }, + options?: ReadResponsePrefixOptions, ): Promise { validateMaxBytes(maxBytes); const body = response.body; @@ -227,7 +234,7 @@ async function readResponsePrefix( continue; } const nextTotal = total + value.length; - if (nextTotal > maxBytes) { + if (nextTotal > maxBytes || (options?.stopAtLimit && nextTotal === maxBytes)) { const remaining = maxBytes - total; if (remaining > 0) { chunks.push(value.subarray(0, remaining)); @@ -260,6 +267,29 @@ async function readResponsePrefix( }; } +export type ReadResponseTextPrefixResult = { + text: string; + size: number; + truncated: boolean; +}; + +/** Reads and decodes a bounded text prefix while cancelling unread overflow. */ +export async function readResponseTextPrefix( + response: Response, + maxBytes: number, + options?: ReadResponseTextPrefixOptions, +): Promise { + const prefix = await readResponsePrefix(response, maxBytes, { + ...options, + stopAtLimit: true, + }); + return { + text: decodeTextPrefix(prefix.buffer, { truncated: prefix.truncated }), + size: prefix.size, + truncated: prefix.truncated, + }; +} + /** Reads a response body under a byte cap, cancelling the stream on overflow or idle timeout. */ export async function readResponseWithLimit( response: Response, @@ -296,20 +326,15 @@ export async function readResponseTextSnippet( ): Promise { const maxBytes = options?.maxBytes ?? 8 * 1024; const maxChars = options?.maxChars ?? 200; - const prefix = await readResponsePrefix(response, maxBytes, { + const prefix = await readResponseTextPrefix(response, maxBytes, { chunkTimeoutMs: options?.chunkTimeoutMs, onIdleTimeout: options?.onIdleTimeout, }); - if (prefix.buffer.length === 0) { + if (!prefix.text) { return undefined; } - const text = new TextDecoder().decode(prefix.buffer); - if (!text) { - return undefined; - } - - const collapsed = text.replace(/\s+/g, " ").trim(); + const collapsed = prefix.text.replace(/\s+/g, " ").trim(); if (!collapsed) { return undefined; } diff --git a/src/infra/http-error-body.test.ts b/src/infra/http-error-body.test.ts index 1a79e9be4c33..2d5cf0979745 100644 --- a/src/infra/http-error-body.test.ts +++ b/src/infra/http-error-body.test.ts @@ -71,6 +71,16 @@ describe("readResponseBodySnippet", () => { expect(byteLen).toBeLessThanOrEqual(100); }); + it("stream path drops partial UTF-8 characters at the byte boundary", async () => { + const response = new Response(new Blob([new TextEncoder().encode("ab😀cd")]).stream()); + const result = await readResponseBodySnippet(response, { + maxBytes: 3, + maxChars: 100, + }); + + expect(result).toBe("ab"); + }); + it("stream path still enforces maxChars", async () => { const data = new Uint8Array(500).fill(97); const response = new Response(new Blob([data]).stream()); diff --git a/src/infra/http-error-body.ts b/src/infra/http-error-body.ts index e30480a80963..5dc93e74ddff 100644 --- a/src/infra/http-error-body.ts +++ b/src/infra/http-error-body.ts @@ -1,4 +1,6 @@ +import { decodeTextPrefix } from "@openclaw/normalization-core"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; +import { readResponseTextPrefix } from "./http-body.js"; export async function readResponseBodySnippet( response: Response, @@ -11,56 +13,15 @@ export async function readResponseBodySnippet( const encoded = new TextEncoder().encode(text); if (encoded.byteLength > limits.maxBytes) { return truncateUtf16Safe( - new TextDecoder().decode(encoded.subarray(0, limits.maxBytes), { - stream: true, - }), + decodeTextPrefix(encoded.subarray(0, limits.maxBytes), { truncated: true }), limits.maxChars, ); } return truncateUtf16Safe(text, limits.maxChars); } - const reader = body.getReader(); - const chunks: Uint8Array[] = []; - let total = 0; - let truncated = false; - try { - while (true) { - const { done, value } = await reader.read(); - if (done || !value?.byteLength) { - break; - } - const remaining = limits.maxBytes - total; - if (remaining <= 0) { - truncated = true; - break; - } - if (value.byteLength > remaining) { - chunks.push(value.subarray(0, remaining)); - total += remaining; - truncated = true; - break; - } - chunks.push(value); - total += value.byteLength; - if (total >= limits.maxBytes) { - truncated = true; - break; - } - } - } finally { - if (truncated) { - await reader.cancel().catch(() => undefined); - } - try { - reader.releaseLock(); - } catch {} - } - - return truncateUtf16Safe( - new TextDecoder().decode(Buffer.concat(chunks, total)), - limits.maxChars, - ); + const prefix = await readResponseTextPrefix(response, limits.maxBytes); + return truncateUtf16Safe(prefix.text, limits.maxChars); } catch { return ""; } diff --git a/src/plugin-sdk/response-limit-runtime.ts b/src/plugin-sdk/response-limit-runtime.ts index 31c94c169442..b48d872320b4 100644 --- a/src/plugin-sdk/response-limit-runtime.ts +++ b/src/plugin-sdk/response-limit-runtime.ts @@ -1,4 +1,8 @@ // Narrow response-size reader for plugins that download bounded HTTP bodies. export { readByteStreamWithLimit } from "@openclaw/media-core/read-byte-stream-with-limit"; -export { readResponseWithLimit } from "../infra/http-body.js"; +export { readResponseTextPrefix, readResponseWithLimit } from "../infra/http-body.js"; +export type { + ReadResponseTextPrefixOptions, + ReadResponseTextPrefixResult, +} from "../infra/http-body.js"; diff --git a/src/plugins/openai-compatible-embedding-provider.test.ts b/src/plugins/openai-compatible-embedding-provider.test.ts index 0d56524ca543..a32e6249d775 100644 --- a/src/plugins/openai-compatible-embedding-provider.test.ts +++ b/src/plugins/openai-compatible-embedding-provider.test.ts @@ -123,6 +123,11 @@ async function startEmbeddingServer(params?: { }; } +const EMBEDDING_ERROR_BOUNDARY_PREFIX = "x".repeat(999); +const EMBEDDING_ERROR_BOUNDARY_BODY = `${EMBEDDING_ERROR_BOUNDARY_PREFIX}😀${"x".repeat( + 8 * 1024 - EMBEDDING_ERROR_BOUNDARY_PREFIX.length - 4, +)}`; + async function startHangingErrorEmbeddingServer(): Promise<{ baseUrl: string; closed: Promise; @@ -137,7 +142,7 @@ async function startHangingErrorEmbeddingServer(): Promise<{ await readJsonBody(req); res.on("close", resolveClosed); res.writeHead(502, { "content-type": "text/plain" }); - res.write("x".repeat(12_000)); + res.write(EMBEDDING_ERROR_BOUNDARY_BODY); })(); }); server.on("connection", (socket) => { @@ -394,7 +399,7 @@ describe("openai-compatible generic embedding provider", () => { }); }); - it("bounds and cancels non-ok embedding error bodies", async () => { + it("bounds exact-limit embedding errors without splitting UTF-16 and cancels", async () => { const server = await startHangingErrorEmbeddingServer(); const { provider } = await createOpenAICompatibleEmbeddingProvider( createOptions({ @@ -418,7 +423,7 @@ describe("openai-compatible generic embedding provider", () => { } expect(outcome.error).toBeInstanceOf(Error); expect((outcome.error as Error).message).toBe( - `openai-compatible embeddings failed: HTTP 502: ${"x".repeat(1_000)}... [truncated]`, + `openai-compatible embeddings failed: HTTP 502: ${EMBEDDING_ERROR_BOUNDARY_PREFIX}... [truncated]`, ); await expect( Promise.race([ diff --git a/src/plugins/openai-compatible-embedding-provider.ts b/src/plugins/openai-compatible-embedding-provider.ts index 1a4e3b3ca1a2..32a64c6228c8 100644 --- a/src/plugins/openai-compatible-embedding-provider.ts +++ b/src/plugins/openai-compatible-embedding-provider.ts @@ -4,6 +4,7 @@ import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import { readProviderJsonResponse } from "../agents/provider-http-errors.js"; import { normalizeSecretInputString } from "../config/types.secrets.js"; import { resolveConfiguredSecretInputString } from "../gateway/resolve-configured-secret-input-string.js"; +import { readResponseTextPrefix } from "../infra/http-body.js"; import { fetchWithSsrFGuard } from "../infra/net/fetch-guard.js"; import { ssrfPolicyFromHttpBaseUrlAllowedHostname, type SsrFPolicy } from "../infra/net/ssrf.js"; import type { @@ -290,57 +291,17 @@ async function readJsonResponse(response: Response): Promise { return await readProviderJsonResponse(response, "openai-compatible embeddings failed"); } -function concatBytes(chunks: Uint8Array[], totalLength: number): Uint8Array { - const combined = new Uint8Array(totalLength); - let offset = 0; - for (const chunk of chunks) { - combined.set(chunk, offset); - offset += chunk.byteLength; - } - return combined; -} - async function readEmbeddingErrorBodySnippet(response: Response): Promise { if (!response.body || response.bodyUsed) { return undefined; } - const reader = response.body.getReader(); - const chunks: Uint8Array[] = []; - let totalLength = 0; - let truncated = false; - try { - while (totalLength < EMBEDDING_ERROR_BODY_MAX_BYTES) { - const next = await reader.read(); - if (next.done) { - break; - } - const remaining = EMBEDDING_ERROR_BODY_MAX_BYTES - totalLength; - if (next.value.byteLength > remaining) { - chunks.push(next.value.slice(0, remaining)); - totalLength += remaining; - truncated = true; - await reader.cancel().catch(() => undefined); - break; - } - chunks.push(next.value); - totalLength += next.value.byteLength; - if (totalLength >= EMBEDDING_ERROR_BODY_MAX_BYTES) { - truncated = true; - await reader.cancel().catch(() => undefined); - break; - } - } - } catch { - await reader.cancel().catch(() => undefined); - return undefined; - } finally { - reader.releaseLock(); - } - - if (totalLength === 0) { + const prefix = await readResponseTextPrefix(response, EMBEDDING_ERROR_BODY_MAX_BYTES).catch( + () => undefined, + ); + if (!prefix?.text) { return undefined; } - const text = new TextDecoder().decode(concatBytes(chunks, totalLength)); + const { text, truncated } = prefix; if (text.length > EMBEDDING_ERROR_BODY_MAX_CHARS) { return `${truncateUtf16Safe(text, EMBEDDING_ERROR_BODY_MAX_CHARS)}${EMBEDDING_ERROR_TRUNCATED_SUFFIX}`; }