smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ai): preserved llama parallel tool args

Browse Source 
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
This commit is contained in:
roboomp
2026-06-06 04:30:05 +00:00
parent d9f5a8c7ac
commit 699adfb386
3 changed files with 96 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/CHANGELOG.md
+4
View File
@@ -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
packages/ai/src/providers/openai-responses-shared.ts
+34 -6
View File
@@ -395,7 +395,7 @@ export async function processResponsesStream<TApi extends Api>(
model: Model<TApi>,
options?: ProcessResponsesStreamOptions,
): Promise<void> {
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<TApi extends Api>(
const openItemsByOutputIndex = new Map<number, StreamingItem>();
const openItemsByItemId = new Map<string, StreamingItem>();
let lastOpenItem: StreamingItem | null = null;
const openItemsInOrder: StreamingItem[] = [];
const registerOpenItem = (
outputIndex: number | undefined,
@@ -417,6 +418,7 @@ export async function processResponsesStream<TApi extends Api>(
): 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<TApi extends Api>(
// 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<TApi extends Api>(
): 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<TApi extends Api>(
}
}
} 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<TApi extends Api>(
});
}
} 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<TApi extends Api>(
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<TApi extends Api>(
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);
packages/ai/test/openai-responses-parallel-tool-calls.test.ts
+58
View File
@@ -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<string, unknown> };
}>;
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" });
});
});