diff --git a/e2e/responses.test.ts b/e2e/responses.test.ts index bfd45cb..1351163 100644 --- a/e2e/responses.test.ts +++ b/e2e/responses.test.ts @@ -374,4 +374,82 @@ describe("openai-responses adapter through runInference", () => { { type: "text", text: "still here" }, ]); }); + + test("parallel function_call items assemble arguments from done envelopes when deltas are absent", async () => { + harness = setupHarness({ adapters: registry }); + harness.scenario.onTool("read_file", () => ({ ok: true })); + const stream = harness.scenario.createStream(); + harness.scenario.whenRequestMatches(() => true, stream); + stream.enqueueAll( + [ + sse({ + type: "response.output_item.added", + output_index: 0, + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "read_file", + }, + }), + sse({ + type: "response.output_item.added", + output_index: 1, + item: { + type: "function_call", + id: "fc_2", + call_id: "call_2", + name: "read_file", + }, + }), + sse({ + type: "response.function_call_arguments.done", + item_id: "fc_1", + arguments: '{"path":"first.ts"}', + }), + sse({ + type: "response.output_item.done", + output_index: 0, + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "read_file", + arguments: '{"path":"first.ts"}', + }, + }), + sse({ + type: "response.output_item.done", + output_index: 1, + item: { + type: "function_call", + id: "fc_2", + call_id: "call_2", + name: "read_file", + arguments: '{"path":"second.ts"}', + }, + }), + sse({ + type: "response.completed", + response: { usage: { input_tokens: 10, output_tokens: 10 } }, + }), + ], + { startAt: 1 }, + ); + const events = await collect(harness, [userTurn("read both")]); + expect(doneEvent(events).data.turn.content).toEqual([ + { + type: "tool_call", + id: "call_1", + name: "read_file", + arguments: { path: "first.ts" }, + }, + { + type: "tool_call", + id: "call_2", + name: "read_file", + arguments: { path: "second.ts" }, + }, + ]); + }); }); diff --git a/src/protocol/iterator.ts b/src/protocol/iterator.ts index ec157ad..471de6e 100644 --- a/src/protocol/iterator.ts +++ b/src/protocol/iterator.ts @@ -36,6 +36,10 @@ const FunctionCallArgsDeltaEvent = type({ item_id: "string > 0", delta: "string", }); +const FunctionCallArgsDoneEvent = type({ + item_id: "string > 0", + arguments: "string", +}); const OutputItemAddedEvent = type({ item: { @@ -43,6 +47,7 @@ const OutputItemAddedEvent = type({ "id?": "string > 0", "call_id?": "string > 0", "name?": "string > 0", + "arguments?": "string", }, }); @@ -51,6 +56,7 @@ const OutputItemDoneEvent = type({ type: "string", "id?": "string > 0", "encrypted_content?": "string", + "arguments?": "string", }, }); @@ -132,11 +138,22 @@ function toInferenceUsage( // tool calls interleave, and lets `response.output_item.done` attach an // encrypted-reasoning signature to the exact thinking block it belongs to. // `kind` is recorded so a signature is only emitted against a real thinking -// block. -type BlockKind = "text" | "thinking" | "tool_call" | "refusal"; +// block. A tool call's arguments may arrive only on `output_item.added`, +// `function_call_arguments.done`, or `output_item.done` (parallel calls on +// some backends skip deltas entirely); `hasEmittedArgs` lets those envelopes +// fill in arguments once without repeating what deltas already streamed. +type BlockEntry = + | { index: number; kind: "text" | "thinking" | "refusal" } + | { + index: number; + kind: "tool_call"; + callId: string; + hasEmittedArgs: boolean; + }; +type ToolCallBlockEntry = Extract; type ResponsesBlockIndexer = { nextIndex: number; - items: Map; + items: Map; }; export function createResponsesBlockIndexer(): ResponsesBlockIndexer { @@ -146,7 +163,7 @@ export function createResponsesBlockIndexer(): ResponsesBlockIndexer { function getOrAssignBlockIndex( state: ResponsesBlockIndexer, itemId: string, - kind: BlockKind, + kind: "text" | "thinking" | "refusal", ): number { const existing = state.items.get(itemId); if (existing !== undefined) return existing.index; @@ -156,6 +173,41 @@ function getOrAssignBlockIndex( return index; } +function registerToolCallBlock( + state: ResponsesBlockIndexer, + itemId: string, + callId: string, +): ToolCallBlockEntry { + const existing = state.items.get(itemId); + if (existing?.kind === "tool_call") return existing; + const entry: ToolCallBlockEntry = { + index: state.nextIndex, + kind: "tool_call", + callId, + hasEmittedArgs: false, + }; + state.nextIndex += 1; + state.items.set(itemId, entry); + return entry; +} + +function toolCallDeltaEvent( + entry: ToolCallBlockEntry, + argumentFragment: string, +): InferenceEvent { + entry.hasEmittedArgs = true; + return { + type: "inference.tool_call.delta", + seq: 0, + data: { + callId: entry.callId, + argumentFragment, + partial: EMPTY_PARTIAL, + index: entry.index, + }, + }; +} + function protocolMismatch( provider: string, message: string, @@ -305,6 +357,7 @@ export function parseResponse( parsed, ); } + const entry = registerToolCallBlock(indexer, item.id, item.call_id); events.push({ type: "inference.tool_call.start", seq, @@ -312,9 +365,12 @@ export function parseResponse( callId: item.call_id, name: decodeToolName(item.name), partial: EMPTY_PARTIAL, - index: getOrAssignBlockIndex(indexer, item.id, "tool_call"), + index: entry.index, }, }); + if (item.arguments !== undefined && item.arguments.length > 0) { + events.push(toolCallDeltaEvent(entry, item.arguments)); + } } else if (item.type === "reasoning") { // A reasoning item with no id cannot be pre-registered, and the // signature it will later carry on output_item.done could never be @@ -392,6 +448,32 @@ export function parseResponse( }, }); } + } else if (item.type === "function_call") { + // Without an announced block the arguments have no tool_call.start + // to attach to; dropping them would finalize the turn without the + // call. + if (item.id === undefined) { + throw protocolMismatch( + provider, + "output_item.done function_call item is missing id", + parsed, + ); + } + const existing = indexer.items.get(item.id); + if (existing?.kind !== "tool_call") { + throw protocolMismatch( + provider, + `output_item.done function_call for item_id ${item.id} was never announced by output_item.added`, + parsed, + ); + } + if ( + !existing.hasEmittedArgs && + item.arguments !== undefined && + item.arguments.length > 0 + ) { + events.push(toolCallDeltaEvent(existing, item.arguments)); + } } return events; } @@ -410,7 +492,7 @@ export function parseResponse( // tool_call.start to route it against, not a fresh block to // synthesize silently. const existing = indexer.items.get(validated.item_id); - if (existing === undefined) { + if (existing?.kind !== "tool_call") { throw protocolMismatch( provider, `function_call_arguments.delta for item_id ${validated.item_id} was never announced by output_item.added`, @@ -418,20 +500,29 @@ export function parseResponse( ); } if (validated.delta.length > 0) { - const blockIndex = existing.index; - events.push({ - type: "inference.tool_call.delta", - seq, - // The harness routes argument fragments by a per-stream - // placeholder keyed to the block index registered on the start - // event. - data: { - callId: String(blockIndex), - argumentFragment: validated.delta, - partial: EMPTY_PARTIAL, - index: blockIndex, - }, - }); + events.push(toolCallDeltaEvent(existing, validated.delta)); + } + return events; + } + case "response.function_call_arguments.done": { + const validated = FunctionCallArgsDoneEvent(parsed); + if (validated instanceof type.errors) { + throw protocolMismatch( + provider, + `function_call_arguments.done failed schema validation: ${validated.summary}`, + parsed, + ); + } + const existing = indexer.items.get(validated.item_id); + if (existing?.kind !== "tool_call") { + throw protocolMismatch( + provider, + `function_call_arguments.done for item_id ${validated.item_id} was never announced by output_item.added`, + parsed, + ); + } + if (!existing.hasEmittedArgs && validated.arguments.length > 0) { + events.push(toolCallDeltaEvent(existing, validated.arguments)); } return events; } @@ -738,7 +829,7 @@ export function parseJSONResponse( type: "inference.tool_call.delta", seq, data: { - callId: String(index), + callId: item.call_id, argumentFragment: item.arguments, partial: EMPTY_PARTIAL, index, diff --git a/src/responses.test.ts b/src/responses.test.ts index 52f90ca..eba41da 100644 --- a/src/responses.test.ts +++ b/src/responses.test.ts @@ -400,6 +400,174 @@ describe("Responses parser — event schema validation", () => { ProtocolMismatchError, ); }); + + test("function_call_arguments.done or output_item.done for an unannounced function_call throws", () => { + const adapter = createOpenAIResponsesAdapter(source, {}); + adapter.buildRequest(turns, "model", {}); + expect(() => + adapter.parseResponse( + JSON.stringify({ + type: "response.function_call_arguments.done", + item_id: "never_announced", + arguments: '{"cmd":"ls"}', + }), + ), + ).toThrow(ProtocolMismatchError); + expect(() => + adapter.parseResponse( + JSON.stringify({ + type: "response.output_item.done", + item: { + type: "function_call", + id: "never_announced", + arguments: '{"cmd":"ls"}', + }, + }), + ), + ).toThrow(ProtocolMismatchError); + }); +}); + +describe("Responses parser — parallel function_call arguments", () => { + type ToolCallDelta = Extract< + InferenceEvent, + { type: "inference.tool_call.delta" } + >; + + // Each path must carry the real callId, not the block-index placeholder. + test("routes parallel function_call arguments across added, delta, arguments.done, and output_item.done without duplication", () => { + const adapter = createOpenAIResponsesAdapter(source, {}); + adapter.buildRequest(turns, "model", {}); + + const frames = [ + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_added", + call_id: "call_added", + name: "read_file", + arguments: '{"path":"a.ts"}', + }, + }, + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_delta", + call_id: "call_delta", + name: "read_file", + arguments: "", + }, + }, + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_args_done", + call_id: "call_args_done", + name: "read_file", + }, + }, + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_item_done", + call_id: "call_item_done", + name: "read_file", + }, + }, + { + type: "response.function_call_arguments.delta", + item_id: "fc_delta", + delta: '{"path":"b.ts"}', + }, + { + type: "response.function_call_arguments.done", + item_id: "fc_added", + arguments: '{"path":"a.ts"}', + }, + { + type: "response.function_call_arguments.done", + item_id: "fc_delta", + arguments: '{"path":"b.ts"}', + }, + { + type: "response.function_call_arguments.done", + item_id: "fc_args_done", + arguments: '{"path":"c.ts"}', + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_added", + call_id: "call_added", + name: "read_file", + arguments: '{"path":"a.ts"}', + }, + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_delta", + call_id: "call_delta", + name: "read_file", + arguments: '{"path":"b.ts"}', + }, + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_args_done", + call_id: "call_args_done", + name: "read_file", + arguments: '{"path":"c.ts"}', + }, + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_item_done", + call_id: "call_item_done", + name: "read_file", + arguments: '{"path":"d.ts"}', + }, + }, + ]; + + const emitted = frames.flatMap((frame) => + adapter.parseResponse(JSON.stringify(frame)), + ); + const deltas = emitted.filter( + (e): e is ToolCallDelta => e.type === "inference.tool_call.delta", + ); + + expect( + deltas.map((d) => ({ + callId: d.data.callId, + argumentFragment: d.data.argumentFragment, + index: d.data.index, + })), + ).toEqual([ + { callId: "call_added", argumentFragment: '{"path":"a.ts"}', index: 0 }, + { callId: "call_delta", argumentFragment: '{"path":"b.ts"}', index: 1 }, + { + callId: "call_args_done", + argumentFragment: '{"path":"c.ts"}', + index: 2, + }, + { + callId: "call_item_done", + argumentFragment: '{"path":"d.ts"}', + index: 3, + }, + ]); + }); }); describe("Responses parser — non-streaming failure states", () => {