mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix(infra): preserve characters in truncated response snippets (#103136)
* fix(infra): preserve bounded response text characters Co-authored-by: qingminlong <34085845+qingminglong@users.noreply.github.com> * ci(plugin-sdk): refresh API baseline --------- Co-authored-by: qingminlong <34085845+qingminglong@users.noreply.github.com>
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
af3a76f320
commit
d185f9a0b4
@@ -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
|
||||
|
||||
@@ -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<typeof fetch>(
|
||||
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<typeof fetch>(
|
||||
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({
|
||||
|
||||
@@ -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<string> {
|
||||
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<string, string> {
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<string> {
|
||||
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<string> {
|
||||
|
||||
@@ -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<Uint8Array>({
|
||||
@@ -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);
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -41,6 +41,19 @@ describe("readResponseTextSnippet", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("drops partial UTF-8 characters when byte-capped snippets truncate a stream", async () => {
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
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(() => {
|
||||
|
||||
@@ -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 "";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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�");
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -23,6 +23,7 @@ function responseFromReader(params: {
|
||||
chunks: string[];
|
||||
cancel: () => Promise<void>;
|
||||
releaseLock: () => void;
|
||||
contentType?: string;
|
||||
readError?: Error;
|
||||
}): Response {
|
||||
const chunks: Array<ReadableStreamReadResult<Uint8Array>> = 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();
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<Uint8Array>({
|
||||
@@ -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",
|
||||
|
||||
+38
-13
@@ -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<ReadResponsePrefixResult> {
|
||||
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<ReadResponseTextPrefixResult> {
|
||||
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<string | undefined> {
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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 "";
|
||||
}
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<void>;
|
||||
@@ -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([
|
||||
|
||||
@@ -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<unknown> {
|
||||
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<string | undefined> {
|
||||
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}`;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user