From 699adfb386d7f7a2d58d3fe11800e91deccdf26e Mon Sep 17 00:00:00 2001 From: roboomp Date: Sat, 6 Jun 2026 04:30:05 +0000 Subject: [PATCH] fix(ai): preserved llama parallel tool args Route identifierless Responses function_call_arguments.done events through open function calls in item order so local llama.cpp streams keep each parallel tool's final arguments. Fixes #1970 --- packages/ai/CHANGELOG.md | 4 ++ .../src/providers/openai-responses-shared.ts | 40 +++++++++++-- ...enai-responses-parallel-tool-calls.test.ts | 58 +++++++++++++++++++ 3 files changed, 96 insertions(+), 6 deletions(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 83afdf41e..f779c4534 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed llama.cpp/OpenAI Responses parallel tool calls losing arguments when `function_call_arguments.done` events omit `output_index` and `item_id`, by routing those identifierless final-argument events through the open function calls in item order. ([#1970](https://github.com/can1357/oh-my-pi/issues/1970)) + ## [15.9.2] - 2026-06-05 ### Added diff --git a/packages/ai/src/providers/openai-responses-shared.ts b/packages/ai/src/providers/openai-responses-shared.ts index 135f92589..2b0f5f2b3 100644 --- a/packages/ai/src/providers/openai-responses-shared.ts +++ b/packages/ai/src/providers/openai-responses-shared.ts @@ -395,7 +395,7 @@ export async function processResponsesStream( model: Model, options?: ProcessResponsesStreamOptions, ): Promise { - type StreamingToolCallBlock = ToolCall & { partialJson: string; lastParseLen?: number }; + type StreamingToolCallBlock = ToolCall & { partialJson: string; lastParseLen?: number; argumentsDone?: boolean }; interface StreamingItem { item: ResponseReasoningItem | ResponseOutputMessage | ResponseFunctionToolCall | ResponseCustomToolCall; block: ThinkingContent | TextContent | StreamingToolCallBlock; @@ -409,6 +409,7 @@ export async function processResponsesStream( const openItemsByOutputIndex = new Map(); const openItemsByItemId = new Map(); let lastOpenItem: StreamingItem | null = null; + const openItemsInOrder: StreamingItem[] = []; const registerOpenItem = ( outputIndex: number | undefined, @@ -417,6 +418,7 @@ export async function processResponsesStream( ): void => { if (typeof outputIndex === "number") openItemsByOutputIndex.set(outputIndex, entry); if (itemId) openItemsByItemId.set(itemId, entry); + openItemsInOrder.push(entry); lastOpenItem = entry; }; const lookupOpenItem = (event: { output_index?: number; item_id?: string }): StreamingItem | undefined => { @@ -431,6 +433,24 @@ export async function processResponsesStream( // Fallback for tests / mock providers that omit identifiers on stream events. return lastOpenItem ?? undefined; }; + const hasOpenItemKey = (event: { output_index?: number; item_id?: string }): boolean => + typeof event.output_index === "number" || event.item_id !== undefined; + const lookupOpenFunctionCallItem = (event: { + output_index?: number; + item_id?: string; + }): StreamingItem | undefined => { + if (hasOpenItemKey(event)) return lookupOpenItem(event); + for (const candidate of openItemsInOrder) { + if ( + candidate.item.type === "function_call" && + candidate.block.type === "toolCall" && + !candidate.block.argumentsDone + ) { + return candidate; + } + } + return lastOpenItem?.item.type === "function_call" ? lastOpenItem : undefined; + }; const closeOpenItem = ( outputIndex: number | undefined, itemId: string | undefined, @@ -438,6 +458,10 @@ export async function processResponsesStream( ): void => { if (typeof outputIndex === "number") openItemsByOutputIndex.delete(outputIndex); if (itemId) openItemsByItemId.delete(itemId); + if (entry) { + const index = openItemsInOrder.indexOf(entry); + if (index >= 0) openItemsInOrder.splice(index, 1); + } if (entry && lastOpenItem === entry) lastOpenItem = null; }; const contentIndexOf = (block: ThinkingContent | TextContent | StreamingToolCallBlock): number => @@ -584,7 +608,7 @@ export async function processResponsesStream( } } } else if (event.type === "response.function_call_arguments.delta") { - const entry = lookupOpenItem(event); + const entry = lookupOpenFunctionCallItem(event); if (entry?.item.type === "function_call" && entry.block.type === "toolCall") { const block = entry.block; block.partialJson += event.delta; @@ -601,11 +625,12 @@ export async function processResponsesStream( }); } } else if (event.type === "response.function_call_arguments.done") { - const entry = lookupOpenItem(event); + const entry = lookupOpenFunctionCallItem(event); if (entry?.item.type === "function_call" && entry.block.type === "toolCall") { const block = entry.block; block.partialJson = event.arguments; block.arguments = parseStreamingJson(block.partialJson); + block.argumentsDone = true; delete (block as { partialJson?: string }).partialJson; delete (block as { lastParseLen?: number }).lastParseLen; } @@ -668,9 +693,11 @@ export async function processResponsesStream( closeOpenItem(event.output_index, item.id, entry); } else if (item.type === "function_call") { const block = entry?.block.type === "toolCall" ? entry.block : undefined; - const args = block?.partialJson - ? parseStreamingJson(block.partialJson) - : parseStreamingJson(item.arguments || "{}"); + const args = block?.argumentsDone + ? block.arguments + : block?.partialJson + ? parseStreamingJson(block.partialJson) + : parseStreamingJson(item.arguments || "{}"); const toolCall: ToolCall = { type: "toolCall", id: encodeResponsesToolCallId(item.call_id, item.id), @@ -685,6 +712,7 @@ export async function processResponsesStream( block.arguments = args; delete (block as { partialJson?: string }).partialJson; delete (block as { lastParseLen?: number }).lastParseLen; + delete (block as { argumentsDone?: boolean }).argumentsDone; } const contentIndex = block ? contentIndexOf(block) : output.content.length - 1; closeOpenItem(event.output_index, item.id, entry); diff --git a/packages/ai/test/openai-responses-parallel-tool-calls.test.ts b/packages/ai/test/openai-responses-parallel-tool-calls.test.ts index 56e076e6d..9523d13db 100644 --- a/packages/ai/test/openai-responses-parallel-tool-calls.test.ts +++ b/packages/ai/test/openai-responses-parallel-tool-calls.test.ts @@ -202,4 +202,62 @@ describe("processResponsesStream: parallel function_call items", () => { expect(byCallId.get("call_a")?.toolCall.arguments).toEqual({ path: "test.txt" }); expect(byCallId.get("call_b")?.toolCall.arguments).toEqual({ path: "test.md" }); }); + + test("routes identifierless final argument events in item order", async () => { + const output = makeOutput(); + const emitted: EmittedEvent[] = []; + const stream = { push: (e: unknown) => emitted.push(e as EmittedEvent), end: () => {} } as never; + + const argsA = JSON.stringify({ command: "printf a" }); + const argsB = JSON.stringify({ command: "printf b" }); + + await processResponsesStream( + makeStream([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "bash", arguments: "" }, + }, + { + type: "response.output_item.added", + output_index: 1, + item: { type: "function_call", id: "fc_b", call_id: "call_b", name: "bash", arguments: "" }, + }, + { + type: "response.function_call_arguments.done", + arguments: argsA, + }, + { + type: "response.function_call_arguments.done", + arguments: argsB, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "bash", arguments: "" }, + }, + { + type: "response.output_item.done", + output_index: 1, + item: { type: "function_call", id: "fc_b", call_id: "call_b", name: "bash", arguments: "" }, + }, + ]), + output, + stream, + makeModel(), + ); + + const [blockA, blockB] = output.content; + if (blockA?.type !== "toolCall" || blockB?.type !== "toolCall") throw new Error("expected toolCalls"); + expect(blockA.arguments).toEqual({ command: "printf a" }); + expect(blockB.arguments).toEqual({ command: "printf b" }); + + const ends = emitted.filter(e => e.type === "toolcall_end") as Array<{ + toolCall: { id: string; arguments: Record }; + }>; + expect(ends).toHaveLength(2); + const byCallId = new Map(ends.map(e => [e.toolCall.id.split("|")[0], e])); + expect(byCallId.get("call_a")?.toolCall.arguments).toEqual({ command: "printf a" }); + expect(byCallId.get("call_b")?.toolCall.arguments).toEqual({ command: "printf b" }); + }); });