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..fd9efbaa3 --- /dev/null +++ b/packages/ai/src/utils/request-body.ts @@ -0,0 +1,353 @@ +/** + * 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; + } + // 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; + } + 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..1ec9a8256 --- /dev/null +++ b/packages/ai/test/request-body.test.ts @@ -0,0 +1,362 @@ +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("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); + 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"); + } + }); + } +});