From 75c3920c47f258bb76f0012075b11e5036e661df Mon Sep 17 00:00:00 2001 From: OpenSec <323596076+opensec-intelligence@users.noreply.github.com> Date: Tue, 20 Oct 2026 06:38:00 +0530 Subject: [PATCH 1/2] perf(ai): send request bodies built from the previous request's bytes Every request resends the whole conversation; the SDK serializes all of it again and fetch copies the string again to encode it. Conversations grow at the end, so each message is now compared with a snapshot of the message at the same index in the conversation's previous request (structure only, strings shared) and reuses that request's UTF-8 bytes when equal; changed or new messages are serialized again. The body goes to fetch as a Blob of the parts. The seam is the SDK client's buildBody (openai, @anthropic-ai/sdk), so the body object is the one the SDK would serialize. Applied in openai-completions, openai-responses, azure-openai-responses and anthropic-messages, only when the client sends with the runtime's own fetch (a custom or wrapping fetch still gets the SDK's string). Anything JSON.stringify treats specially falls back to re-serializing or to the SDK. Bodies byte-identical to 0.9.5 over 1,017 requests (1.8 GB) of completions, anthropic and responses sessions, with cache markers moving, compaction and unique file reads. long_session.py, 10 interleaved sessions, per prompt at 20 / 40 prompts: instructions -18.9% / -31.7%, cycles -10.5% / -18.1%, CPU -13.0% / -21.0%. --- packages/ai/src/api/anthropic-messages.ts | 5 +- packages/ai/src/api/azure-openai-responses.ts | 4 +- packages/ai/src/api/openai-completions.ts | 4 +- packages/ai/src/api/openai-responses.ts | 4 +- packages/ai/src/utils/json-body.ts | 28 +- packages/ai/src/utils/request-body.ts | 351 ++++++++++++++++++ packages/ai/test/request-body.test.ts | 336 +++++++++++++++++ 7 files changed, 713 insertions(+), 19 deletions(-) create mode 100644 packages/ai/src/utils/request-body.ts create mode 100644 packages/ai/test/request-body.test.ts diff --git a/packages/ai/src/api/anthropic-messages.ts b/packages/ai/src/api/anthropic-messages.ts index f39bfaa3d..71236573c 100644 --- a/packages/ai/src/api/anthropic-messages.ts +++ b/packages/ai/src/api/anthropic-messages.ts @@ -40,11 +40,11 @@ import type { import { appendAssistantMessageDiagnostic } from "../utils/diagnostics.ts"; import { AssistantMessageEventStream } from "../utils/event-stream.ts"; import { headersToRecord } from "../utils/headers.ts"; -import { cacheJsonSerialization } from "../utils/json-body.ts"; import { parseJsonWithRepair, parseStreamingJsonWhileStreaming, parseToolCallArguments } from "../utils/json-parse.ts"; import { getPiUserAgent } from "../utils/pi-user-agent.ts"; import { getProviderEnvValue } from "../utils/provider-env.ts"; import { retryProviderRequest } from "../utils/provider-retry.ts"; +import { cacheRequestBody } from "../utils/request-body.ts"; import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts"; import { iterateSseMessages } from "../utils/sse.ts"; import { getSystemMessageText, renderSystemMessageUpdate } from "../utils/text.ts"; @@ -514,7 +514,8 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = ( if (nextParams !== undefined) { params = { ...(nextParams as MessageCreateParamsStreaming), stream: true }; } - cacheJsonSerialization(params.messages); + // A client passed in by the caller is not wrapped. + cacheRequestBody(options?.client ? undefined : client, params.messages, "messages", options?.sessionId); try { response = await retryProviderRequest( () => client.beta.messages.create(params, requestOptions).asResponse(), diff --git a/packages/ai/src/api/azure-openai-responses.ts b/packages/ai/src/api/azure-openai-responses.ts index d33963d5e..f7f05b1ae 100644 --- a/packages/ai/src/api/azure-openai-responses.ts +++ b/packages/ai/src/api/azure-openai-responses.ts @@ -5,9 +5,9 @@ import type { Api, AssistantMessage, Model, SimpleStreamOptions, StreamFunction, import { formatProviderError, normalizeProviderError } from "../utils/error-body.ts"; import { AssistantMessageEventStream } from "../utils/event-stream.ts"; import { headersToRecord } from "../utils/headers.ts"; -import { cacheJsonSerialization } from "../utils/json-body.ts"; import { getPiUserAgent } from "../utils/pi-user-agent.ts"; import { retryProviderRequest } from "../utils/provider-retry.ts"; +import { cacheRequestBody } from "../utils/request-body.ts"; import { getDeclaredTools, resolveTranscript, resolveTranscriptTools } from "../utils/transcript.ts"; import { type AzureEndpointOptions, resolveAzureConfig, resolveDeploymentName } from "./azure-openai-config.ts"; import { createGrammarToolInputProperties } from "./constrained-sampling.ts"; @@ -80,7 +80,7 @@ export const stream: StreamFunction<"azure-openai-responses", AzureOpenAIRespons if (nextParams !== undefined) { params = nextParams as ResponseCreateParamsStreaming; } - cacheJsonSerialization(params.input); + cacheRequestBody(client, params.input, "input", options?.sessionId); const requestOptions = { ...(options?.signal ? { signal: options.signal } : {}), ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), diff --git a/packages/ai/src/api/openai-completions.ts b/packages/ai/src/api/openai-completions.ts index c923aeeec..d232b8e78 100644 --- a/packages/ai/src/api/openai-completions.ts +++ b/packages/ai/src/api/openai-completions.ts @@ -39,11 +39,11 @@ import { formatProviderError, normalizeProviderError } from "../utils/error-body import { AssistantMessageEventStream } from "../utils/event-stream.ts"; import { shortHash } from "../utils/hash.ts"; import { headersToRecord } from "../utils/headers.ts"; -import { cacheJsonSerialization } from "../utils/json-body.ts"; import { parseStreamingJsonWhileStreaming, parseToolCallArguments } from "../utils/json-parse.ts"; import { getPiUserAgent } from "../utils/pi-user-agent.ts"; import { getProviderEnvValue } from "../utils/provider-env.ts"; import { retryProviderRequest } from "../utils/provider-retry.ts"; +import { cacheRequestBody } from "../utils/request-body.ts"; import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts"; import { getSystemMessageText, renderSystemMessageUpdate } from "../utils/text.ts"; import { @@ -370,7 +370,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio if (nextParams !== undefined) { params = nextParams as OpenAI.Chat.Completions.ChatCompletionCreateParamsStreaming; } - cacheJsonSerialization(params.messages); + cacheRequestBody(client, params.messages, "messages", options?.sessionId); const requestOptions = { ...(options?.signal ? { signal: options.signal } : {}), ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), diff --git a/packages/ai/src/api/openai-responses.ts b/packages/ai/src/api/openai-responses.ts index 02163d3b6..108da5872 100644 --- a/packages/ai/src/api/openai-responses.ts +++ b/packages/ai/src/api/openai-responses.ts @@ -18,10 +18,10 @@ import type { import { formatProviderError, normalizeProviderError } from "../utils/error-body.ts"; import { AssistantMessageEventStream } from "../utils/event-stream.ts"; import { headersToRecord } from "../utils/headers.ts"; -import { cacheJsonSerialization } from "../utils/json-body.ts"; import { getPiUserAgent } from "../utils/pi-user-agent.ts"; import { getProviderEnvValue } from "../utils/provider-env.ts"; import { retryProviderRequest } from "../utils/provider-retry.ts"; +import { cacheRequestBody } from "../utils/request-body.ts"; import { getDeclaredTools, resolveTranscript, resolveTranscriptTools } from "../utils/transcript.ts"; import { createGrammarToolInputProperties } from "./constrained-sampling.ts"; import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./github-copilot-headers.ts"; @@ -206,7 +206,7 @@ export const stream: StreamFunction<"openai-responses", OpenAIResponsesOptions> if (nextParams !== undefined) { params = nextParams as ResponseCreateParamsStreaming; } - cacheJsonSerialization(params.input); + cacheRequestBody(client, params.input, "input", options?.sessionId); const requestOptions = { ...(options?.signal ? { signal: options.signal } : {}), ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), diff --git a/packages/ai/src/utils/json-body.ts b/packages/ai/src/utils/json-body.ts index f3c1af0a3..9e2f603bd 100644 --- a/packages/ai/src/utils/json-body.ts +++ b/packages/ai/src/utils/json-body.ts @@ -99,15 +99,29 @@ function substituteItems( return out; } +/** + * The toJSON cacheJsonSerialization gives an array: a copy of it with long strings replaced by their cached raw JSON, + * which JSON.stringify serializes to the same text as the array. + */ +export function cachedArrayToJSON(this: unknown[]): unknown { + try { + if (!rawJSON) return this; + startGeneration(); + return substituteItems(this, rawJSON, new Set()); + } catch { + // JSON.stringify serializes what toJSON returns without calling it again: the array as it is. + return this; + } +} + /** * Makes JSON.stringify of `items` (a request's messages) reuse the serialized form of long strings serialized before, * by defining a non-enumerable toJSON on the array. Does nothing to anything but an extensible plain array without a * toJSON (an onPayload hook may return a frozen one). Never throws. */ export function cacheJsonSerialization(items: unknown): void { - const toRaw = rawJSON; if ( - !toRaw || + !rawJSON || !isRawJSON || !Array.isArray(items) || Object.getPrototypeOf(items) !== Array.prototype || @@ -118,15 +132,7 @@ export function cacheJsonSerialization(items: unknown): void { } try { Object.defineProperty(items, "toJSON", { - value(this: unknown[]): unknown { - try { - startGeneration(); - return substituteItems(this, toRaw, new Set()); - } catch { - // JSON.stringify serializes what toJSON returns without calling it again: the array as it is. - return this; - } - }, + value: cachedArrayToJSON, configurable: true, writable: true, enumerable: false, diff --git a/packages/ai/src/utils/request-body.ts b/packages/ai/src/utils/request-body.ts new file mode 100644 index 000000000..f25d48966 --- /dev/null +++ b/packages/ai/src/utils/request-body.ts @@ -0,0 +1,351 @@ +/** + * Sends a request body built from the bytes of the messages already sent in the conversation's previous request. + * + * Every request resends the whole conversation, and conversations grow at the end: request N+1 is request N's + * messages plus a few new ones. Serializing the body walks and copies all of it, and fetch then copies the string again + * to encode it as UTF-8: at 400 messages (9 MB) that is most of the CPU a model turn costs the client. + * + * sendJsonBodyAsBytes(client, items, key, conversation) makes the SDK client send the body that holds `items` (the + * request's messages or input array) as a Blob of UTF-8 bytes. Each item is compared with a snapshot of the item at the + * same index in the conversation's previous request; an equal item reuses that request's bytes, any other is + * serialized again. Only the comparison walks the whole conversation, and it copies nothing. The Blob joins the parts + * (one copy), and fetch sends the Blob's bytes as they are (it would copy a string or a Uint8Array once more). + * + * The bytes are the UTF-8 encoding of what JSON.stringify(body) returns, which is what the SDK would send: + * - JSON.stringify of an array is "[" + the items' JSON joined with "," + "]", and an item's JSON depends only on the + * item. The rest of the body is serialized with a placeholder in place of the array. + * - A snapshot keeps the item's structure (plain objects with their keys in order, arrays, primitives) and shares its + * strings. Items are compared with their snapshot, never with the previous request's objects, so a message changed in + * place (a cache marker added to the latest one, compaction, an extension) is not equal and is serialized again. + * - Anything JSON.stringify treats specially (toJSON, raw JSON, boxed primitives, class instances) is never equal to + * a snapshot, so it is serialized every time. An item with a toJSON of its own, or a body or array that is not plain, + * sends the body as the SDK would. + * - Values are read as plain data: a getter that returns something else on every read, or a Proxy, can make an item + * compare equal while it would now serialize differently. + * + * The client's own body encoding is wrapped (one wrapper per client): it uses the bytes only for a body whose array was + * registered here and a client that sends with the runtime's own fetch (Bun's; Node's fetch is JavaScript, and any + * other fetch may read the body as a string), keeps the headers the SDK sets for a JSON body, and on any error leaves + * the body to the SDK. + * + * Conversations are told apart by a key (the session id); the last few keep their previous request's snapshots and + * bytes, about one request's size each. A conversation not seen in the latest requests is dropped. + */ + +import { cachedArrayToJSON, cacheJsonSerialization } from "./json-body.ts"; + +const { isRawJSON } = JSON as unknown as { isRawJSON?: (value: unknown) => boolean }; + +type Bytes = Uint8Array; + +/** The structure of an object serialized before; strings are shared with it. */ +class ObjectSnapshot { + readonly keys: string[]; + readonly values: unknown[]; + constructor(keys: string[], values: unknown[]) { + this.keys = keys; + this.values = values; + } +} + +class ArraySnapshot { + readonly items: unknown[]; + constructor(items: unknown[]) { + this.items = items; + } +} + +/** A snapshot of something that is serialized again for every request. */ +const UNCACHEABLE = Symbol("uncacheable"); + +function hasToJSON(value: object): boolean { + return typeof (value as { toJSON?: unknown }).toJSON === "function"; +} + +function isPlainRecord(value: object): boolean { + const proto = Object.getPrototypeOf(value); + return proto === Object.prototype || (proto === null && !isRawJSON?.(value)); +} + +function snapshot(value: unknown): unknown { + switch (typeof value) { + case "string": + case "number": + case "boolean": + case "undefined": + return value; + case "object": { + if (value === null) return null; + if (hasToJSON(value)) return UNCACHEABLE; + if (Array.isArray(value)) { + const items = new Array(value.length); + for (let i = 0; i < value.length; i++) { + const item = snapshot(value[i]); + if (item === UNCACHEABLE) return UNCACHEABLE; + items[i] = item; + } + return new ArraySnapshot(items); + } + if (!isPlainRecord(value)) return UNCACHEABLE; + const record = value as Record; + const keys = Object.keys(record); + const values = new Array(keys.length); + for (let i = 0; i < keys.length; i++) { + const item = snapshot(record[keys[i]]); + if (item === UNCACHEABLE) return UNCACHEABLE; + values[i] = item; + } + return new ObjectSnapshot(keys, values); + } + default: + // Functions, symbols, bigints. + return UNCACHEABLE; + } +} + +/** Whether `value` serializes as the value the snapshot was taken of did. */ +function matches(snap: unknown, value: unknown): boolean { + if (snap instanceof ObjectSnapshot) { + if (typeof value !== "object" || value === null || Array.isArray(value)) return false; + if (!isPlainRecord(value) || hasToJSON(value)) return false; + const { keys, values } = snap; + const record = value as Record; + let i = 0; + // for-in also lists inherited enumerable keys, which JSON.stringify skips: such an object is never equal. + for (const key in record) { + if (i >= keys.length || key !== keys[i] || !matches(values[i], record[key])) return false; + i++; + } + return i === keys.length; + } + if (snap instanceof ArraySnapshot) { + if (!Array.isArray(value) || hasToJSON(value)) return false; + const { items } = snap; + if (value.length !== items.length) return false; + for (let i = 0; i < items.length; i++) { + if (!matches(items[i], value[i])) return false; + } + return true; + } + return Object.is(snap, value); +} + +interface Conversation { + snapshots: unknown[]; + bytes: Bytes[]; + head: string; + headBytes: Bytes; + tail: string; + tailBytes: Bytes; + lastUse: number; +} + +const MAX_CONVERSATIONS = 4; +/** A conversation that none of the latest requests used is dropped. */ +const MAX_IDLE_REQUESTS = 32; + +const conversations = new Map(); +let requests = 0; +const encoder = new TextEncoder(); +const EMPTY: Bytes = new Uint8Array(0); + +function conversationFor(key: string): Conversation { + requests++; + for (const [other, state] of conversations) { + if (requests - state.lastUse > MAX_IDLE_REQUESTS) conversations.delete(other); + } + let state = conversations.get(key); + if (state) { + conversations.delete(key); + } else { + state = { snapshots: [], bytes: [], head: "", headBytes: EMPTY, tail: "", tailBytes: EMPTY, lastUse: 0 }; + if (conversations.size >= MAX_CONVERSATIONS) { + const oldest = conversations.keys().next().value; + if (oldest !== undefined) conversations.delete(oldest); + } + } + state.lastUse = requests; + conversations.set(key, state); + return state; +} + +/** Stands in for the array while the rest of the body is serialized; it cannot occur by chance. */ +const PLACEHOLDER = `\u0000pi-request-body-${Math.random().toString(36).slice(2)}${Date.now().toString(36)}`; +const PLACEHOLDER_JSON = JSON.stringify(PLACEHOLDER); + +/** Whether `items` is a plain array without a toJSON, or with json-body.ts's (which serializes to the same text). */ +function isPlainItems(items: unknown[], allowedToJSON: unknown): boolean { + if (Object.getPrototypeOf(items) !== Array.prototype) return false; + const toJSON = (items as { toJSON?: unknown }).toJSON; + return toJSON === undefined || toJSON === allowedToJSON; +} + +const COMMA: Bytes = encoder.encode(","); + +/** + * The UTF-8 bytes of JSON.stringify(body) in parts, in order, reusing the conversation's previous bytes of each item + * of body[key] that is unchanged; undefined when the body is not plain data this can serialize exactly. + */ +export function jsonBodyParts( + body: Record, + key: string, + conversation: string, + allowedToJSON?: unknown, +): Bytes[] | undefined { + const items = body[key]; + if (!Array.isArray(items) || !isPlainItems(items, allowedToJSON) || !isPlainRecord(body) || hasToJSON(body)) { + return undefined; + } + for (const item of items) { + // JSON.stringify passes an item's own toJSON its index as the key; serialized alone it would get "". + if (typeof item === "object" && item !== null && hasToJSON(item)) return undefined; + } + const text = JSON.stringify({ ...body, [key]: PLACEHOLDER }); + const at = text.indexOf(PLACEHOLDER_JSON); + if (at < 0 || text.lastIndexOf(PLACEHOLDER_JSON) !== at) return undefined; + const head = `${text.slice(0, at)}[`; + const tail = `]${text.slice(at + PLACEHOLDER_JSON.length)}`; + + const state = conversationFor(conversation); + const headBytes = head === state.head ? state.headBytes : encoder.encode(head); + const tailBytes = tail === state.tail ? state.tailBytes : encoder.encode(tail); + const snapshots = new Array(items.length); + const bytes = new Array(items.length); + const parts: Bytes[] = [headBytes]; + for (let i = 0; i < items.length; i++) { + const item = items[i]; + const previous = state.snapshots[i]; + if (previous !== undefined && previous !== UNCACHEABLE && matches(previous, item)) { + snapshots[i] = previous; + bytes[i] = state.bytes[i]; + } else { + // An item that serializes to nothing (undefined, a function) is null in an array. + bytes[i] = encoder.encode(JSON.stringify(item) ?? "null"); + snapshots[i] = snapshot(item); + } + if (i > 0) parts.push(COMMA); + parts.push(bytes[i]); + } + parts.push(tailBytes); + + state.snapshots = snapshots; + state.bytes = bytes; + state.head = head; + state.headBytes = headBytes; + state.tail = tail; + state.tailBytes = tailBytes; + return parts; +} + +interface Registration { + key: string; + conversation: string; + allowedToJSON: unknown; +} + +type BuildBody = ( + this: unknown, + args: { options: { body?: unknown; headers?: unknown } }, +) => { + bodyHeaders?: unknown; + body?: unknown; +}; + +const registered = new WeakMap(); +const wrappedClients = new WeakSet(); +const runtimeFetches = new WeakMap(); + +/** + * Whether `fetch` is the runtime's own (Bun's), not a function that wraps it: a wrapper passed to the SDK or installed + * as the global fetch may read the body as the string the SDK sends otherwise, so it gets that string. + */ +function isRuntimeFetch(fetch: unknown): boolean { + if (typeof fetch !== "function") return false; + let native = runtimeFetches.get(fetch); + if (native === undefined) { + // A bound function prints as native code too, but its name starts with "bound ". + native = fetch.name === "fetch" && Function.prototype.toString.call(fetch).includes("[native code]"); + runtimeFetches.set(fetch, native); + } + return native; +} + +function registrationFor(body: unknown): Registration | undefined { + if (typeof body !== "object" || body === null) return undefined; + const record = body as Record; + for (const key of ["messages", "input"]) { + const items = record[key]; + if (typeof items === "object" && items !== null) { + const registration = registered.get(items); + if (registration?.key === key) return registration; + } + } + return undefined; +} + +/** + * Makes `client` (an openai or @anthropic-ai/sdk client) send the request body whose `key` property is `items` as + * UTF-8 bytes built by jsonBodyParts. `allowedToJSON` is a toJSON on `items` known to serialize the array as it is. + * Does nothing to a client without the SDK's body encoding. Never throws. + */ +export function sendJsonBodyAsBytes( + client: unknown, + items: unknown, + key: "messages" | "input", + conversation: string | undefined, + allowedToJSON?: unknown, +): void { + if (typeof client !== "object" || client === null || !Array.isArray(items)) return; + const target = client as { buildBody?: unknown }; + const original = target.buildBody; + if (typeof original !== "function" || typeof Blob !== "function") return; + registered.set(items, { key, conversation: conversation ?? "", allowedToJSON }); + if (wrappedClients.has(client)) return; + const buildBody = original as BuildBody; + try { + Object.defineProperty(client, "buildBody", { + value(this: unknown, args: Parameters[0]): ReturnType { + const body = args?.options?.body; + const registration = registrationFor(body); + if (registration && isRuntimeFetch((this as { fetch?: unknown }).fetch)) { + try { + // The SDK's encoding of an empty JSON body gives the headers it sends with one. + const probe = buildBody.call(this, { ...args, options: { ...args.options, body: {} } }); + if (probe?.body === "{}") { + const parts = jsonBodyParts( + body as Record, + registration.key, + registration.conversation, + registration.allowedToJSON, + ); + if (parts) return { ...probe, body: new Blob(parts) }; + } + } catch { + // Left to the SDK. + } + } + return buildBody.call(this, args); + }, + configurable: true, + writable: true, + enumerable: false, + }); + wrappedClients.add(client); + } catch { + // Not wrapped; the SDK serializes the body as before. + } +} + +/** + * Prepares a request's `items` (body[key]) for serialization: long strings reuse their JSON (json-body.ts), and the + * body is sent as bytes that reuse the conversation's previous request (sendJsonBodyAsBytes). + */ +export function cacheRequestBody( + client: unknown, + items: unknown, + key: "messages" | "input", + conversation: string | undefined, +): void { + cacheJsonSerialization(items); + sendJsonBodyAsBytes(client, items, key, conversation, cachedArrayToJSON); +} diff --git a/packages/ai/test/request-body.test.ts b/packages/ai/test/request-body.test.ts new file mode 100644 index 000000000..c4f85479a --- /dev/null +++ b/packages/ai/test/request-body.test.ts @@ -0,0 +1,336 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { streamSimple as streamAnthropic } from "../src/api/anthropic-messages.ts"; +import { streamSimple as streamOpenAICompletions } from "../src/api/openai-completions.ts"; +import { streamSimple as streamOpenAIResponses } from "../src/api/openai-responses.ts"; +import type { Api, FetchFunction, Message, Model, TranscriptContext } from "../src/types.ts"; +import { cacheJsonSerialization } from "../src/utils/json-body.ts"; +import { jsonBodyParts } from "../src/utils/request-body.ts"; +import { normalizeContext } from "../src/utils/transcript.ts"; + +const long = (seed: string, length = 3000) => seed.repeat(Math.ceil(length / seed.length)).slice(0, length); +const decoder = new TextDecoder(); + +class Point { + x = 1; + y = 2; +} + +/** Items JSON.stringify treats in every special way it has; none of them may change the bytes. */ +function awkward(): unknown[] { + return [ + { role: "system", content: long("You are a helpful assistant. ") }, + { role: "user", content: long('quotes " backslashes \\ newlines \n tabs \t controls \u0001 ') }, + { role: "user", content: long("unicode é ü 漢字 → emoji 😀 ") }, + { role: "user", content: long("lone surrogate \ud800 and \udc00 ") }, + { + role: "tool", + content: [ + { type: "text", text: long("tool output\n") }, + { type: "image", data: long("QUJD") }, + ], + }, + { role: "assistant", content: "short", extra: undefined, fn: () => 1, nan: Number.NaN, inf: Infinity, neg: -0 }, + { nested: { deeper: { list: [long("x"), undefined, null, 1, true, () => 1, Symbol("s")] } } }, + { date: new Date(0), point: new Point(), boxed: new String("boxed"), map: new Map([[1, 2]]) }, + JSON.parse('{"__proto__": "kept", "text": "short"}'), + Object.assign(Object.create(null), { nullProto: long("null prototype ") }), + { raw: (JSON as unknown as { rawJSON: (text: string) => unknown }).rawJSON('"raw"') }, + [long("array in array"), [long("deeper array")]], + // biome-ignore lint/suspicious/noSparseArray: a hole serializes as null + [1, , 3], + long("bare string item"), + 42, + null, + undefined, + { 2: "integer keys", 1: long("first"), b: "b", a: "a" }, + ]; +} + +/** The text of jsonBodyParts(...), or undefined. */ +function encode(body: Record, key: string, conversation: string, allowedToJSON?: unknown) { + const parts = jsonBodyParts(body, key, conversation, allowedToJSON); + return parts && decoder.decode(Buffer.concat(parts)); +} + +function expectSame(body: Record, key: string, conversation = "c"): void { + const expected = JSON.stringify(body); + expect(encode(body, key, conversation)).toBe(expected); +} + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); + +describe("jsonBodyParts", () => { + it("encodes JSON.stringify of the body as UTF-8, first serialized and then reused", () => { + const messages = awkward(); + for (let request = 0; request < 3; request++) { + expectSame({ model: "m", messages: [...messages], stream: true, tools: [{ name: "é" }] }, "messages"); + } + }); + + it("serializes only the items that are new or changed", () => { + const messages: Record[] = []; + for (let i = 0; i < 20; i++) messages.push({ role: "user", content: [{ type: "text", text: long(`m${i} `) }] }); + expectSame({ model: "m", messages: [...messages] }, "messages", "counted"); + const stringify = vi.spyOn(JSON, "stringify"); + messages.push({ role: "assistant", content: "new" }); + expectSame({ model: "m", messages: [...messages] }, "messages", "counted"); + // Only the new message is serialized on its own. + const itemsSerialized = stringify.mock.calls.filter(([value]) => value === messages[20]).length; + expect(itemsSerialized).toBe(1); + expect(stringify.mock.calls.filter(([value]) => messages.slice(0, 20).includes(value as never)).length).toBe(0); + }); + + it("serializes again what changed in place, was replaced, or moved", () => { + const messages: Record[] = []; + const body = () => ({ model: "m", messages: [...messages], stream: true }); + for (let i = 0; i < 10; i++) { + messages.push({ role: i % 2 ? "assistant" : "user", content: [{ type: "text", text: long(`turn ${i} `) }] }); + } + expectSame(body(), "messages"); + // A cache marker added to an earlier message's block, in place. + const block = (messages[3].content as Record[])[0]; + block.cache_control = { type: "ephemeral" }; + expectSame(body(), "messages"); + // ...and removed again, the next message marked instead. + delete block.cache_control; + (messages[4].content as Record[])[0].cache_control = { type: "ephemeral" }; + expectSame(body(), "messages"); + // The same keys in another order, a string changed by one character, a number by sign. + messages[1] = { content: messages[1].content, role: messages[1].role }; + (messages[2].content as Record[])[0].text = `${long("turn 2 ").slice(0, -1)}!`; + messages[5].n = 0; + expectSame(body(), "messages"); + messages[5].n = -0; + expectSame(body(), "messages"); + // Compaction: earlier messages replaced by a summary, everything after moves. + messages.splice(0, 6, { role: "user", content: "summary" }); + expectSame(body(), "messages"); + // An item that becomes a class instance or gets an inherited toJSON, a nested Date. + messages[1] = Object.assign(new Point(), messages[1]); + expectSame(body(), "messages"); + (messages[2].content as Record[])[0].when = new Date(5); + expectSame(body(), "messages"); + expectSame(body(), "messages"); + // Fields added or removed, undefined values, an array grown. + messages[0].extra = undefined; + expectSame(body(), "messages"); + messages[0].extra = [1, 2]; + expectSame(body(), "messages"); + (messages[0].extra as number[]).push(3); + expectSame(body(), "messages"); + delete messages[0].extra; + expectSame(body(), "messages"); + // The rest of the body changes. + expectSame({ ...body(), stream: false, tools: [{ name: "t" }] }, "messages"); + expectSame({ tools: [], messages: [...messages] }, "messages"); + }); + + it("stays byte-identical through random edits of a growing conversation", () => { + let seed = 7; + const random = (n: number) => { + seed = (seed * 1103515245 + 12345) % 2147483648; + return seed % n; + }; + const pool = awkward(); + const messages: unknown[] = []; + for (let request = 0; request < 300; request++) { + const action = random(10); + if (action < 5 || messages.length === 0) { + messages.push(structuredClone({ role: "user", content: [{ type: "text", text: long(`r${request} é `) }] })); + } else if (action === 5) { + messages.push(pool[random(pool.length)]); + } else if (action === 6) { + const target = messages[random(messages.length)]; + if ( + typeof target === "object" && + target !== null && + !Array.isArray(target) && + Object.isExtensible(target) + ) { + (target as Record).marker = random(3) === 0 ? undefined : { n: random(5) }; + } + } else if (action === 7) { + messages.splice(random(messages.length), random(3)); + } else if (action === 8) { + messages[random(messages.length)] = { role: "assistant", content: `x${random(4)}` }; + } else { + messages.reverse(); + } + expectSame({ model: "m", messages: [...messages], n: random(3) }, "messages", `fuzz${random(2)}`); + } + }); + + it("keeps conversations apart", () => { + const a = [{ role: "user", content: long("a ") }]; + const b = [{ role: "user", content: long("b ") }]; + for (let request = 0; request < 5; request++) { + a.push({ role: "user", content: `a${request}` }); + b.push({ role: "user", content: `b${request}` }); + expectSame({ messages: [...a] }, "messages", "a"); + expectSame({ messages: [...b] }, "messages", "b"); + // The same key used by both: still correct, only not reused. + expectSame({ messages: [...a] }, "messages", "shared"); + expectSame({ messages: [...b] }, "messages", "shared"); + } + }); + + it("leaves to JSON.stringify what it cannot reproduce exactly", () => { + const items = [{ role: "user", content: "x" }]; + // An item's own toJSON gets its index as the key. + expect(encode({ messages: [{ toJSON: (key: string) => key }] }, "messages", "c")).toBeUndefined(); + expect(encode({ messages: [new Date(0)] }, "messages", "c")).toBeUndefined(); + // A toJSON on the array or the body, an array that is not plain, no array. + const withToJSON = Object.assign([...items], { toJSON: () => [] }); + expect(encode({ messages: withToJSON }, "messages", "c")).toBeUndefined(); + expect(encode({ messages: items, toJSON: () => ({}) }, "messages", "c")).toBeUndefined(); + class Items extends Array {} + expect(encode({ messages: Items.from(items) }, "messages", "c")).toBeUndefined(); + expect(encode({ messages: "x" }, "messages", "c")).toBeUndefined(); + expect( + encode(Object.assign(new Point(), { messages: items }) as unknown as Record, "messages", "c"), + ).toBeUndefined(); + // A BigInt makes JSON.stringify throw: the SDK's own serialization throws it. + expect(() => encode({ messages: [{ n: 1n }] }, "messages", "c")).toThrow(TypeError); + }); + + it("accepts the array toJSON of json-body.ts, which serializes the array as it is", () => { + const items = awkward(); + cacheJsonSerialization(items); + const toJSON = (items as unknown as { toJSON: unknown }).toJSON; + expect(encode({ messages: items }, "messages", "c")).toBeUndefined(); + expect(encode({ messages: items }, "messages", "c", toJSON)).toBe(JSON.stringify({ messages: items })); + }); +}); + +function createModel(api: TApi): Model { + return { + id: "test-model", + name: "Test Model", + api, + provider: "test-provider", + baseUrl: "https://upstream.test/v1", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 100_000, + maxTokens: 1_000, + }; +} + +function turn(i: number): Message[] { + return [ + { role: "user", content: long(`Question ${i}: read the file. `), timestamp: i * 3 }, + { + role: "assistant", + content: [ + { type: "text", text: long(`Reading ${i}. `) }, + { type: "toolCall", id: `call_${i}`, name: "read", arguments: { path: `${i}.ts` } }, + ], + api: "openai-completions", + provider: "test-provider", + model: "test-model", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "toolUse", + timestamp: i * 3 + 1, + }, + { + role: "toolResult", + toolCallId: `call_${i}`, + toolName: "read", + content: [{ type: "text", text: long(`export const v${i} = "\\u00e9";\n`) }], + isError: false, + timestamp: i * 3 + 2, + }, + ]; +} + +/** + * A fetch the body encoding takes for the runtime's own (Bun's): a bound function, which prints as native code, named + * "fetch". Any other fetch gets the SDK's string (json-body.test.ts). + */ +function runtimeLikeFetch(spy: FetchFunction): FetchFunction { + return Object.defineProperty(spy.bind(undefined), "name", { value: "fetch" }); +} + +type Run = ( + context: TranscriptContext, + fetch: FetchFunction, + onPayload: (payload: unknown) => unknown, +) => Promise; + +describe("request bodies sent as bytes", () => { + const options = { apiKey: "k", maxRetries: 0, sessionId: "session-1" }; + const adapters: [string, Run][] = [ + [ + "openai-completions", + (context, fetch, onPayload) => + streamOpenAICompletions(createModel("openai-completions"), context, { + ...options, + fetch, + onPayload, + }).result(), + ], + [ + "openai-responses", + (context, fetch, onPayload) => + streamOpenAIResponses(createModel("openai-responses"), context, { ...options, fetch, onPayload }).result(), + ], + [ + "anthropic-messages", + (context, fetch, onPayload) => + streamAnthropic(createModel("anthropic-messages"), context, { ...options, fetch, onPayload }).result(), + ], + ]; + + it("leaves the body to the SDK for a fetch that is not the runtime's own", async () => { + let sent: unknown; + const fetch = vi.fn(async (_url, init) => { + sent = init?.body; + return new Response("{}", { status: 400 }); + }); + await adapters[0][1](normalizeContext({ messages: turn(0) }), fetch, () => undefined); + expect(typeof sent).toBe("string"); + }); + + for (const [name, run] of adapters) { + it(`sends JSON.stringify of the payload as a conversation grows and changes (${name})`, async () => { + const messages: Message[] = []; + for (let request = 0; request < 8; request++) { + messages.push(...turn(request)); + // Compaction halfway: the first turns replaced by a summary. + if (request === 4) + messages.splice(0, 9, { role: "user", content: "Summary of earlier turns.", timestamp: 0 }); + let expected: string | undefined; + let sent: unknown; + let contentType: string | null = null; + const spy = vi.fn(async (_url, init) => { + sent = init?.body; + contentType = new Headers(init?.headers).get("content-type"); + return new Response(JSON.stringify({ error: { message: "rejected" } }), { + status: 400, + headers: { "content-type": "application/json" }, + }); + }); + await run(normalizeContext({ messages: [...messages] }), runtimeLikeFetch(spy), (payload) => { + const { betas: _betas, ...body } = payload as Record; + expected = JSON.stringify(body); + return undefined; + }); + expect(spy).toHaveBeenCalledOnce(); + expect(sent).toBeInstanceOf(Blob); + expect(await (sent as Blob).text()).toBe(expected); + expect(contentType).toBe("application/json"); + } + }); + } +}); From ff235b1b1f2924a91d306eb45bb24ebe0d0eee95 Mon Sep 17 00:00:00 2001 From: OpenSec <323596076+opensec-intelligence@users.noreply.github.com> Date: Tue, 20 Oct 2026 07:34:08 +0530 Subject: [PATCH 2/2] fix(ai): read the request array by index, never through its own iterator JSON.stringify reads an array by index and ignores an overridden Symbol.iterator. The check for items with their own toJSON used for..of, so an onPayload array whose iterator skips such an item bypassed the fallback, and a side-effecting iterator ran during serialization. It now reads by index like the rest of the path. --- packages/ai/src/utils/request-body.ts | 4 +++- packages/ai/test/request-body.test.ts | 26 ++++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/packages/ai/src/utils/request-body.ts b/packages/ai/src/utils/request-body.ts index f25d48966..fd9efbaa3 100644 --- a/packages/ai/src/utils/request-body.ts +++ b/packages/ai/src/utils/request-body.ts @@ -196,7 +196,9 @@ export function jsonBodyParts( if (!Array.isArray(items) || !isPlainItems(items, allowedToJSON) || !isPlainRecord(body) || hasToJSON(body)) { return undefined; } - for (const item of items) { + // By index, as JSON.stringify reads an array: an iterator of the array's own is never run. + for (let i = 0; i < items.length; i++) { + const item: unknown = items[i]; // JSON.stringify passes an item's own toJSON its index as the key; serialized alone it would get "". if (typeof item === "object" && item !== null && hasToJSON(item)) return undefined; } diff --git a/packages/ai/test/request-body.test.ts b/packages/ai/test/request-body.test.ts index c4f85479a..1ec9a8256 100644 --- a/packages/ai/test/request-body.test.ts +++ b/packages/ai/test/request-body.test.ts @@ -196,6 +196,32 @@ describe("jsonBodyParts", () => { expect(() => encode({ messages: [{ n: 1n }] }, "messages", "c")).toThrow(TypeError); }); + it("reads the array by index, as JSON.stringify does, never through an iterator of its own", () => { + let iterated = 0; + const withIterator = (items: T[]): T[] => + Object.defineProperty(items, Symbol.iterator, { + value: () => { + iterated++; + // Skips everything, and changes the array: neither may affect the body. + items.push({ role: "user", content: "added by the iterator" } as T); + return [][Symbol.iterator](); + }, + }); + // An item with its own toJSON that the iterator would hide: still left to JSON.stringify. + expect( + encode({ messages: withIterator([{ role: "user" }, { toJSON: (key: string) => key }]) }, "messages", "c"), + ).toBeUndefined(); + const plain = withIterator([ + { role: "user", content: long("a ") }, + { role: "assistant", content: "b" }, + ]); + for (let request = 0; request < 2; request++) { + expectSame({ model: "m", messages: plain }, "messages", "iterator"); + } + expect(iterated).toBe(0); + expect(plain).toHaveLength(2); + }); + it("accepts the array toJSON of json-body.ts, which serializes the array as it is", () => { const items = awkward(); cacheJsonSerialization(items);