fix(ai): handle mixed chat completion deltas

Fixes #4228
2026-05-07 00:12:19 +02:00
parent ace3fd7e30
commit 6b271842e2
2 changed files with 367 additions and 121 deletions
--- a/packages/ai/src/providers/openai-completions.ts
+++ b/packages/ai/src/providers/openai-completions.ts
@@ -160,45 +160,104 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
 				partialArgs?: string;
 				streamIndex?: number;
 			}
+			type StreamingBlock = TextContent | ThinkingContent | StreamingToolCallBlock;
+			type StreamingToolCallDelta = NonNullable<ChatCompletionChunk.Choice.Delta["tool_calls"]>[number];

-			let currentBlock: TextContent | ThinkingContent | StreamingToolCallBlock | null = null;
-			const blocks = output.content;
-			const getContentIndex = (block: typeof currentBlock) => (block ? blocks.indexOf(block) : -1);
-			const currentContentIndex = () => getContentIndex(currentBlock);
-			const finishCurrentBlock = (block?: typeof currentBlock) => {
-				if (block) {
-					const contentIndex = getContentIndex(block);
-					if (contentIndex === -1) {
-						return;
-					}
-					if (block.type === "text") {
-						stream.push({
-							type: "text_end",
-							contentIndex,
-							content: block.text,
-							partial: output,
-						});
-					} else if (block.type === "thinking") {
-						stream.push({
-							type: "thinking_end",
-							contentIndex,
-							content: block.thinking,
-							partial: output,
-						});
-					} else if (block.type === "toolCall") {
-						block.arguments = parseStreamingJson(block.partialArgs);
-						// 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,
-							toolCall: block,
-							partial: output,
-						});
-					}
+			let textBlock: TextContent | null = null;
+			let thinkingBlock: ThinkingContent | null = null;
+			const toolCallBlocksByIndex = new Map<number, StreamingToolCallBlock>();
+			const toolCallBlocksById = new Map<string, StreamingToolCallBlock>();
+			const blocks = output.content as StreamingBlock[];
+			const getContentIndex = (block: StreamingBlock) => blocks.indexOf(block);
+			const finishBlock = (block: StreamingBlock) => {
+				const contentIndex = getContentIndex(block);
+				if (contentIndex === -1) {
+					return;
 				}
+				if (block.type === "text") {
+					stream.push({
+						type: "text_end",
+						contentIndex,
+						content: block.text,
+						partial: output,
+					});
+				} else if (block.type === "thinking") {
+					stream.push({
+						type: "thinking_end",
+						contentIndex,
+						content: block.thinking,
+						partial: output,
+					});
+				} else if (block.type === "toolCall") {
+					block.arguments = parseStreamingJson(block.partialArgs);
+					// 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,
+						toolCall: block,
+						partial: output,
+					});
+				}
+			};
+			const ensureTextBlock = () => {
+				if (!textBlock) {
+					textBlock = { type: "text", text: "" };
+					blocks.push(textBlock);
+					stream.push({ type: "text_start", contentIndex: getContentIndex(textBlock), partial: output });
+				}
+				return textBlock;
+			};
+			const ensureThinkingBlock = (thinkingSignature: string) => {
+				if (!thinkingBlock) {
+					thinkingBlock = {
+						type: "thinking",
+						thinking: "",
+						thinkingSignature,
+					};
+					blocks.push(thinkingBlock);
+					stream.push({ type: "thinking_start", contentIndex: getContentIndex(thinkingBlock), partial: output });
+				}
+				return thinkingBlock;
+			};
+			const ensureToolCallBlock = (toolCall: StreamingToolCallDelta) => {
+				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) {
+					block = {
+						type: "toolCall",
+						id: toolCall.id || "",
+						name: toolCall.function?.name || "",
+						arguments: {},
+						partialArgs: "",
+						streamIndex,
+					};
+					if (streamIndex !== undefined) {
+						toolCallBlocksByIndex.set(streamIndex, block);
+					}
+					if (toolCall.id) {
+						toolCallBlocksById.set(toolCall.id, block);
+					}
+					blocks.push(block);
+					stream.push({
+						type: "toolcall_start",
+						contentIndex: getContentIndex(block),
+						partial: output,
+					});
+				}
+				if (streamIndex !== undefined && block.streamIndex === undefined) {
+					block.streamIndex = streamIndex;
+					toolCallBlocksByIndex.set(streamIndex, block);
+				}
+				if (toolCall.id) {
+					toolCallBlocksById.set(toolCall.id, block);
+				}
+				return block;
 			};

 			for await (const chunk of openaiStream) {
@@ -237,22 +296,14 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
 						choice.delta.content !== undefined &&
 						choice.delta.content.length > 0
 					) {
-						if (!currentBlock || currentBlock.type !== "text") {
-							finishCurrentBlock(currentBlock);
-							currentBlock = { type: "text", text: "" };
-							output.content.push(currentBlock);
-							stream.push({ type: "text_start", contentIndex: currentContentIndex(), partial: output });
-						}
-
-						if (currentBlock.type === "text") {
-							currentBlock.text += choice.delta.content;
-							stream.push({
-								type: "text_delta",
-								contentIndex: currentContentIndex(),
-								delta: choice.delta.content,
-								partial: output,
-							});
-						}
+						const block = ensureTextBlock();
+						block.text += choice.delta.content;
+						stream.push({
+							type: "text_delta",
+							contentIndex: getContentIndex(block),
+							delta: choice.delta.content,
+							partial: output,
+						});
 					}

 					// Some endpoints return reasoning in reasoning_content (llama.cpp),
@@ -260,38 +311,24 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
 					// 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 = choice.delta as Record<string, unknown>;
 					let foundReasoningField: string | null = null;
 					for (const field of reasoningFields) {
-						if (
-							(choice.delta as any)[field] !== null &&
-							(choice.delta as any)[field] !== undefined &&
-							(choice.delta as any)[field].length > 0
-						) {
-							if (!foundReasoningField) {
-								foundReasoningField = field;
-								break;
-							}
+						const value = deltaFields[field];
+						if (typeof value === "string" && value.length > 0) {
+							foundReasoningField = field;
+							break;
 						}
 					}

 					if (foundReasoningField) {
-						if (!currentBlock || currentBlock.type !== "thinking") {
-							finishCurrentBlock(currentBlock);
-							currentBlock = {
-								type: "thinking",
-								thinking: "",
-								thinkingSignature: foundReasoningField,
-							};
-							output.content.push(currentBlock);
-							stream.push({ type: "thinking_start", contentIndex: currentContentIndex(), partial: output });
-						}
-
-						if (currentBlock.type === "thinking") {
-							const delta = (choice.delta as any)[foundReasoningField];
-							currentBlock.thinking += delta;
+						const delta = deltaFields[foundReasoningField];
+						if (typeof delta === "string" && delta.length > 0) {
+							const block = ensureThinkingBlock(foundReasoningField);
+							block.thinking += delta;
 							stream.push({
 								type: "thinking_delta",
-								contentIndex: currentContentIndex(),
+								contentIndex: getContentIndex(block),
 								delta,
 								partial: output,
 							});
@@ -300,52 +337,27 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA

 					if (choice?.delta?.tool_calls) {
 						for (const toolCall of choice.delta.tool_calls) {
-							const streamIndex = typeof toolCall.index === "number" ? toolCall.index : undefined;
-							const sameToolCall =
-								currentBlock?.type === "toolCall" &&
-								((streamIndex !== undefined && currentBlock.streamIndex === streamIndex) ||
-									(streamIndex === undefined && toolCall.id && currentBlock.id === toolCall.id));
-
-							if (!sameToolCall) {
-								finishCurrentBlock(currentBlock);
-								currentBlock = {
-									type: "toolCall",
-									id: toolCall.id || "",
-									name: toolCall.function?.name || "",
-									arguments: {},
-									partialArgs: "",
-									streamIndex,
-								};
-								output.content.push(currentBlock);
-								stream.push({
-									type: "toolcall_start",
-									contentIndex: getContentIndex(currentBlock),
-									partial: output,
-								});
+							const block = ensureToolCallBlock(toolCall);
+							if (!block.id && toolCall.id) {
+								block.id = toolCall.id;
+								toolCallBlocksById.set(toolCall.id, block);
+							}
+							if (!block.name && toolCall.function?.name) {
+								block.name = toolCall.function.name;
 							}

-							const currentToolCallBlock = currentBlock?.type === "toolCall" ? currentBlock : null;
-							if (currentToolCallBlock) {
-								if (!currentToolCallBlock.id && toolCall.id) currentToolCallBlock.id = toolCall.id;
-								if (!currentToolCallBlock.name && toolCall.function?.name) {
-									currentToolCallBlock.name = toolCall.function.name;
-								}
-								if (currentToolCallBlock.streamIndex === undefined && streamIndex !== undefined) {
-									currentToolCallBlock.streamIndex = streamIndex;
-								}
-								let delta = "";
-								if (toolCall.function?.arguments) {
-									delta = toolCall.function.arguments;
-									currentToolCallBlock.partialArgs += toolCall.function.arguments;
-									currentToolCallBlock.arguments = parseStreamingJson(currentToolCallBlock.partialArgs);
-								}
-								stream.push({
-									type: "toolcall_delta",
-									contentIndex: getContentIndex(currentToolCallBlock),
-									delta,
-									partial: output,
-								});
+							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,
+							});
 						}
 					}

@@ -365,7 +377,9 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
 				}
 			}

-			finishCurrentBlock(currentBlock);
+			for (const block of blocks) {
+				finishBlock(block);
+			}
 			if (options?.signal?.aborted) {
 				throw new Error("Request was aborted");
 			}