diff --git a/packages/ai/src/providers/openai-completions-tool-calls.ts b/packages/ai/src/providers/openai-completions-tool-calls.ts new file mode 100644 index 000000000000..81672c4a7fc7 --- /dev/null +++ b/packages/ai/src/providers/openai-completions-tool-calls.ts @@ -0,0 +1,234 @@ +import { randomUUID } from "node:crypto"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import type { ChatCompletionChunk } from "openai/resources/chat/completions.js"; + +type ChatCompletionToolCallDelta = ChatCompletionChunk.Choice.Delta.ToolCall; +const MAX_BUFFERED_LEGACY_TOOL_CALL_ARGUMENT_BYTES = 256_000; +const MAX_BUFFERED_LEGACY_FOLLOWING_DELTA_BYTES = 256_000; +const MAX_BUFFERED_LEGACY_FOLLOWING_DELTAS = 1_024; + +type NormalizedOpenAICompletionsDelta = { + delta: ChatCompletionChunk.Choice.Delta; + toolCalls: ChatCompletionToolCallDelta[]; +}; + +type OpenAICompletionsToolCallFinalizationOptions = { + allowSilentToolCallPromotion?: boolean; + onConfirmedToolCall?: (block: TBlock, contentIndex: number) => void; +}; + +/** Normalize the SDK's legacy single-function lane into its modern tool-call shape. */ +export function createOpenAICompletionsToolCallDeltaNormalizer(): ( + delta: ChatCompletionChunk.Choice.Delta, + finishReason: ChatCompletionChunk.Choice["finish_reason"], +) => NormalizedOpenAICompletionsDelta[] { + let sawModernToolCall = false; + let pendingLegacyArgumentBytes = 0; + let pendingFollowingDeltaBytes = 0; + let pendingLegacyToolCall: ChatCompletionToolCallDelta | undefined; + const pendingFollowingDeltas: ChatCompletionChunk.Choice.Delta[] = []; + + const takePendingFollowingDeltas = (): NormalizedOpenAICompletionsDelta[] => { + pendingFollowingDeltaBytes = 0; + return pendingFollowingDeltas.splice(0).map((delta) => ({ delta, toolCalls: [] })); + }; + + const bufferFollowingDelta = (delta: ChatCompletionChunk.Choice.Delta): void => { + const nextDeltaBytes = Buffer.byteLength(JSON.stringify(delta), "utf8"); + if ( + pendingFollowingDeltaBytes + nextDeltaBytes > MAX_BUFFERED_LEGACY_FOLLOWING_DELTA_BYTES || + pendingFollowingDeltas.length >= MAX_BUFFERED_LEGACY_FOLLOWING_DELTAS + ) { + throw new Error("Exceeded legacy tool-call content buffer limit"); + } + pendingFollowingDeltaBytes += nextDeltaBytes; + pendingFollowingDeltas.push(delta); + }; + + const withoutToolCalls = ( + delta: ChatCompletionChunk.Choice.Delta, + ): ChatCompletionChunk.Choice.Delta => { + const ordinaryDelta = { ...delta }; + delete ordinaryDelta.function_call; + delete ordinaryDelta.tool_calls; + return ordinaryDelta; + }; + + const hasObservableContent = (value: unknown): boolean => { + if (typeof value === "string") { + return value.length > 0; + } + if (Array.isArray(value)) { + return value.some(hasObservableContent); + } + if (isRecord(value)) { + return Object.entries(value).some( + ([field, nestedValue]) => + field !== "type" && + field !== "id" && + field !== "index" && + hasObservableContent(nestedValue), + ); + } + return false; + }; + + const hasOrdinaryContent = (delta: ChatCompletionChunk.Choice.Delta): boolean => + Object.entries(delta).some(([field, value]) => field !== "role" && hasObservableContent(value)); + + return (delta, finishReason) => { + const ordinaryDelta = withoutToolCalls(delta); + if (delta.tool_calls && delta.tool_calls.length > 0) { + const precedingDeltas = takePendingFollowingDeltas(); + sawModernToolCall = true; + pendingLegacyArgumentBytes = 0; + pendingLegacyToolCall = undefined; + return [...precedingDeltas, { delta: ordinaryDelta, toolCalls: delta.tool_calls }]; + } + + const functionCall = delta.function_call; + if (sawModernToolCall) { + return [{ delta: ordinaryDelta, toolCalls: [] }]; + } + + const hadPendingLegacyCall = pendingLegacyToolCall !== undefined; + const leadingDeltas: NormalizedOpenAICompletionsDelta[] = []; + if (hasOrdinaryContent(ordinaryDelta)) { + if (hadPendingLegacyCall) { + // Keep every lane behind its provisional call; publishing text early + // permanently reverses the assistant's original tool/content order. + bufferFollowingDelta(ordinaryDelta); + } else { + leadingDeltas.push({ delta: ordinaryDelta, toolCalls: [] }); + } + } + + if (functionCall && (functionCall.name || functionCall.arguments)) { + // Legacy events are provisional until their terminal reason confirms no + // modern call superseded them. Coalesce fragments so hostile empty/name + // frames cannot create an unbounded queue or quadratic terminal parsing. + const nextFunctionName = pendingLegacyToolCall?.function?.name + ? undefined + : functionCall.name; + const nextArgumentBytes = + Buffer.byteLength(functionCall.arguments ?? "", "utf8") + + Buffer.byteLength(nextFunctionName ?? "", "utf8"); + if ( + pendingLegacyArgumentBytes + nextArgumentBytes > + MAX_BUFFERED_LEGACY_TOOL_CALL_ARGUMENT_BYTES + ) { + throw new Error("Exceeded tool-call argument buffer limit"); + } + pendingLegacyArgumentBytes += nextArgumentBytes; + pendingLegacyToolCall ??= { + index: 0, + id: `call_${randomUUID().replaceAll("-", "").slice(0, 24)}`, + type: "function", + function: {}, + }; + const pendingFunction = (pendingLegacyToolCall.function ??= {}); + if (nextFunctionName) { + pendingFunction.name = nextFunctionName; + } + if (functionCall.arguments) { + pendingFunction.arguments = (pendingFunction.arguments ?? "") + functionCall.arguments; + } + } + + if (!pendingLegacyToolCall) { + return leadingDeltas.length > 0 ? leadingDeltas : [{ delta: ordinaryDelta, toolCalls: [] }]; + } + + if (finishReason === "function_call") { + const confirmedLegacyToolCall = pendingLegacyToolCall; + pendingLegacyToolCall = undefined; + pendingLegacyArgumentBytes = 0; + return [ + ...leadingDeltas, + { delta: {}, toolCalls: [confirmedLegacyToolCall] }, + ...takePendingFollowingDeltas(), + ]; + } + + if (finishReason) { + pendingLegacyArgumentBytes = 0; + pendingLegacyToolCall = undefined; + return [...leadingDeltas, ...takePendingFollowingDeltas()]; + } + + return leadingDeltas; + }; +} + +/** Publish only executable calls; streaming scratch state never belongs in replay. */ +export function finalizeOpenAICompletionsToolCalls( + output: { content: TBlock[]; stopReason: string; errorMessage?: string }, + options: OpenAICompletionsToolCallFinalizationOptions = {}, +): void { + const isToolCall = (block: TBlock) => (block as { type?: unknown }).type === "toolCall"; + const hasToolCalls = output.content.some(isToolCall); + + if (output.stopReason === "toolUse" && !hasToolCalls) { + output.stopReason = "stop"; + } + + if ( + output.stopReason === "stop" && + hasToolCalls && + options.allowSilentToolCallPromotion !== false && + !output.content.some((block) => { + const candidate = block as { type?: unknown; text?: unknown }; + return ( + candidate.type === "text" && + typeof candidate.text === "string" && + candidate.text.trim().length > 0 + ); + }) + ) { + output.stopReason = "toolUse"; + } + + if (output.stopReason !== "toolUse") { + if (hasToolCalls) { + output.content = output.content.filter((block) => !isToolCall(block)); + } + return; + } + + for (const block of output.content) { + if (!isToolCall(block)) { + continue; + } + const toolCall = block as { name?: unknown; arguments?: unknown; partialArgs?: unknown }; + let completeArguments: unknown; + try { + completeArguments = + typeof toolCall.partialArgs === "string" && toolCall.partialArgs.trim().length > 0 + ? (JSON.parse(toolCall.partialArgs) as unknown) + : undefined; + } catch { + completeArguments = undefined; + } + if ( + typeof toolCall.name !== "string" || + toolCall.name.trim().length === 0 || + !isRecord(completeArguments) + ) { + output.stopReason = "error"; + output.errorMessage = "Provider returned an incomplete or malformed tool call"; + output.content = output.content.filter((candidate) => !isToolCall(candidate)); + return; + } + toolCall.arguments = completeArguments; + } + + for (let contentIndex = 0; contentIndex < output.content.length; contentIndex += 1) { + const block = output.content[contentIndex]; + if (!block || !isToolCall(block)) { + continue; + } + delete (block as { partialArgs?: string }).partialArgs; + delete (block as { streamIndex?: number }).streamIndex; + options.onConfirmedToolCall?.(block, contentIndex); + } +} diff --git a/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts b/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts new file mode 100644 index 000000000000..53a79b012f09 --- /dev/null +++ b/packages/ai/src/providers/openai-completions.legacy-function-call.test.ts @@ -0,0 +1,700 @@ +import type { ChatCompletionChunk } from "openai/resources/chat/completions.js"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { configureAiTransportHost, getAiTransportHost } from "../host.js"; +import { createOpenAICompletionsTransportStreamFn } from "../transports/openai-completions-transport.js"; +import type { AssistantMessageEventStreamLike, Context, Model } from "../types.js"; +import { streamOpenAICompletions } from "./openai-completions.js"; + +const model = { + id: "gpt-4", + name: "GPT-4", + api: "openai-completions", + provider: "openai", + baseUrl: "https://api.openai.com/v1", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 8_192, + maxTokens: 1_024, +} satisfies Model<"openai-completions">; + +const context = { + messages: [{ role: "user", content: "Look up cats", timestamp: 1 }], + tools: [ + { + name: "lookup", + description: "Look up a query", + parameters: { + type: "object", + properties: { query: { type: "string" } }, + required: ["query"], + }, + }, + ], +} satisfies Context; + +function chunk( + delta: ChatCompletionChunk.Choice.Delta, + finishReason: ChatCompletionChunk.Choice["finish_reason"] = null, +): ChatCompletionChunk { + return { + id: "chatcmpl-legacy-fixture", + object: "chat.completion.chunk", + created: 0, + model: model.id, + choices: [{ index: 0, delta, finish_reason: finishReason }], + }; +} + +function installStream(chunks: ChatCompletionChunk[]): void { + const body = `${chunks.map((value) => `data: ${JSON.stringify(value)}\n\n`).join("")}data: [DONE]\n\n`; + const fetch = vi.fn(async () => { + return new Response(body, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); + }); + configureAiTransportHost({ buildModelFetch: () => fetch }); +} + +let previousHost: ReturnType; + +beforeEach(() => { + previousHost = getAiTransportHost(); +}); + +afterEach(() => { + configureAiTransportHost(previousHost); +}); + +const createManagedStream = createOpenAICompletionsTransportStreamFn(); + +function createManagedFixtureStream( + fixtureModel: Model<"openai-completions">, + fixtureContext: Context, + fixtureOptions?: Parameters[2], +): AssistantMessageEventStreamLike { + const stream = createManagedStream(fixtureModel, fixtureContext, fixtureOptions); + if (stream instanceof Promise) { + throw new Error("OpenAI Chat transport must return its event stream synchronously"); + } + return stream; +} + +describe.each([ + { name: "package", createStream: streamOpenAICompletions }, + { name: "managed", createStream: createManagedFixtureStream }, +])("$name OpenAI Chat Completions stream", ({ createStream }) => { + it("preserves legacy function_call deltas and reassembles split arguments", async () => { + installStream([ + chunk({ role: "assistant", function_call: { name: "lookup" } }), + chunk({ function_call: { arguments: '{"query":"ca' } }), + chunk({ function_call: { arguments: 'ts"}' } }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content).toHaveLength(1); + expect(result.content[0]).toMatchObject({ + type: "toolCall", + id: expect.stringMatching(/^call_[a-f0-9]{24}$/), + name: "lookup", + arguments: { query: "cats" }, + }); + expect(result.content[0]).not.toHaveProperty("partialArgs"); + expect(result.content[0]).not.toHaveProperty("streamIndex"); + expect(eventTypes).toContain("toolcall_start"); + expect(eventTypes).toContain("toolcall_delta"); + }); + + it("preserves a confirmed legacy call before later visible text", async () => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"cats"}' } }), + chunk({ content: "Trailing commentary." }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content.map((block) => block.type)).toEqual(["toolCall", "text"]); + expect(result.content[1]).toMatchObject({ text: "Trailing commentary." }); + expect(eventTypes.indexOf("toolcall_start")).toBeLessThan(eventTypes.indexOf("text_start")); + }); + + it("keeps content preceding a legacy call in the same provider delta", async () => { + installStream([ + chunk({ + content: "Commentary before the call.", + function_call: { name: "lookup", arguments: '{"query":"cats"}' }, + }), + chunk({}, "function_call"), + ]); + + const result = await createStream(model, context, { apiKey: "fixture-token" }).result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content.map((block) => block.type)).toEqual(["text", "toolCall"]); + expect(result.content[0]).toMatchObject({ text: "Commentary before the call." }); + }); + + it("preserves a confirmed legacy call before later streamed reasoning", async () => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"cats"}' } }), + chunk({ reasoning_content: "Reasoning after the call." } as ChatCompletionChunk.Choice.Delta), + chunk({}, "function_call"), + ]); + + const stream = createStream({ ...model, reasoning: true }, context, { + apiKey: "fixture-token", + reasoningEffort: "medium", + }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content.map((block) => block.type)).toEqual(["toolCall", "thinking"]); + expect(eventTypes.indexOf("toolcall_start")).toBeLessThan(eventTypes.indexOf("thinking_start")); + }); + + it("replays buffered visible text before a superseding modern call", async () => { + installStream([ + chunk({ function_call: { name: "legacy", arguments: '{"query":"wrong"}' } }), + chunk({ content: "Visible before the modern call." }), + chunk({ + tool_calls: [ + { + index: 0, + id: "call_modern", + type: "function", + function: { name: "lookup", arguments: '{"query":"cats"}' }, + }, + ], + }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const observedStarts: string[] = []; + for await (const event of stream) { + if (event.type === "toolcall_start") { + const block = event.partial.content[event.contentIndex]; + if (block?.type === "toolCall") { + observedStarts.push(block.id); + } + } + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content.map((block) => block.type)).toEqual(["text", "toolCall"]); + expect(result.content[0]).toMatchObject({ text: "Visible before the modern call." }); + expect(observedStarts).toEqual(["call_modern"]); + }); + + it("releases buffered visible text when a legacy call is not confirmed", async () => { + installStream([ + chunk({ function_call: { name: "legacy", arguments: '{"query":"discard"}' } }), + chunk({ content: "The provider supplied an answer instead." }), + chunk({}, "stop"), + ]); + + const result = await createStream(model, context, { apiKey: "fixture-token" }).result(); + + expect(result.stopReason).toBe("stop"); + expect(result.content).toEqual([ + { type: "text", text: "The provider supplied an answer instead." }, + ]); + }); + + it("keeps modern tool_calls authoritative when legacy data is also present", async () => { + installStream([ + chunk({ + function_call: { name: "legacy", arguments: '{"query":"wrong"}' }, + tool_calls: [ + { + index: 0, + id: "call_modern", + type: "function", + function: { name: "lookup", arguments: '{"query":"ca' }, + }, + ], + }), + chunk({ function_call: { arguments: "ignore this legacy continuation" } }), + chunk({ tool_calls: [{ index: 0, function: { arguments: 'ts"}' } }] }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const observedStarts: Array<{ id: string; name: string }> = []; + const observedArgumentDeltas: string[] = []; + for await (const event of stream) { + if (event.type === "toolcall_start") { + const block = event.partial.content[event.contentIndex]; + if (block?.type === "toolCall") { + observedStarts.push({ id: block.id, name: block.name }); + } + } else if (event.type === "toolcall_delta" && event.delta) { + observedArgumentDeltas.push(event.delta); + } + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content).toHaveLength(1); + expect(result.content[0]).toMatchObject({ + type: "toolCall", + id: "call_modern", + name: "lookup", + arguments: { query: "cats" }, + }); + expect(observedStarts).toEqual([{ id: "call_modern", name: "lookup" }]); + expect(observedArgumentDeltas).toEqual(['{"query":"ca', 'ts"}']); + }); + + it("replaces an earlier legacy function_call when modern tool_calls arrive later", async () => { + installStream([ + chunk({ function_call: { name: "legacy", arguments: '{"query":"wrong"}' } }), + chunk({ + tool_calls: [ + { + index: 1, + id: "call_modern", + type: "function", + function: { name: "lookup", arguments: '{"query":"ca' }, + }, + ], + }), + chunk({ tool_calls: [{ index: 1, function: { arguments: 'ts"}' } }] }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const observedStarts: Array<{ id: string; name: string }> = []; + const observedArgumentDeltas: string[] = []; + for await (const event of stream) { + if (event.type === "toolcall_start") { + const block = event.partial.content[event.contentIndex]; + if (block?.type === "toolCall") { + observedStarts.push({ id: block.id, name: block.name }); + } + } else if (event.type === "toolcall_delta" && event.delta) { + observedArgumentDeltas.push(event.delta); + } + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content).toHaveLength(1); + expect(result.content[0]).toMatchObject({ + type: "toolCall", + id: "call_modern", + name: "lookup", + arguments: { query: "cats" }, + }); + expect(observedStarts).toEqual([{ id: "call_modern", name: "lookup" }]); + expect(observedArgumentDeltas).toEqual(['{"query":"ca', 'ts"}']); + }); + + it("keeps ordinary text and stop finish reasons outside the tool-call lane", async () => { + installStream([chunk({ role: "assistant", content: "No tool needed." }), chunk({}, "stop")]); + + const result = await createStream(model, context, { apiKey: "fixture-token" }).result(); + + expect(result.stopReason).toBe("stop"); + expect(result.content).toEqual([{ type: "text", text: "No tool needed." }]); + }); + + it.each([ + { finishReason: "length", visibleText: false, stopReason: "length" }, + { finishReason: "content_filter", visibleText: false, stopReason: "error" }, + { finishReason: "stop", visibleText: true, stopReason: "stop" }, + ] as const)( + "does not publish completed modern calls discarded by a $finishReason terminal", + async ({ finishReason, visibleText, stopReason }) => { + installStream([ + ...(visibleText ? [chunk({ content: "Visible final answer." })] : []), + chunk({ + tool_calls: [ + { + index: 0, + id: "call_unconfirmed", + type: "function", + function: { name: "lookup", arguments: '{"query":"discard"}' }, + }, + ], + }), + chunk({}, finishReason), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + + const result = await stream.result(); + expect(result.stopReason).toBe(stopReason); + expect(result.content.filter((block) => block.type === "toolCall")).toHaveLength(0); + expect(eventTypes).not.toContain("toolcall_end"); + }, + ); + + it.each([ + { reason: "incomplete JSON", name: "lookup", arguments: '{"query":"cats"' }, + { reason: "malformed JSON", name: "lookup", arguments: '{"query":}' }, + { reason: "invalid string escape", name: "lookup", arguments: String.raw`{"query":"cats\q"}` }, + { reason: "unescaped control character", name: "lookup", arguments: '{"query":"cats\n"}' }, + { reason: "non-object JSON", name: "lookup", arguments: '["cats"]' }, + { reason: "empty arguments", name: "lookup", arguments: "" }, + { reason: "missing function name", name: "", arguments: '{"query":"cats"}' }, + ] as const)( + "rejects an authoritative modern tool terminal with $reason", + async ({ name, arguments: rawArguments }) => { + installStream([ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_malformed", + type: "function", + function: { name, arguments: rawArguments }, + }, + ], + }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("incomplete or malformed tool call"); + expect(result.content.filter((block) => block.type === "toolCall")).toHaveLength(0); + expect(eventTypes).not.toContain("toolcall_end"); + }, + ); + + it.each([ + { reason: "incomplete JSON", arguments: '{"query":"cats"' }, + { reason: "invalid string escape", arguments: String.raw`{"query":"cats\q"}` }, + { reason: "unescaped control character", arguments: '{"query":"cats\n"}' }, + ] as const)("rejects an authoritative legacy function terminal with $reason", async (value) => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: value.arguments } }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("incomplete or malformed tool call"); + expect(result.content.filter((block) => block.type === "toolCall")).toHaveLength(0); + expect(eventTypes).not.toContain("toolcall_end"); + }); + + it("rejects every parallel call when a sibling has incomplete arguments", async () => { + installStream([ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_valid", + type: "function", + function: { name: "lookup", arguments: '{"query":"cats"}' }, + }, + { + index: 1, + id: "call_invalid", + type: "function", + function: { name: "lookup", arguments: '{"query":"dogs"' }, + }, + ], + }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.content.filter((block) => block.type === "toolCall")).toHaveLength(0); + expect(eventTypes).not.toContain("toolcall_end"); + }); + + it("removes provisional calls when the response is aborted mid-stream", async () => { + installStream([ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_aborted", + type: "function", + function: { name: "lookup", arguments: '{"query":"cats"}' }, + }, + ], + }), + chunk({}, "tool_calls"), + ]); + + const abort = new AbortController(); + const stream = createStream(model, context, { apiKey: "fixture-token", signal: abort.signal }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + if (event.type === "toolcall_start") { + abort.abort(); + } + } + + const result = await stream.result(); + expect(result.stopReason).toBe("aborted"); + expect(result.content.filter((block) => block.type === "toolCall")).toHaveLength(0); + expect(eventTypes).not.toContain("toolcall_end"); + }); + + it("retains an executable modern call after its confirmed tool terminal", async () => { + installStream([ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_confirmed", + type: "function", + function: { name: "lookup", arguments: '{"query":"cats"}' }, + }, + ], + }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const events: Array<{ type: string; toolCallId?: string }> = []; + for await (const event of stream) { + events.push({ + type: event.type, + ...(event.type === "toolcall_end" ? { toolCallId: event.toolCall.id } : {}), + }); + } + + const result = await stream.result(); + expect(result.stopReason).toBe("toolUse"); + expect(result.content).toContainEqual({ + type: "toolCall", + id: "call_confirmed", + name: "lookup", + arguments: { query: "cats" }, + }); + expect(events.filter((event) => event.type === "toolcall_end")).toEqual([ + { type: "toolcall_end", toolCallId: "call_confirmed" }, + ]); + }); + + it("preserves confirmed tool completion before following text blocks close", async () => { + installStream([ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_before_text", + type: "function", + function: { name: "lookup", arguments: '{"query":"cats"}' }, + }, + ], + }), + chunk({ content: "Trailing commentary." }), + chunk({}, "tool_calls"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + + expect((await stream.result()).stopReason).toBe("toolUse"); + const toolEndIndex = eventTypes.indexOf("toolcall_end"); + expect(toolEndIndex).toBeGreaterThanOrEqual(0); + expect(toolEndIndex).toBeLessThan(eventTypes.indexOf("done")); + + const followingTextEndIndex = eventTypes.indexOf("text_end"); + if (followingTextEndIndex >= 0) { + expect(toolEndIndex).toBeLessThan(followingTextEndIndex); + } + }); + + it.each([ + { finishReason: "stop", stopReason: "stop" }, + { finishReason: "length", stopReason: "length" }, + { finishReason: "content_filter", stopReason: "error" }, + { finishReason: "tool_calls", stopReason: "stop" }, + ] as const)( + "discards provisional legacy fragments when the provider finishes with $finishReason", + async ({ finishReason, stopReason }) => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"discard"}' } }), + chunk({}, finishReason), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe(stopReason); + expect(result.content.filter((block) => block.type === "toolCall")).toHaveLength(0); + expect(eventTypes).not.toContain("toolcall_start"); + expect(eventTypes).not.toContain("toolcall_delta"); + }, + ); + + it("rejects oversized legacy arguments without publishing a provisional tool call", async () => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: "x".repeat(256_001) } }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("Exceeded tool-call argument buffer limit"); + expect(eventTypes).not.toContain("toolcall_start"); + }); + + it("rejects an oversized legacy function name before publishing a provisional call", async () => { + installStream([ + chunk({ function_call: { name: "x".repeat(256_001), arguments: "{}" } }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("Exceeded tool-call argument buffer limit"); + expect(eventTypes).not.toContain("toolcall_start"); + }); + + it("coalesces tiny provisional legacy fragments into one bounded executable delta", async () => { + const query = "x".repeat(1_100); + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"' } }), + ...Array.from(query, (character) => + chunk({ + content: "", + refusal: "", + reasoning_details: [], + function_call: { arguments: character }, + } as ChatCompletionChunk.Choice.Delta), + ), + chunk({ function_call: { arguments: '"}' } }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const argumentDeltas: string[] = []; + for await (const event of stream) { + if (event.type === "toolcall_delta" && event.delta) { + argumentDeltas.push(event.delta); + } + } + const result = await stream.result(); + + expect(result.stopReason).toBe("toolUse"); + expect(result.content[0]).toMatchObject({ + type: "toolCall", + name: "lookup", + arguments: { query }, + }); + expect(argumentDeltas).toEqual([JSON.stringify({ query })]); + }); + + it("bounds visible content buffered behind a provisional legacy call", async () => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"cats"}' } }), + chunk({ content: "x".repeat(256_001) }), + chunk({}, "function_call"), + ]); + + const stream = createStream(model, context, { apiKey: "fixture-token" }); + const eventTypes: string[] = []; + for await (const event of stream) { + eventTypes.push(event.type); + } + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("Exceeded legacy tool-call content buffer limit"); + expect(eventTypes).not.toContain("toolcall_start"); + expect(eventTypes).not.toContain("text_start"); + }); + + it("bounds the number of tiny deltas buffered behind a provisional legacy call", async () => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"cats"}' } }), + ...Array.from({ length: 1_025 }, () => chunk({ content: "x" })), + chunk({}, "function_call"), + ]); + + const result = await createStream(model, context, { apiKey: "fixture-token" }).result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("Exceeded legacy tool-call content buffer limit"); + expect(result.content).toEqual([]); + }); + + it("generates a distinct legacy call id for each assistant response", async () => { + installStream([ + chunk({ function_call: { name: "lookup", arguments: '{"query":"cats"}' } }), + chunk({}, "function_call"), + ]); + + const first = await createStream(model, context, { apiKey: "fixture-token" }).result(); + const second = await createStream(model, context, { apiKey: "fixture-token" }).result(); + + expect(first.content[0]).toMatchObject({ type: "toolCall" }); + expect(second.content[0]).toMatchObject({ type: "toolCall" }); + if (first.content[0]?.type === "toolCall" && second.content[0]?.type === "toolCall") { + expect(first.content[0].id).not.toBe(second.content[0].id); + } + }); +}); diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index eb07a040301c..9157acd2732b 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -58,6 +58,10 @@ import { splitSystemPromptCacheBoundary } from "../utils/system-prompt-cache-bou import { resolveCacheRetention } from "./cache-retention.js"; import { isCloudflareProvider, resolveCloudflareBaseUrl } from "./cloudflare.js"; import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./github-copilot-headers.js"; +import { + createOpenAICompletionsToolCallDeltaNormalizer, + finalizeOpenAICompletionsToolCalls, +} from "./openai-completions-tool-calls.js"; import { clampOpenAIPromptCacheKey } from "./openai-prompt-cache.js"; import { resolveOpenAICompletionsResponseFormat, @@ -203,6 +207,7 @@ export const streamOpenAICompletions: StreamFunction< const toolCallBlocksByIndex = new Map(); const toolCallBlocksById = new Map(); const toolCallBlocksByFirstId = new Map(); + const normalizeToolCallDeltas = createOpenAICompletionsToolCallDeltaNormalizer(); const pendingReasoningDetailsByToolCallId = new Map(); const blocks = output.content as StreamingBlock[]; // A block can be finished mid-stream (native reasoning sealed at the @@ -249,10 +254,6 @@ export const streamOpenAICompletions: StreamFunction< partial: output, }); } else if (block.type === "toolCall") { - // Finalize in-place and strip the scratch buffers so replay only - // carries parsed arguments. - delete block.partialArgs; - delete block.streamIndex; stream.push({ type: "toolcall_end", contentIndex, @@ -433,100 +434,107 @@ export const streamOpenAICompletions: StreamFunction< // Some OpenAI-compatible endpoints deliver a full `message` instead of // `delta` (including refusal-only turns with content: null). Normalize // the same way the managed agent transport does. - const choiceDelta = + const rawChoiceDelta = choice.delta ?? (choice as { message?: ChatCompletionChunk["choices"][number]["delta"] }).message; - if (choiceDelta) { - // Some endpoints return reasoning in reasoning_content (llama.cpp), - // or reasoning (other openai compatible endpoints) - // Use the first non-empty reasoning field to avoid duplication - // (e.g., chutes.ai returns both reasoning_content and reasoning with same content) - const reasoningFields = ["reasoning_content", "reasoning", "reasoning_text"]; - const deltaFields = choiceDelta as Record; - const shouldEmitReasoning = Boolean(model.reasoning && options?.reasoningEffort); - let foundReasoningField: string | null = null; - for (const field of reasoningFields) { - const value = deltaFields[field]; - if (typeof value === "string" && value.length > 0) { - foundReasoningField = field; - break; - } - } - if (foundReasoningField) { - reasoningTagTextPartitioner.markStrict(); - } - if (shouldEmitReasoning && foundReasoningField) { - const delta = deltaFields[foundReasoningField]; - if (typeof delta === "string" && delta.length > 0) { - const thinkingSignature = - model.provider === "opencode-go" && foundReasoningField === "reasoning" - ? "reasoning_content" - : foundReasoningField; - appendThinkingDelta(thinkingSignature, delta); - } - } - if ( - choiceDelta.content !== null && - choiceDelta.content !== undefined && - choiceDelta.content.length > 0 - ) { - appendPartitionedContent(choiceDelta.content, Boolean(foundReasoningField)); - } - - // Chat Completions can put safety/structured-output refusals in a - // top-level `refusal` field with content null. Surface that as - // visible text so the assistant turn is not empty. - const refusalText = typeof choiceDelta.refusal === "string" ? choiceDelta.refusal : ""; - if (refusalText.length > 0) { - appendPartitionedContent(refusalText, Boolean(foundReasoningField)); - } - - if (choiceDelta.tool_calls && choiceDelta.tool_calls.length > 0) { - flushPartitionedContent(); - // The tool-call lane is also a reasoning boundary; seal the thought - // before toolcall_start so thinking_end never trails the action. - sealNativeReasoningBeforeText(); - rememberPendingCommentaryTags( - provisionalCommentaryTags, - tagPendingCommentaryText(output.content), - ); - for (const toolCall of choiceDelta.tool_calls) { - const block = ensureToolCallBlock(toolCall); - if (!block.id && toolCall.id) { - block.id = toolCall.id; - toolCallBlocksById.set(toolCall.id, block); - rememberFirstToolCallById(toolCall.id, block); + if (rawChoiceDelta) { + for (const normalizedDelta of normalizeToolCallDeltas( + rawChoiceDelta, + choice.finish_reason, + )) { + const choiceDelta = normalizedDelta.delta; + // Some endpoints return reasoning in reasoning_content (llama.cpp), + // or reasoning (other openai compatible endpoints) + // Use the first non-empty reasoning field to avoid duplication + // (e.g., chutes.ai returns both reasoning_content and reasoning with same content) + const reasoningFields = ["reasoning_content", "reasoning", "reasoning_text"]; + const deltaFields = choiceDelta as Record; + const shouldEmitReasoning = Boolean(model.reasoning && options?.reasoningEffort); + let foundReasoningField: string | null = null; + for (const field of reasoningFields) { + const value = deltaFields[field]; + if (typeof value === "string" && value.length > 0) { + foundReasoningField = field; + break; } - if (!block.name && toolCall.function?.name) { - block.name = toolCall.function.name; - } - - let delta = ""; - if (toolCall.function?.arguments) { - delta = toolCall.function.arguments; - block.partialArgs = (block.partialArgs ?? "") + toolCall.function.arguments; - block.arguments = parseStreamingJson(block.partialArgs); - } - stream.push({ - type: "toolcall_delta", - contentIndex: getContentIndex(block), - delta, - partial: output, - }); } - } + if (foundReasoningField) { + reasoningTagTextPartitioner.markStrict(); + } + if (shouldEmitReasoning && foundReasoningField) { + const delta = deltaFields[foundReasoningField]; + if (typeof delta === "string" && delta.length > 0) { + const thinkingSignature = + model.provider === "opencode-go" && foundReasoningField === "reasoning" + ? "reasoning_content" + : foundReasoningField; + appendThinkingDelta(thinkingSignature, delta); + } + } + if ( + choiceDelta.content !== null && + choiceDelta.content !== undefined && + choiceDelta.content.length > 0 + ) { + appendPartitionedContent(choiceDelta.content, Boolean(foundReasoningField)); + } - const reasoningDetails = (choiceDelta as { reasoning_details?: unknown }) - .reasoning_details; - if (Array.isArray(reasoningDetails)) { - for (const detail of reasoningDetails) { - if (isEncryptedReasoningDetail(detail)) { - const serializedDetail = JSON.stringify(detail); - const matchingToolCall = toolCallBlocksByFirstId.get(detail.id); - if (matchingToolCall) { - matchingToolCall.thoughtSignature = serializedDetail; - } else { - pendingReasoningDetailsByToolCallId.set(detail.id, serializedDetail); + // Chat Completions can put safety/structured-output refusals in a + // top-level `refusal` field with content null. Surface that as + // visible text so the assistant turn is not empty. + const refusalText = typeof choiceDelta.refusal === "string" ? choiceDelta.refusal : ""; + if (refusalText.length > 0) { + appendPartitionedContent(refusalText, Boolean(foundReasoningField)); + } + + const toolCallDeltas = normalizedDelta.toolCalls; + if (toolCallDeltas.length > 0) { + flushPartitionedContent(); + // The tool-call lane is also a reasoning boundary; seal the thought + // before toolcall_start so thinking_end never trails the action. + sealNativeReasoningBeforeText(); + rememberPendingCommentaryTags( + provisionalCommentaryTags, + tagPendingCommentaryText(output.content), + ); + for (const toolCall of toolCallDeltas) { + const block = ensureToolCallBlock(toolCall); + if (!block.id && toolCall.id) { + block.id = toolCall.id; + toolCallBlocksById.set(toolCall.id, block); + rememberFirstToolCallById(toolCall.id, block); + } + if (!block.name && toolCall.function?.name) { + block.name = toolCall.function.name; + } + + let delta = ""; + if (toolCall.function?.arguments) { + delta = toolCall.function.arguments; + block.partialArgs = (block.partialArgs ?? "") + toolCall.function.arguments; + block.arguments = parseStreamingJson(block.partialArgs); + } + stream.push({ + type: "toolcall_delta", + contentIndex: getContentIndex(block), + delta, + partial: output, + }); + } + } + + const reasoningDetails = (choiceDelta as { reasoning_details?: unknown }) + .reasoning_details; + if (Array.isArray(reasoningDetails)) { + for (const detail of reasoningDetails) { + if (isEncryptedReasoningDetail(detail)) { + const serializedDetail = JSON.stringify(detail); + const matchingToolCall = toolCallBlocksByFirstId.get(detail.id); + if (matchingToolCall) { + matchingToolCall.thoughtSignature = serializedDetail; + } else { + pendingReasoningDetailsByToolCallId.set(detail.id, serializedDetail); + } } } } @@ -536,35 +544,46 @@ export const streamOpenAICompletions: StreamFunction< flushPartitionedContent(); - for (const block of blocks) { - finishBlock(block); - } + let terminalError: Error | undefined; if (options?.signal?.aborted) { - throw transportAbortError(options.signal); + terminalError = transportAbortError(options.signal); + } else if (output.stopReason === "aborted") { + terminalError = new Error("Request was aborted"); + } else if (output.stopReason === "error") { + terminalError = new Error(output.errorMessage || "Provider returned an error stop reason"); + } else if (!hasFinishReason) { + terminalError = new Error("Stream ended without finish_reason"); } - if (output.stopReason === "aborted") { - throw new Error("Request was aborted"); - } - if (output.stopReason === "error") { - throw new Error(output.errorMessage || "Provider returned an error stop reason"); - } - if (!hasFinishReason) { - throw new Error("Stream ended without finish_reason"); + if (terminalError) { + for (const block of blocks) { + if (block.type !== "toolCall") { + finishBlock(block); + } + } + throw terminalError; } - const hasToolCalls = output.content.some((block) => block.type === "toolCall"); - const hasVisibleText = output.content.some( - (block) => block.type === "text" && block.text.trim().length > 0, - ); - if (output.stopReason === "toolUse" && !hasToolCalls) { - output.stopReason = "stop"; + finalizeOpenAICompletionsToolCalls(output); + if (output.stopReason === "aborted" || output.stopReason === "error") { + for (const block of blocks) { + if (block.type !== "toolCall") { + finishBlock(block); + } + } + throw new Error( + output.errorMessage || + (output.stopReason === "aborted" + ? "Request was aborted" + : "Provider returned an invalid tool call"), + ); } - if (output.stopReason === "stop" && hasToolCalls && !hasVisibleText) { - output.stopReason = "toolUse"; - } - if (hasToolCalls && output.stopReason !== "toolUse") { - output.content = output.content.filter((block) => block.type !== "toolCall"); + // Tool completion is irreversible: confirm the terminal before closing + // blocks, then preserve their original text/thinking/tool event order. + for (const block of blocks) { + if (block.type !== "toolCall" || output.stopReason === "toolUse") { + finishBlock(block); + } } if (output.stopReason !== "toolUse") { clearPendingCommentaryText(provisionalCommentaryTags); @@ -576,13 +595,14 @@ export const streamOpenAICompletions: StreamFunction< stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { + output.stopReason = options?.signal?.aborted ? "aborted" : "error"; + finalizeOpenAICompletionsToolCalls(output, { allowSilentToolCallPromotion: false }); for (const block of output.content) { delete (block as { index?: number }).index; // Streaming scratch buffers are only used during parsing; never persist them. delete (block as { partialArgs?: string }).partialArgs; delete (block as { streamIndex?: number }).streamIndex; } - output.stopReason = options?.signal?.aborted ? "aborted" : "error"; output.errorMessage = formatProviderError(error); // Some providers via OpenRouter give additional information in this field. const rawMetadata = (error as { error?: { metadata?: { raw?: string } } })?.error?.metadata diff --git a/packages/ai/src/transports/openai-completions-transport.ts b/packages/ai/src/transports/openai-completions-transport.ts index f457fe4bc003..70edd3dcde8b 100644 --- a/packages/ai/src/transports/openai-completions-transport.ts +++ b/packages/ai/src/transports/openai-completions-transport.ts @@ -9,6 +9,10 @@ import { applyProviderReportedUsageCost, calculateCost } from "../model-utils.js import { convertMessages } from "../openai-completions-messages.js"; import type { OpenAICompletionsOptions } from "../provider-options.js"; import { resolveCacheRetention } from "../providers/cache-retention.js"; +import { + createOpenAICompletionsToolCallDeltaNormalizer, + finalizeOpenAICompletionsToolCalls, +} from "../providers/openai-completions-tool-calls.js"; import { isOpenAIGpt54MiniModel, isOpenAIGpt55Model, @@ -353,7 +357,16 @@ export function createOpenAICompletionsTransportStreamFn(): StreamFn { }); finalizeTransportStream({ stream, output, signal: options?.signal }); } catch (error) { - failTransportStream({ stream, output, signal: options?.signal, error }); + failTransportStream({ + stream, + output, + signal: options?.signal, + error, + cleanup: () => { + output.stopReason = options?.signal?.aborted ? "aborted" : "error"; + finalizeOpenAICompletionsToolCalls(output, { allowSilentToolCallPromotion: false }); + }, + }); } finally { firstEventAbort?.dispose(); } @@ -408,6 +421,7 @@ async function processOpenAICompletionsStream( const provisionalCommentaryTags: PendingCommentaryTags = new Map(); const toolCallBlockBytes = new WeakMap(); const toolCallBlockIndices = new WeakMap(); + const normalizeToolCallDeltas = createOpenAICompletionsToolCallDeltaNormalizer(); let sawStopFinishReason = false; let sawNativeToolCallDelta = false; const blockIndex = () => output.content.length - 1; @@ -669,132 +683,137 @@ async function processOpenAICompletionsStream( output.errorMessage = finishReasonResult.errorMessage; } } - const choiceDelta = + const rawChoiceDelta = choice.delta ?? (choice as unknown as { message?: ChatCompletionChunk["choices"][number]["delta"] }).message; - if (!choiceDelta) { + if (!rawChoiceDelta) { emitReasoningUsageActivity(hasReasoningUsageActivity); await cooperativeScheduler.afterEvent(); continue; } - const reasoningDeltas = getCompletionsReasoningDeltas( - choiceDelta as Record, - compat.visibleReasoningDetailTypes, - ); - const hasMirroredReasoning = reasoningDeltas.some((delta) => delta.kind === "thinking"); - if (hasMirroredReasoning) { - reasoningTagTextPartitioner.markStrict(); - } - if (choiceDelta.content) { - // Structured content can contain visible text and thinking blocks in the - // same delta, so route each extracted block through the normal stream path. - const contentDeltas = getCompletionsContentDeltas(choiceDelta.content); - for (const contentDelta of contentDeltas) { - if (contentDelta.kind === "text") { - const routedDeltas = hasMirroredReasoning - ? reasoningTagTextPartitioner.push(contentDelta.text) - : reasoningTagTextPartitioner.pushVisible(contentDelta.text); - for (const routedDelta of routedDeltas) { - appendPartitionedVisibleDelta(routedDelta); - } - } else { - reasoningTagTextPartitioner.markStrict(); - appendRoutedContentDelta(contentDelta); - } - } - } - // Chat Completions can put safety/structured-output refusals in a top-level - // `refusal` field with content null. Surface that as visible text so the - // assistant turn is not empty (Responses path already routes refusal deltas). - const refusalText = typeof choiceDelta.refusal === "string" ? choiceDelta.refusal : ""; - if (refusalText) { - const routedDeltas = hasMirroredReasoning - ? reasoningTagTextPartitioner.push(refusalText) - : reasoningTagTextPartitioner.pushVisible(refusalText); - for (const routedDelta of routedDeltas) { - appendPartitionedVisibleDelta(routedDelta); - } - } - for (const reasoningDelta of reasoningDeltas) { - if (reasoningDelta.kind === "thinking" && !emitReasoning) { - continue; - } - if (currentBlock?.type === "toolCall") { - queuePostToolCallDelta({ ...reasoningDelta }); - continue; - } - if (reasoningDelta.kind === "text") { - appendTextDelta(reasoningDelta.text); - } else if (emitReasoning) { - appendThinkingDelta(reasoningDelta); - } - } - if (choiceDelta.tool_calls && choiceDelta.tool_calls.length > 0) { - sawNativeToolCallDelta = true; - flushReasoningTagTextPartitionerAtEnd(); - rememberPendingCommentaryTags( - provisionalCommentaryTags, - tagPendingCommentaryText(output.content), + for (const normalizedDelta of normalizeToolCallDeltas(rawChoiceDelta, choice.finish_reason)) { + const choiceDelta = normalizedDelta.delta; + const reasoningDeltas = getCompletionsReasoningDeltas( + choiceDelta as Record, + compat.visibleReasoningDetailTypes, ); - for (const toolCall of choiceDelta.tool_calls) { - const streamIndex = typeof toolCall.index === "number" ? toolCall.index : undefined; - let block = streamIndex !== undefined ? toolCallBlocksByIndex.get(streamIndex) : undefined; - if (!block && toolCall.id) { - block = toolCallBlocksById.get(toolCall.id); - } - if (!block) { - const switchingToolCall = currentBlock?.type === "toolCall"; - if (switchingToolCall) { - currentBlock = null; - flushPendingPostToolCallDeltas(); + const hasMirroredReasoning = reasoningDeltas.some((delta) => delta.kind === "thinking"); + if (hasMirroredReasoning) { + reasoningTagTextPartitioner.markStrict(); + } + if (choiceDelta.content) { + // Structured content can contain visible text and thinking blocks in the + // same delta, so route each extracted block through the normal stream path. + const contentDeltas = getCompletionsContentDeltas(choiceDelta.content); + for (const contentDelta of contentDeltas) { + if (contentDelta.kind === "text") { + const routedDeltas = hasMirroredReasoning + ? reasoningTagTextPartitioner.push(contentDelta.text) + : reasoningTagTextPartitioner.pushVisible(contentDelta.text); + for (const routedDelta of routedDeltas) { + appendPartitionedVisibleDelta(routedDelta); + } + } else { + reasoningTagTextPartitioner.markStrict(); + appendRoutedContentDelta(contentDelta); } - const initialSig = extractGoogleThoughtSignature(toolCall); - block = { - type: "toolCall", - id: toolCall.id || "", - name: toolCall.function?.name || "", - arguments: {}, - partialArgs: "", - ...(initialSig ? { thoughtSignature: initialSig } : {}), - }; - output.content.push(block); - toolCallBlockIndices.set(block, output.content.length - 1); - pushStreamEvent({ - type: "toolcall_start", - contentIndex: toolCallBlockIndices.get(block) ?? -1, - partial: output, - }); } - if (streamIndex !== undefined && !toolCallBlocksByIndex.has(streamIndex)) { - toolCallBlocksByIndex.set(streamIndex, block); + } + // Chat Completions can put safety/structured-output refusals in a top-level + // `refusal` field with content null. Surface that as visible text so the + // assistant turn is not empty (Responses path already routes refusal deltas). + const refusalText = typeof choiceDelta.refusal === "string" ? choiceDelta.refusal : ""; + if (refusalText) { + const routedDeltas = hasMirroredReasoning + ? reasoningTagTextPartitioner.push(refusalText) + : reasoningTagTextPartitioner.pushVisible(refusalText); + for (const routedDelta of routedDeltas) { + appendPartitionedVisibleDelta(routedDelta); } - if (toolCall.id) { - block.id = toolCall.id; - toolCallBlocksById.set(toolCall.id, block); + } + for (const reasoningDelta of reasoningDeltas) { + if (reasoningDelta.kind === "thinking" && !emitReasoning) { + continue; } - currentBlock = block; - if (toolCall.function?.name) { - block.name = toolCall.function.name; + if (currentBlock?.type === "toolCall") { + queuePostToolCallDelta({ ...reasoningDelta }); + continue; } - const deltaSig = extractGoogleThoughtSignature(toolCall); - if (deltaSig) { - block.thoughtSignature = deltaSig; + if (reasoningDelta.kind === "text") { + appendTextDelta(reasoningDelta.text); + } else if (emitReasoning) { + appendThinkingDelta(reasoningDelta); } - if (toolCall.function?.arguments) { - const nextArgumentBytes = measureUtf8Bytes(toolCall.function.arguments); - const currentBlockArgBytes = toolCallBlockBytes.get(block) ?? 0; - if (currentBlockArgBytes + nextArgumentBytes > MAX_TOOL_CALL_ARGUMENT_BUFFER_BYTES) { - throw new Error("Exceeded tool-call argument buffer limit"); + } + const toolCallDeltas = normalizedDelta.toolCalls; + if (toolCallDeltas.length > 0) { + sawNativeToolCallDelta = true; + flushReasoningTagTextPartitionerAtEnd(); + rememberPendingCommentaryTags( + provisionalCommentaryTags, + tagPendingCommentaryText(output.content), + ); + for (const toolCall of toolCallDeltas) { + const streamIndex = typeof toolCall.index === "number" ? toolCall.index : undefined; + let block = + streamIndex !== undefined ? toolCallBlocksByIndex.get(streamIndex) : undefined; + if (!block && toolCall.id) { + block = toolCallBlocksById.get(toolCall.id); + } + if (!block) { + const switchingToolCall = currentBlock?.type === "toolCall"; + if (switchingToolCall) { + currentBlock = null; + flushPendingPostToolCallDeltas(); + } + const initialSig = extractGoogleThoughtSignature(toolCall); + block = { + type: "toolCall", + id: toolCall.id || "", + name: toolCall.function?.name || "", + arguments: {}, + partialArgs: "", + ...(initialSig ? { thoughtSignature: initialSig } : {}), + }; + output.content.push(block); + toolCallBlockIndices.set(block, output.content.length - 1); + pushStreamEvent({ + type: "toolcall_start", + contentIndex: toolCallBlockIndices.get(block) ?? -1, + partial: output, + }); + } + if (streamIndex !== undefined && !toolCallBlocksByIndex.has(streamIndex)) { + toolCallBlocksByIndex.set(streamIndex, block); + } + if (toolCall.id) { + block.id = toolCall.id; + toolCallBlocksById.set(toolCall.id, block); + } + currentBlock = block; + if (toolCall.function?.name) { + block.name = toolCall.function.name; + } + const deltaSig = extractGoogleThoughtSignature(toolCall); + if (deltaSig) { + block.thoughtSignature = deltaSig; + } + if (toolCall.function?.arguments) { + const nextArgumentBytes = measureUtf8Bytes(toolCall.function.arguments); + const currentBlockArgBytes = toolCallBlockBytes.get(block) ?? 0; + if (currentBlockArgBytes + nextArgumentBytes > MAX_TOOL_CALL_ARGUMENT_BUFFER_BYTES) { + throw new Error("Exceeded tool-call argument buffer limit"); + } + toolCallBlockBytes.set(block, currentBlockArgBytes + nextArgumentBytes); + block.partialArgs += toolCall.function.arguments; + block.arguments = parseStreamingJson(block.partialArgs); + pushStreamEvent({ + type: "toolcall_delta", + contentIndex: toolCallBlockIndices.get(block) ?? -1, + delta: toolCall.function.arguments, + partial: output, + }); } - toolCallBlockBytes.set(block, currentBlockArgBytes + nextArgumentBytes); - block.partialArgs += toolCall.function.arguments; - block.arguments = parseStreamingJson(block.partialArgs); - pushStreamEvent({ - type: "toolcall_delta", - contentIndex: toolCallBlockIndices.get(block) ?? -1, - delta: toolCall.function.arguments, - partial: output, - }); } } } @@ -807,14 +826,6 @@ async function processOpenAICompletionsStream( flushDeepSeekTextFilterAtEnd(); currentBlock = null; flushPendingPostToolCallDeltas(); - const hasToolCalls = output.content.some((block) => block.type === "toolCall"); - const hasVisibleText = output.content.some( - (block) => - block.type === "text" && typeof block.text === "string" && block.text.trim().length > 0, - ); - if (output.stopReason === "toolUse" && !hasToolCalls) { - output.stopReason = "stop"; - } // Promote complete silent tool-call-only responses when the stream finished // cleanly (reached post-loop). Two paths: // sawStopFinishReason: explicit provider terminal (legacy DSML / #88791) @@ -823,17 +834,18 @@ async function processOpenAICompletionsStream( // DeepSeek V4). [DONE] tracking distinguishes clean termination from // connection drops (EOF without [DONE] remains fail-closed). // Truncated streams throw before reaching this code. - if ( - output.stopReason === "stop" && - hasToolCalls && - !hasVisibleText && - (sawStopFinishReason || (sawNativeToolCallDelta && (options?.sawStreamDONE?.() ?? false))) - ) { - output.stopReason = "toolUse"; - } - if (hasToolCalls && output.stopReason !== "toolUse") { - output.content = output.content.filter((block) => block.type !== "toolCall"); - } + finalizeOpenAICompletionsToolCalls(output, { + allowSilentToolCallPromotion: + sawStopFinishReason || (sawNativeToolCallDelta && (options?.sawStreamDONE?.() ?? false)), + onConfirmedToolCall(block, contentIndex) { + pushStreamEvent({ + type: "toolcall_end", + contentIndex, + toolCall: block, + partial: output, + }); + }, + }); if (output.stopReason !== "toolUse") { clearPendingCommentaryText(provisionalCommentaryTags); } diff --git a/src/agents/openai-transport-stream.deepseek-and-shaping.test.ts b/src/agents/openai-transport-stream.deepseek-and-shaping.test.ts index 7a7c34a72d4b..40fc8273eb6d 100644 --- a/src/agents/openai-transport-stream.deepseek-and-shaping.test.ts +++ b/src/agents/openai-transport-stream.deepseek-and-shaping.test.ts @@ -101,7 +101,6 @@ describe("openai transport stream", () => { id: "call_native_1", name: "read", arguments: { path: "/tmp/native.md" }, - partialArgs: '{"path":"/tmp/native.md"}', }, ]); expect(JSON.stringify(events)).not.toContain("DSML"); @@ -144,7 +143,6 @@ describe("openai transport stream", () => { id: "call_native_1", name: "read", arguments: { path: "/tmp/native.md" }, - partialArgs: '{"path":"/tmp/native.md"}', }, ]); }); @@ -184,7 +182,6 @@ describe("openai transport stream", () => { id: "call_native_1", name: "read", arguments: { path: "/tmp/native.md" }, - partialArgs: '{"path":"/tmp/native.md"}', }, { type: "text", @@ -236,7 +233,6 @@ describe("openai transport stream", () => { id: "call_native_1", name: "read", arguments: { path: "/tmp/native.md" }, - partialArgs: '{"path":"/tmp/native.md"}', }, { type: "text", @@ -274,7 +270,6 @@ describe("openai transport stream", () => { id: expect.stringMatching(/^call_[0-9a-f]{24}$/), name: "session_status", arguments: { sessionKey: "current" }, - partialArgs: '{"sessionKey":"current"}', }, ]); expect(JSON.stringify(events)).not.toContain("DSML"); @@ -436,7 +431,6 @@ describe("openai transport stream", () => { id: expect.stringMatching(/^call_[0-9a-f]{24}$/), name: "session_status", arguments: { sessionKey: "current" }, - partialArgs: '{"sessionKey":"current"}', }, ]); } @@ -465,7 +459,6 @@ describe("openai transport stream", () => { id: expect.stringMatching(/^call_[0-9a-f]{24}$/), name: "read", arguments: { path: "/tmp/native.md" }, - partialArgs: '{"path":"/tmp/native.md"}', }, ]); });