From 826e0debf6d48e793043d6263c1c194e4461f942 Mon Sep 17 00:00:00 2001 From: Ingwannu Date: Sat, 1 Aug 2026 07:22:55 +0000 Subject: [PATCH] bound streamed tool argument memory --- src/adapters/openai-chat.ts | 53 ++++++- src/bridge.ts | 70 ++++++++- src/chat/outbound.ts | 194 +++++++++++++++--------- src/lib/sse-decoder.ts | 35 ++++- src/lib/tool-argument-bounds.ts | 55 +++++++ structure/04_transports-and-sidecars.md | 23 +++ tests/sse-decoder.test.ts | 34 ++++- tests/tool-argument-bounds.test.ts | 135 +++++++++++++++++ 8 files changed, 512 insertions(+), 87 deletions(-) create mode 100644 src/lib/tool-argument-bounds.ts create mode 100644 tests/tool-argument-bounds.test.ts diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index a053e0c48..564717c81 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -10,6 +10,8 @@ import { contentPartsToText } from "./image"; import { neutralizeIdentity } from "./identity"; import { buildNonOpenAIToolCatalogNudgeForTools, shouldInjectNonOpenAIToolCatalogNudge } from "./tool-catalog-nudge"; import { openRouterProviderPayload, resolveOpenRouterRouting } from "../providers/openrouter-routing"; +import { MAX_SSE_RECORD_BYTES } from "../lib/sse-decoder"; +import { appendToolArguments, utf8ByteLength } from "../lib/tool-argument-bounds"; // Providers may opt into stripping one trailing "[...]" group from the wire model id. // Z.AI needs this because its OpenAI path rejects glm-5.2[1m] with 400 code 1211; @@ -693,8 +695,9 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd // chunks, whole-chunk calls) cannot be represented live without overlapping sequences. // Keyed by `index` (OpenAI wire standard), falling back to `id`, falling back to the // last-seen call for providers that omit both on continuation chunks. - interface PendingToolCall { key: string; id: string; name: string; args: string } + interface PendingToolCall { key: string; id: string; name: string; args: string; argsBytes: number } const pendingToolCalls: PendingToolCall[] = []; + let pendingToolArgumentBytes = 0; let toolCallSeq = 0; const flushToolCalls = function* (): Generator { // Do not treat flushed tool calls as user-facing output for the finish-less EOF @@ -706,6 +709,7 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd yield { type: "tool_call_end" }; } pendingToolCalls.length = 0; + pendingToolArgumentBytes = 0; }; let pendingUsage: OcxUsage | undefined; // Track terminal signals so a socket EOF without any terminator can fail closed instead of @@ -722,6 +726,17 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd // Yields adapter events and returns "terminate" for a terminal frame ([DONE] / error) that // must end the stream, or "continue" otherwise. Mutates the closure's terminal-signal state. const handleDataLine = function* (line: string): Generator { + if (utf8ByteLength(line) > MAX_SSE_RECORD_BYTES) { + pendingToolCalls.length = 0; + pendingToolArgumentBytes = 0; + yield { + type: "error", + message: "upstream SSE record exceeded the 4 MiB limit", + status: 502, + errorType: "upstream_error", + }; + return "terminate"; + } if (!line.startsWith("data: ")) return "continue"; const payload = line.slice(6).trim(); if (payload === "[DONE]") { @@ -803,12 +818,33 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd // splitting into two calls that share one call_id downstream. if (!call && tc.id) call = pendingToolCalls.find(c => c.id === tc.id); if (!call) { - call = { key: key ?? `seq:${pendingToolCalls.length}`, id: "", name: "", args: "" }; + call = { key: key ?? `seq:${pendingToolCalls.length}`, id: "", name: "", args: "", argsBytes: 0 }; pendingToolCalls.push(call); } if (tc.id && !call.id) call.id = tc.id; if (tc.function?.name && !call.name) call.name = tc.function.name; - if (tc.function?.arguments) call.args += tc.function.arguments; + if (tc.function?.arguments) { + const appended = appendToolArguments( + call.args, + call.argsBytes, + pendingToolArgumentBytes, + tc.function.arguments, + ); + if (!appended) { + pendingToolCalls.length = 0; + pendingToolArgumentBytes = 0; + yield { + type: "error", + message: "upstream tool call arguments exceeded the configured byte limit", + status: 502, + errorType: "upstream_error", + }; + return "terminate"; + } + call.args = appended.value; + call.argsBytes = appended.valueBytes; + pendingToolArgumentBytes = appended.retainedTurnBytes; + } } } } @@ -833,6 +869,17 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd for (const line of lines) { if ((yield* handleDataLine(line)) === "terminate") return; } + if (utf8ByteLength(buffer) > MAX_SSE_RECORD_BYTES) { + pendingToolCalls.length = 0; + pendingToolArgumentBytes = 0; + yield { + type: "error", + message: "upstream SSE record exceeded the 4 MiB limit", + status: 502, + errorType: "upstream_error", + }; + return; + } } // Some providers send the terminal `data:` frame (carrying the final delta, finish_reason, diff --git a/src/bridge.ts b/src/bridge.ts index 9c1c62969..7d3c1eba2 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -4,6 +4,7 @@ import { encodeCompactionSummary } from "./responses/compaction"; import { encodeReasoningEnvelope, type ReasoningEnvelope } from "./responses/reasoning-envelope"; import { resolveStallTimeoutSec } from "./stall-timeout"; import { usageDisplayTotalTokens } from "./usage/totals"; +import { appendToolArguments } from "./lib/tool-argument-bounds"; function uuid(): string { return crypto.randomUUID().replace(/-/g, ""); @@ -314,7 +315,8 @@ export function bridgeToResponsesSSE( // Full assistant text of a compaction turn (across message boundaries) — becomes the // synthetic compaction item's payload on done. let compactionText = ""; - let currentToolCall: { itemId: string; outputIndex: number; callId: string; name: string; args: string; namespace?: string; freeform?: boolean; toolSearch?: boolean; inputEmitted?: string } | null = null; + let retainedToolArgumentBytes = 0; + let currentToolCall: { itemId: string; outputIndex: number; callId: string; name: string; args: string; argsBytes: number; namespace?: string; freeform?: boolean; toolSearch?: boolean; inputEmitted?: string } | null = null; // Open native web-search cell (between begin and end). Holds the output index allocated on // begin so the matching done reuses it; closed as `failed` if the stream terminates early. let currentWebSearch: { itemId: string; eventId: string; outputIndex: number } | null = null; @@ -676,12 +678,42 @@ export function bridgeToResponsesSSE( ? { type: "custom_tool_call", id: itemId, call_id: event.id, name: realName, input: "", status: "in_progress" } : { type: "function_call", id: itemId, call_id: event.id, name: realName, arguments: "", status: "in_progress", ...(ns ? { namespace: ns } : {}) }; emit("response.output_item.added", { output_index: outputIndex, item }); - currentToolCall = { itemId, outputIndex, callId: event.id, name: realName, args: "", namespace: ns, freeform, toolSearch }; + currentToolCall = { itemId, outputIndex, callId: event.id, name: realName, args: "", argsBytes: 0, namespace: ns, freeform, toolSearch }; break; } case "tool_call_delta": { if (currentToolCall) { - currentToolCall.args += event.arguments; + const appended = appendToolArguments( + currentToolCall.args, + currentToolCall.argsBytes, + retainedToolArgumentBytes, + event.arguments, + ); + if (!appended) { + retainedToolArgumentBytes = Math.max(0, retainedToolArgumentBytes - currentToolCall.argsBytes); + currentToolCall.args = ""; + currentToolCall.argsBytes = 0; + currentToolCall.inputEmitted = undefined; + failCurrentToolCall(); + const failure = responseError( + 502, + "upstream_error", + "upstream tool call arguments exceeded the configured byte limit", + ); + emit("response.failed", { + response: { + ...responseSnapshot("failed", finishedItems), + error: failure, + last_error: failure, + }, + }); + reportTerminal("failed"); + terminalEvent = true; + break; + } + currentToolCall.args = appended.value; + currentToolCall.argsBytes = appended.valueBytes; + retainedToolArgumentBytes = appended.retainedTurnBytes; if (!currentToolCall.freeform && !currentToolCall.toolSearch) { emit("response.function_call_arguments.delta", { item_id: currentToolCall.itemId, output_index: currentToolCall.outputIndex, @@ -1041,6 +1073,9 @@ export function buildResponseJSON( let currentToolCallId = ""; let currentToolCallName = ""; let currentToolCallArgs = ""; + let currentToolCallArgsBytes = 0; + let retainedToolArgumentBytes = 0; + let toolArgumentOverflow = false; // Web-search citations awaiting the next assistant message (attached as url_citation annotations). let pendingWebSources: { url: string; title?: string }[] = []; @@ -1132,6 +1167,7 @@ export function buildResponseJSON( currentToolCallId = ""; currentToolCallName = ""; currentToolCallArgs = ""; + currentToolCallArgsBytes = 0; }; for (const e of events) { @@ -1186,9 +1222,34 @@ export function buildResponseJSON( currentToolCallId = e.id; currentToolCallName = e.name; currentToolCallArgs = ""; + currentToolCallArgsBytes = 0; break; case "tool_call_delta": - currentToolCallArgs += e.arguments; + { + const appended = appendToolArguments( + currentToolCallArgs, + currentToolCallArgsBytes, + retainedToolArgumentBytes, + e.arguments, + ); + if (!appended) { + retainedToolArgumentBytes = Math.max(0, retainedToolArgumentBytes - currentToolCallArgsBytes); + currentToolCallArgs = ""; + currentToolCallArgsBytes = 0; + flushToolCall("incomplete"); + errorEvent = { + type: "error", + message: "upstream tool call arguments exceeded the configured byte limit", + status: 502, + errorType: "upstream_error", + }; + toolArgumentOverflow = true; + break; + } + currentToolCallArgs = appended.value; + currentToolCallArgsBytes = appended.valueBytes; + retainedToolArgumentBytes = appended.retainedTurnBytes; + } break; case "tool_call_end": if (!toolCallArgumentsUsable(currentToolCallArgs) && currentToolCallId) { @@ -1249,6 +1310,7 @@ export function buildResponseJSON( if (e.stopReason === "max_tokens" || e.stopReason === "content_filter") stopReason = e.stopReason; break; } + if (toolArgumentOverflow) break; } flushText(cleanDone && !errorEvent && !incompleteEvent ? "final_answer" : undefined); flushSummaryReasoning(); diff --git a/src/chat/outbound.ts b/src/chat/outbound.ts index eedf5f76e..ff075fa06 100644 --- a/src/chat/outbound.ts +++ b/src/chat/outbound.ts @@ -9,6 +9,7 @@ type Rec = Record; import { decodeServerSentEvents } from "../lib/sse-decoder"; import { classifyError, CYBER_POLICY_ERROR_CODE, isCyberPolicyCode, isCyberPolicyMessage } from "../lib/errors"; +import { appendToolArguments, replaceToolArguments } from "../lib/tool-argument-bounds"; function isRec(v: unknown): v is Rec { return !!v && typeof v === "object" && !Array.isArray(v); @@ -155,6 +156,8 @@ export function responsesSseToChatCompletionsSse( const toolCallIdByIndex = new Map(); const toolNameByIndex = new Map(); const toolArgumentsByIndex = new Map(); + const toolArgumentBytesByIndex = new Map(); + let retainedToolArgumentBytes = 0; const emittedToolIndexes = new Set(); let nextToolIndex = 0; let sseIterator: AsyncGenerator<{ event?: string; data: string }> | undefined; @@ -272,6 +275,40 @@ export function responsesSseToChatCompletionsSse( } try { controller.close(); } catch { /* already closed */ } }; + const failToolArgumentOverflow = () => { + toolArgumentsByIndex.clear(); + toolArgumentBytesByIndex.clear(); + retainedToolArgumentBytes = 0; + fail("upstream tool call arguments exceeded the configured byte limit", { + status: 502, + type: "upstream_error", + }); + }; + const retainToolArguments = (toolIndex: number, replacement: string): boolean => { + const currentBytes = toolArgumentBytesByIndex.get(toolIndex) ?? 0; + const retained = replaceToolArguments(currentBytes, retainedToolArgumentBytes, replacement); + if (!retained) { + failToolArgumentOverflow(); + return false; + } + toolArgumentsByIndex.set(toolIndex, replacement); + toolArgumentBytesByIndex.set(toolIndex, retained.valueBytes); + retainedToolArgumentBytes = retained.retainedTurnBytes; + return true; + }; + const appendToolArgumentDelta = (toolIndex: number, delta: string): boolean => { + const current = toolArgumentsByIndex.get(toolIndex) ?? ""; + const currentBytes = toolArgumentBytesByIndex.get(toolIndex) ?? 0; + const appended = appendToolArguments(current, currentBytes, retainedToolArgumentBytes, delta); + if (!appended) { + failToolArgumentOverflow(); + return false; + } + toolArgumentsByIndex.set(toolIndex, appended.value); + toolArgumentBytesByIndex.set(toolIndex, appended.valueBytes); + retainedToolArgumentBytes = appended.retainedTurnBytes; + return true; + }; const handleFrame = (eventName: string, data: Rec) => { switch (eventName) { @@ -304,10 +341,10 @@ export function responsesSseToChatCompletionsSse( if (typeof item.id === "string") toolIndexByItemId.set(item.id, toolIndex); if (name) toolNameByIndex.set(toolIndex, name); if (!toolArgumentsByIndex.has(toolIndex)) { - toolArgumentsByIndex.set( + if (!retainToolArguments( toolIndex, typeof item.arguments === "string" ? item.arguments : "", - ); + )) break; } break; } @@ -317,10 +354,7 @@ export function responsesSseToChatCompletionsSse( const toolIndex = (itemId ? toolIndexByItemId.get(itemId) : undefined) ?? (nextToolIndex > 0 ? nextToolIndex - 1 : 0); ensureRole(); - toolArgumentsByIndex.set( - toolIndex, - (toolArgumentsByIndex.get(toolIndex) ?? "") + data.delta, - ); + if (!appendToolArgumentDelta(toolIndex, data.delta)) break; break; } case "response.function_call_arguments.done": { @@ -331,10 +365,10 @@ export function responsesSseToChatCompletionsSse( const name = typeof data.name === "string" ? data.name : ""; if (name) toolNameByIndex.set(toolIndex, name); const streamedArgs = toolArgumentsByIndex.get(toolIndex) ?? ""; - toolArgumentsByIndex.set( + if (!retainToolArguments( toolIndex, resolveFinalArguments(data.arguments, streamedArgs), - ); + )) break; break; } case "response.output_item.done": { @@ -361,7 +395,7 @@ export function responsesSseToChatCompletionsSse( // accumulate them. Emit one complete call here; finish() flushes any item whose // done event was omitted, and replace-style clients still get a complete object. const args = resolveFinalArguments(item.arguments, streamedArgs); - toolArgumentsByIndex.set(toolIndex, args); + if (!retainToolArguments(toolIndex, args)) break; emitToolCall(toolIndex, callId, finalName, args); } break; @@ -546,84 +580,96 @@ export async function collectChatCompletion( stream: ReadableStream, model: string, ): Promise { - const decoder = new TextDecoder(); - let buffer = ""; let content = ""; let reasoning = ""; - const toolCalls = new Map(); + const toolCalls = new Map(); + let retainedToolArgumentBytes = 0; let finishReason = "stop"; let usage: unknown; let streamError: ChatCompletionsStreamError | null = null; - const reader = stream.getReader(); try { - for (;;) { - let done = false; - let value: Uint8Array | undefined; - try { - ({ done, value } = await reader.read()); - } catch (err) { - if (isChatCompletionsStreamError(err)) throw err; - throw new ChatCompletionsStreamError(err instanceof Error ? err.message : String(err)); + for await (const record of decodeServerSentEvents(stream)) { + const data = record.data.trim(); + if (!data || data === "[DONE]") continue; + let parsed: unknown; + try { parsed = JSON.parse(data); } catch { continue; } + if (!isRec(parsed)) continue; + if (isRec(parsed.error)) { + const message = typeof parsed.error.message === "string" + ? parsed.error.message + : "upstream request failed"; + const type = typeof parsed.error.type === "string" ? parsed.error.type : "server_error"; + const code = typeof parsed.error.code === "string" ? parsed.error.code : null; + const status = code === CYBER_POLICY_ERROR_CODE || isCyberPolicyMessage(message) + ? 400 + : streamErrorStatus(message); + streamError = new ChatCompletionsStreamError(message, { status, type, code }); + continue; } - if (done) break; - if (!value) continue; - buffer += decoder.decode(value, { stream: true }); - let sep: number; - while ((sep = buffer.indexOf("\n\n")) !== -1) { - const rawFrame = buffer.slice(0, sep); - buffer = buffer.slice(sep + 2); - for (const line of rawFrame.split("\n")) { - if (!line.startsWith("data: ")) continue; - const data = line.slice(6).trim(); - if (!data || data === "[DONE]") continue; - let parsed: unknown; - try { parsed = JSON.parse(data); } catch { continue; } - if (!isRec(parsed)) continue; - if (isRec(parsed.error)) { - const message = typeof parsed.error.message === "string" - ? parsed.error.message - : "upstream request failed"; - const type = typeof parsed.error.type === "string" ? parsed.error.type : "server_error"; - const code = typeof parsed.error.code === "string" ? parsed.error.code : null; - const status = code === CYBER_POLICY_ERROR_CODE || isCyberPolicyMessage(message) - ? 400 - : streamErrorStatus(message); - streamError = new ChatCompletionsStreamError(message, { status, type, code }); - continue; - } - if (parsed.usage) usage = parsed.usage; - const choices = Array.isArray(parsed.choices) ? parsed.choices : []; - const choice = isRec(choices[0]) ? choices[0] : null; - if (!choice) continue; - if (typeof choice.finish_reason === "string") finishReason = choice.finish_reason; - const delta = isRec(choice.delta) ? choice.delta : null; - if (!delta) continue; - if (typeof delta.content === "string") content += delta.content; - if (typeof delta.reasoning_content === "string") reasoning += delta.reasoning_content; - if (Array.isArray(delta.tool_calls)) { - for (const tc of delta.tool_calls) { - if (!isRec(tc)) continue; - const index = typeof tc.index === "number" ? tc.index : 0; - const current = toolCalls.get(index) ?? { id: "", name: "", arguments: "" }; - if (typeof tc.id === "string") current.id = tc.id; - const fn = isRec(tc.function) ? tc.function : {}; - // Done-frame final arguments are authoritative last-write-wins snapshots. - if (typeof fn.name === "string" && fn.name.length > 0) current.name = fn.name; - if (typeof fn.arguments === "string") { - if (fn.arguments.startsWith("{") || fn.arguments.startsWith("[") || current.arguments.length === 0) { - current.arguments = fn.arguments; - } else { - current.arguments += fn.arguments; - } + if (parsed.usage) usage = parsed.usage; + const choices = Array.isArray(parsed.choices) ? parsed.choices : []; + const choice = isRec(choices[0]) ? choices[0] : null; + if (!choice) continue; + if (typeof choice.finish_reason === "string") finishReason = choice.finish_reason; + const delta = isRec(choice.delta) ? choice.delta : null; + if (!delta) continue; + if (typeof delta.content === "string") content += delta.content; + if (typeof delta.reasoning_content === "string") reasoning += delta.reasoning_content; + if (Array.isArray(delta.tool_calls)) { + for (const tc of delta.tool_calls) { + if (!isRec(tc)) continue; + const index = typeof tc.index === "number" ? tc.index : 0; + const current = toolCalls.get(index) ?? { id: "", name: "", arguments: "", argumentBytes: 0 }; + if (typeof tc.id === "string") current.id = tc.id; + const fn = isRec(tc.function) ? tc.function : {}; + // Done-frame final arguments are authoritative last-write-wins snapshots. + if (typeof fn.name === "string" && fn.name.length > 0) current.name = fn.name; + if (typeof fn.arguments === "string") { + const snapshot = fn.arguments.startsWith("{") || fn.arguments.startsWith("[") || current.arguments.length === 0; + if (snapshot) { + const retained = replaceToolArguments( + current.argumentBytes, + retainedToolArgumentBytes, + fn.arguments, + ); + if (!retained) { + toolCalls.clear(); + retainedToolArgumentBytes = 0; + throw new ChatCompletionsStreamError( + "upstream tool call arguments exceeded the configured byte limit", + { status: 502, type: "upstream_error" }, + ); + } + current.arguments = fn.arguments; + current.argumentBytes = retained.valueBytes; + retainedToolArgumentBytes = retained.retainedTurnBytes; + } else { + const appended = appendToolArguments( + current.arguments, + current.argumentBytes, + retainedToolArgumentBytes, + fn.arguments, + ); + if (!appended) { + toolCalls.clear(); + retainedToolArgumentBytes = 0; + throw new ChatCompletionsStreamError( + "upstream tool call arguments exceeded the configured byte limit", + { status: 502, type: "upstream_error" }, + ); } - toolCalls.set(index, current); + current.arguments = appended.value; + current.argumentBytes = appended.valueBytes; + retainedToolArgumentBytes = appended.retainedTurnBytes; } } + toolCalls.set(index, current); } } } - } finally { - reader.releaseLock(); + } catch (err) { + if (isChatCompletionsStreamError(err)) throw err; + throw new ChatCompletionsStreamError(err instanceof Error ? err.message : String(err)); } if (streamError) throw streamError; diff --git a/src/lib/sse-decoder.ts b/src/lib/sse-decoder.ts index 7ab8579e4..7e8272ce5 100644 --- a/src/lib/sse-decoder.ts +++ b/src/lib/sse-decoder.ts @@ -3,6 +3,16 @@ export interface ServerSentEvent { data: string; } +export const MAX_SSE_RECORD_BYTES = 4 * 1024 * 1024; + +export class SseRecordTooLargeError extends Error { + readonly code = "sse_record_too_large"; + constructor(limit: number) { + super(`upstream SSE record exceeded the ${limit}-byte limit`); + this.name = "SseRecordTooLargeError"; + } +} + export type SseRecord = | { kind: "event"; event?: string; data: string } | { kind: "comment"; comment: string }; @@ -16,21 +26,23 @@ export type SseRecord = */ export function decodeServerSentEvents( source: ReadableStream, - options: { includeComments: true; signal?: AbortSignal }, + options: { includeComments: true; signal?: AbortSignal; maxRecordBytes?: number }, ): AsyncGenerator; export function decodeServerSentEvents( source: ReadableStream, - options?: { includeComments?: false; signal?: AbortSignal }, + options?: { includeComments?: false; signal?: AbortSignal; maxRecordBytes?: number }, ): AsyncGenerator; export async function* decodeServerSentEvents( source: ReadableStream, - options?: { includeComments?: boolean; signal?: AbortSignal }, + options?: { includeComments?: boolean; signal?: AbortSignal; maxRecordBytes?: number }, ): AsyncGenerator { const reader = source.getReader(); const decoder = new TextDecoder(); let buffer = ""; let event: string | undefined; let dataLines: string[] = []; + let recordBytes = 0; + const maxRecordBytes = options?.maxRecordBytes ?? MAX_SSE_RECORD_BYTES; // Prompt cancellation channel: an abort cancels the underlying reader directly, which // settles any in-flight read() so a consumer's iterator.return() cannot hang behind an // idle upstream (a plain generator return waits for the pending await first). @@ -49,6 +61,7 @@ export async function* decodeServerSentEvents( const record = { ...(event ? { event } : {}), data: dataLines.join("\n") }; event = undefined; dataLines = []; + recordBytes = 0; return includeComments ? { kind: "event", ...record } : record; }; @@ -68,7 +81,12 @@ export async function* decodeServerSentEvents( if (value.startsWith(" ")) value = value.slice(1); if (field === "event") event = value; - else if (field === "data") dataLines.push(value); + else if (field === "data") { + const added = Buffer.byteLength(value, "utf8") + (dataLines.length > 0 ? 1 : 0); + if (recordBytes + added > maxRecordBytes) throw new SseRecordTooLargeError(maxRecordBytes); + recordBytes += added; + dataLines.push(value); + } return undefined; }; @@ -79,14 +97,21 @@ export async function* decodeServerSentEvents( let newline: number; while ((newline = buffer.indexOf("\n")) >= 0) { - const record = acceptLine(buffer.slice(0, newline)); + const line = buffer.slice(0, newline); + if (Buffer.byteLength(line, "utf8") > maxRecordBytes) throw new SseRecordTooLargeError(maxRecordBytes); + const record = acceptLine(line); buffer = buffer.slice(newline + 1); if (record) yield record; } + if (Buffer.byteLength(buffer, "utf8") > maxRecordBytes) { + throw new SseRecordTooLargeError(maxRecordBytes); + } + if (!done) continue; buffer += decoder.decode(); if (buffer.length > 0) { + if (Buffer.byteLength(buffer, "utf8") > maxRecordBytes) throw new SseRecordTooLargeError(maxRecordBytes); const record = acceptLine(buffer); buffer = ""; if (record) yield record; diff --git a/src/lib/tool-argument-bounds.ts b/src/lib/tool-argument-bounds.ts new file mode 100644 index 000000000..57fd81572 --- /dev/null +++ b/src/lib/tool-argument-bounds.ts @@ -0,0 +1,55 @@ +export interface ToolArgumentLimits { + perCallBytes: number; + perTurnBytes: number; +} + +const DEFAULT_TOOL_ARGUMENT_LIMITS: ToolArgumentLimits = { + perCallBytes: 8 * 1024 * 1024, + perTurnBytes: 32 * 1024 * 1024, +}; + +let limits = { ...DEFAULT_TOOL_ARGUMENT_LIMITS }; + +export function currentToolArgumentLimits(): ToolArgumentLimits { + return limits; +} + +export function utf8ByteLength(value: string): number { + return Buffer.byteLength(value, "utf8"); +} + +export function appendToolArguments( + current: string, + currentBytes: number, + retainedTurnBytes: number, + fragment: string, +): { value: string; valueBytes: number; retainedTurnBytes: number } | null { + const fragmentBytes = utf8ByteLength(fragment); + if ( + currentBytes + fragmentBytes > limits.perCallBytes + || retainedTurnBytes + fragmentBytes > limits.perTurnBytes + ) return null; + return { + value: current + fragment, + valueBytes: currentBytes + fragmentBytes, + retainedTurnBytes: retainedTurnBytes + fragmentBytes, + }; +} + +export function replaceToolArguments( + currentBytes: number, + retainedTurnBytes: number, + replacement: string, +): { valueBytes: number; retainedTurnBytes: number } | null { + const valueBytes = utf8ByteLength(replacement); + const nextTurnBytes = retainedTurnBytes - currentBytes + valueBytes; + if (valueBytes > limits.perCallBytes || nextTurnBytes > limits.perTurnBytes) return null; + return { valueBytes, retainedTurnBytes: Math.max(0, nextTurnBytes) }; +} + +/** Test seam: lower bounds without allocating production-sized fixtures. */ +export function setToolArgumentLimitsForTests(overrides: Partial | null): void { + limits = overrides + ? { ...DEFAULT_TOOL_ARGUMENT_LIMITS, ...overrides } + : { ...DEFAULT_TOOL_ARGUMENT_LIMITS }; +} diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 5a39c59e2..7aabfb822 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -374,6 +374,29 @@ adapters advertise the catalog bit only on explicit `true`; cursor keeps its own Providers with flaky parallel streaming can be opted out individually. Evidence and provider ledger: `devlog/_fin/260709_parallel_tool_calls/`. +## Streamed tool-argument and SSE memory bounds + +SSE records decoded by the shared transport parser are capped at 4 MiB. OpenAI Chat's compatible +line parser applies the same cap to complete and unterminated records. Tool-call arguments retained +by the OpenAI Chat adapter, Responses bridge, Responses-to-Chat translator, and non-stream Chat +collector are independently capped at 8 MiB per call and 32 MiB across one turn, measured as UTF-8 +bytes. Final argument snapshots replace their call's prior byte ownership instead of double-counting +the streamed fragments. + +Crossing a bound clears the affected live argument buffers and terminates with a typed 502 upstream +error. An open Responses function call is finalized as `incomplete`; no arguments-done, completed +tool item, successful finish reason, or Chat `[DONE]` marker is synthesized after the overflow. + +```text +[Decision Log] +- 목적과 의도: Bound provider-controlled active-turn memory without presenting a truncated tool call as executable output. +- 기존 구현 및 제약 조건: Compatible providers can stream arbitrarily many argument fragments or omit an SSE delimiter; the bridge must preserve its sequential tool-call contract and authoritative final snapshots. +- 검토한 주요 대안: Keep relying on upstream token limits; truncate and complete the call; redesign every adapter around keyed live parallel calls; impose byte ownership at the existing buffering boundaries. +- 선택한 방식: Enforce shared UTF-8 per-call/per-turn argument budgets plus a shared SSE-record budget, and fail the turn before emitting completion when a budget is exceeded. +- 다른 대안 대신 이 방식을 선택한 이유: Upstream defaults are inconsistent, truncation produces invalid or dangerous calls, and a parallel event-contract rewrite is unrelated to bounding retained memory. +- 장점, 단점 및 영향: Malformed providers cannot grow these buffers without bound and clients receive truthful failure semantics; exceptionally large legitimate calls now fail explicitly and must be split upstream. +``` + ## Reasoning display parity (hideThinkingSummary) `hideThinkingSummary` (request reasoning summary absent/"none" — the routed catalog default) is diff --git a/tests/sse-decoder.test.ts b/tests/sse-decoder.test.ts index 2029260be..ca833a88c 100644 --- a/tests/sse-decoder.test.ts +++ b/tests/sse-decoder.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { decodeServerSentEvents } from "../src/lib/sse-decoder"; +import { decodeServerSentEvents, SseRecordTooLargeError } from "../src/lib/sse-decoder"; function chunkedStream(chunks: string[]): ReadableStream { const encoder = new TextEncoder(); @@ -73,4 +73,36 @@ describe("text/event-stream decoder", () => { expect(records).toEqual([{ event: "custom", data: "payload" }]); expect("kind" in records[0]).toBe(false); }); + + test("rejects an oversized unterminated record instead of retaining it to EOF", async () => { + const read = async () => { + for await (const _record of decodeServerSentEvents( + chunkedStream(["data: 12345678901234567"]), + { maxRecordBytes: 16 }, + )) { /* drain */ } + }; + + await expect(read()).rejects.toBeInstanceOf(SseRecordTooLargeError); + }); + + test("counts multiline data as one bounded SSE record", async () => { + const read = async () => { + for await (const _record of decodeServerSentEvents( + chunkedStream(["data: 12345678\n", "data: 12345678\n\n"]), + { maxRecordBytes: 16 }, + )) { /* drain */ } + }; + + await expect(read()).rejects.toBeInstanceOf(SseRecordTooLargeError); + }); + + test("accepts a complete record at the configured raw-line boundary", async () => { + const records = []; + for await (const record of decodeServerSentEvents( + chunkedStream(["data: 1234567890\n\n"]), + { maxRecordBytes: 16 }, + )) records.push(record); + + expect(records).toEqual([{ data: "1234567890" }]); + }); }); diff --git a/tests/tool-argument-bounds.test.ts b/tests/tool-argument-bounds.test.ts new file mode 100644 index 000000000..77ab02155 --- /dev/null +++ b/tests/tool-argument-bounds.test.ts @@ -0,0 +1,135 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { createOpenAIChatAdapter } from "../src/adapters/openai-chat"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../src/bridge"; +import { + ChatCompletionsStreamError, + collectChatCompletion, + responsesSseToChatCompletionsSse, +} from "../src/chat/outbound"; +import { setToolArgumentLimitsForTests } from "../src/lib/tool-argument-bounds"; +import type { AdapterEvent } from "../src/types"; + +const encoder = new TextEncoder(); + +function byteStream(chunks: string[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + }, + }); +} + +async function* replay(events: AdapterEvent[]): AsyncGenerator { + for (const event of events) yield event; +} + +async function streamText(stream: ReadableStream): Promise { + return new Response(stream).text(); +} + +function responseSse(event: string, data: Record): string { + return `event: ${event}\ndata: ${JSON.stringify({ type: event, ...data })}\n\n`; +} + +afterEach(() => setToolArgumentLimitsForTests(null)); + +describe("streamed tool argument memory bounds", () => { + test("openai-chat rejects one oversized buffered call without flushing it as complete", async () => { + setToolArgumentLimitsForTests({ perCallBytes: 16, perTurnBytes: 32 }); + const body = [ + { choices: [{ delta: { tool_calls: [{ index: 0, id: "call_1", function: { name: "shell", arguments: "{\"value\":\"" } }] } }] }, + { choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: "1234567890\"}" } }] }, finish_reason: "tool_calls" }] }, + ].map(chunk => `data: ${JSON.stringify(chunk)}\n\n`).join("") + "data: [DONE]\n\n"; + const events: AdapterEvent[] = []; + + for await (const event of createOpenAIChatAdapter({ + adapter: "openai-chat", + baseUrl: "https://example.test/v1", + apiKey: "key", + }).parseStream(new Response(body))) events.push(event); + + expect(events.map(event => event.type)).toEqual(["error"]); + expect(events[0]).toMatchObject({ status: 502, errorType: "upstream_error" }); + }); + + test("Responses streaming bridge fails the turn instead of completing an oversized call", async () => { + setToolArgumentLimitsForTests({ perCallBytes: 16, perTurnBytes: 32 }); + const text = await streamText(bridgeToResponsesSSE(replay([ + { type: "tool_call_start", id: "call_1", name: "shell" }, + { type: "tool_call_delta", arguments: "{\"value\":\"" }, + { type: "tool_call_delta", arguments: "1234567890\"}" }, + { type: "tool_call_end" }, + { type: "done" }, + ]), "routed/model")); + + expect(text).toContain("response.failed"); + expect(text).toContain("configured byte limit"); + expect(text).not.toContain("response.function_call_arguments.done"); + expect(text).not.toContain("response.completed"); + }); + + test("batch bridge enforces the aggregate retained budget across completed calls", () => { + setToolArgumentLimitsForTests({ perCallBytes: 16, perTurnBytes: 20 }); + const response = buildResponseJSON([ + { type: "tool_call_start", id: "call_1", name: "first" }, + { type: "tool_call_delta", arguments: "{\"a\":\"123\"}" }, + { type: "tool_call_end" }, + { type: "tool_call_start", id: "call_2", name: "second" }, + { type: "tool_call_delta", arguments: "{\"b\":\"456\"}" }, + { type: "tool_call_end" }, + { type: "done" }, + ], "routed/model"); + + expect(response.status).toBe("failed"); + expect((response.error as { message?: string }).message).toContain("configured byte limit"); + const output = response.output as Array<{ call_id?: string; status?: string; arguments?: string }>; + expect(output).toContainEqual(expect.objectContaining({ call_id: "call_1", status: "completed" })); + expect(output).toContainEqual(expect.objectContaining({ call_id: "call_2", status: "incomplete", arguments: "{}" })); + }); + + test("Responses-to-Chat streaming closes with an error and no DONE on overflow", async () => { + setToolArgumentLimitsForTests({ perCallBytes: 16, perTurnBytes: 32 }); + const upstream = byteStream([ + responseSse("response.output_item.added", { + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "shell", arguments: "" }, + }), + responseSse("response.function_call_arguments.delta", { + item_id: "fc_1", + delta: "{\"value\":\"1234567890\"}", + }), + responseSse("response.completed", { response: { status: "completed" } }), + ]); + const text = await streamText(responsesSseToChatCompletionsSse(upstream, "routed/model")); + + expect(text).toContain('"error"'); + expect(text).toContain("configured byte limit"); + expect(text).not.toContain("data: [DONE]"); + expect(text).not.toContain('"finish_reason":"tool_calls"'); + }); + + test("non-stream Chat collector rejects oversized tool arguments with a typed error", async () => { + setToolArgumentLimitsForTests({ perCallBytes: 16, perTurnBytes: 32 }); + const chatFrame = { + choices: [{ + index: 0, + delta: { + tool_calls: [{ + index: 0, + id: "call_1", + type: "function", + function: { name: "shell", arguments: "{\"value\":\"1234567890\"}" }, + }], + }, + finish_reason: null, + }], + }; + const completion = collectChatCompletion( + byteStream([`data: ${JSON.stringify(chatFrame)}\n\ndata: [DONE]\n\n`]), + "routed/model", + ); + + await expect(completion).rejects.toBeInstanceOf(ChatCompletionsStreamError); + await expect(completion).rejects.toMatchObject({ status: 502, type: "upstream_error" }); + }); +});