diff --git a/.changeset/fix-responses-tool-arguments.md b/.changeset/fix-responses-tool-arguments.md new file mode 100644 index 000000000..1c0a91947 --- /dev/null +++ b/.changeset/fix-responses-tool-arguments.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix tool calls from OpenAI Responses-compatible providers when final arguments correct streamed partial arguments. diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts index 9953a0d14..0368783cc 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts @@ -733,28 +733,31 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { private async *_convertStreamResponse( response: AsyncIterable, ): AsyncGenerator { - const functionCallArgumentsByIndex = new Map(); - let unindexedFunctionCallArguments: string | undefined; - - const hasFunctionCallArguments = (streamIndex: number | string | undefined): boolean => - streamIndex === undefined - ? unindexedFunctionCallArguments !== undefined - : functionCallArgumentsByIndex.has(streamIndex); + type FunctionCallState = { + toolCall: ToolCall; + accumulatedArguments: string; + functionDoneArguments?: string; + committed: boolean; + }; - const getFunctionCallArguments = (streamIndex: number | string | undefined): string => - streamIndex === undefined - ? (unindexedFunctionCallArguments as string) - : functionCallArgumentsByIndex.get(streamIndex)!; + const functionCallsByIndex = new Map(); + let unindexedFunctionCall: FunctionCallState | undefined; - const setFunctionCallArguments = ( + const getFunctionCall = ( streamIndex: number | string | undefined, - argumentsValue: string, - ): void => { - if (streamIndex === undefined) { - unindexedFunctionCallArguments = argumentsValue; - } else { - functionCallArgumentsByIndex.set(streamIndex, argumentsValue); + context: string, + ): FunctionCallState => { + const state = + streamIndex === undefined + ? unindexedFunctionCall + : functionCallsByIndex.get(streamIndex); + if (state === undefined) { + failResponsesDecode( + context, + `received function-call arguments for unknown stream index ${formatResponseStreamIndex(streamIndex)}.`, + ); } + return state; }; const appendFunctionCallArguments = ( @@ -762,57 +765,32 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { argumentsPart: string, context: string, ): void => { - if (!hasFunctionCallArguments(streamIndex)) { - failResponsesDecode( - context, - `received function-call arguments for unknown stream index ${formatResponseStreamIndex(streamIndex)}.`, - ); - } - setFunctionCallArguments( - streamIndex, - getFunctionCallArguments(streamIndex) + argumentsPart, - ); + const state = getFunctionCall(streamIndex, context); + state.accumulatedArguments += argumentsPart; }; - const yieldFinalArgumentsSuffix = function* ( + const commitFunctionCall = function* ( streamIndex: number | string | undefined, finalArguments: string, context: string, ): Generator { - if (!hasFunctionCallArguments(streamIndex)) { - failResponsesDecode( - context, - `received final function-call arguments for unknown stream index ${formatResponseStreamIndex(streamIndex)}.`, - ); - } - - const accumulatedArguments = getFunctionCallArguments(streamIndex); - if (finalArguments === accumulatedArguments) { + const state = getFunctionCall(streamIndex, context); + if (state.committed) { return; } + state.committed = true; - if (!finalArguments.startsWith(accumulatedArguments)) { - throw new ChatProviderError( - `OpenAI Responses final function-call arguments for stream index ${formatResponseStreamIndex( - streamIndex, - )} do not match the streamed argument deltas.`, - ); - } - - const suffix = finalArguments.slice(accumulatedArguments.length); - setFunctionCallArguments(streamIndex, finalArguments); - if (suffix.length === 0) { + if (streamIndex === undefined) { + unindexedFunctionCall = undefined; + yield { ...state.toolCall, arguments: finalArguments }; return; } - const part: StreamedMessagePart = { + yield { type: 'tool_call_part', - argumentsPart: suffix, + argumentsPart: finalArguments, + index: streamIndex, }; - if (streamIndex !== undefined) { - (part as { index: number | string }).index = streamIndex; - } - yield part; }; try { @@ -845,18 +823,31 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { const item = readResponseOutputItem(chunk['item'], `${type}.item`); const outputIndex = readNumberField(chunk, 'output_index'); if (item.type === 'function_call') { + if (unindexedFunctionCall !== undefined) { + failResponsesDecode( + `${type}.item`, + 'cannot start a function call while an unindexed function call is unresolved.', + ); + } const streamIndex = responseStreamIndex(item.itemId, outputIndex); - setFunctionCallArguments(streamIndex, item.arguments ?? ''); const tc: ToolCall = { type: 'function', id: functionCallId(item.callId), name: requireFunctionCallName(item), - arguments: item.arguments ?? null, + arguments: null, + }; + const state: FunctionCallState = { + toolCall: tc, + accumulatedArguments: item.arguments ?? '', + committed: false, }; if (streamIndex !== undefined) { tc._streamIndex = streamIndex; + functionCallsByIndex.set(streamIndex, state); + yield tc; + } else { + unindexedFunctionCall = state; } - yield tc; } break; } @@ -871,7 +862,7 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { yield thinkPart; } else if (item.type === 'function_call' && typeof item.arguments === 'string') { const streamIndex = responseStreamIndex(item.itemId, outputIndex); - yield* yieldFinalArgumentsSuffix(streamIndex, item.arguments, type); + yield* commitFunctionCall(streamIndex, item.arguments, type); } break; } @@ -881,15 +872,7 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { readNumberField(chunk, 'output_index'), ); const argumentsPart = requireStringField(chunk, 'delta', type); - const part: StreamedMessagePart = { - type: 'tool_call_part', - argumentsPart, - }; appendFunctionCallArguments(streamIndex, argumentsPart, type); - if (streamIndex !== undefined) { - (part as { index: number | string }).index = streamIndex; - } - yield part; break; } case 'response.function_call_arguments.done': { @@ -898,7 +881,8 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { readStringField(chunk, 'item_id'), readNumberField(chunk, 'output_index'), ); - yield* yieldFinalArgumentsSuffix(streamIndex, functionArguments, type); + const state = getFunctionCall(streamIndex, type); + state.functionDoneArguments ??= functionArguments; break; } case 'response.reasoning_summary_part.added': @@ -952,6 +936,24 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { } catch (error: unknown) { throw convertOpenAIError(error); } + + for (const [streamIndex, state] of functionCallsByIndex) { + if (!state.committed) { + yield* commitFunctionCall( + streamIndex, + state.functionDoneArguments ?? state.accumulatedArguments, + 'OpenAI Responses stream completion', + ); + } + } + if (unindexedFunctionCall !== undefined) { + const state = unindexedFunctionCall; + yield* commitFunctionCall( + undefined, + state.functionDoneArguments ?? state.accumulatedArguments, + 'OpenAI Responses stream completion', + ); + } } } export class OpenAIResponsesChatProvider implements ChatProvider { diff --git a/packages/agent-core-v2/test/app/llmProtocol/providers/openai-responses.test.ts b/packages/agent-core-v2/test/app/llmProtocol/providers/openai-responses.test.ts new file mode 100644 index 000000000..3d18286d0 --- /dev/null +++ b/packages/agent-core-v2/test/app/llmProtocol/providers/openai-responses.test.ts @@ -0,0 +1,464 @@ +/** + * `llmProtocol` OpenAI Responses streaming contract — final tool arguments + * override provisional deltas without leaking partial values or mixing calls. + */ + +import { describe, expect, it, vi } from 'vitest'; + +import { ChatProviderError } from '#/app/llmProtocol/errors'; +import { generate } from '#/app/llmProtocol/generate'; +import type { StreamedMessagePart, ToolCall } from '#/app/llmProtocol/message'; +import { + OpenAIResponsesChatProvider, + OpenAIResponsesStreamedMessage, +} from '#/app/llmProtocol/providers/openai-responses'; + +describe('OpenAI Responses authoritative function-call arguments', () => { + it('uses function_call_arguments.done when it corrects streamed deltas', async () => { + const events = [ + functionCallAdded('item_corrected', 'call_corrected', 'add'), + functionArgumentsDelta('item_corrected', '{"a":1}'), + functionArgumentsDone('item_corrected', '{"a":2}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_corrected', 'call_corrected', 'add'), + indexedArguments('item_corrected', '{"a":2}'), + ]); + }); + + it('prefers output_item.done over function_call_arguments.done and buffered deltas', async () => { + const events = [ + functionCallAdded('item_precedence', 'call_precedence', 'add', '{"draft":'), + functionArgumentsDelta('item_precedence', '1}'), + functionArgumentsDone('item_precedence', '{"source":"function-done"}'), + functionCallOutputDone('item_precedence', '{"source":"output-item"}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_precedence', 'call_precedence', 'add'), + indexedArguments('item_precedence', '{"source":"output-item"}'), + ]); + }); + + it.each(['response.completed', 'response.incomplete'] as const)( + '%s waits for normal EOF before emitting accumulated fallback arguments', + async (terminationType) => { + const terminationEvent = + terminationType === 'response.completed' + ? { + type: terminationType, + response: { id: 'resp_eof', status: 'completed' }, + } + : { + type: terminationType, + response: { + id: 'resp_eof', + status: 'incomplete', + incomplete_details: { reason: 'max_output_tokens' }, + }, + }; + const controlled = controlledEOFIterable([ + functionCallAdded('item_eof', 'call_eof', 'add', '{"a":'), + functionArgumentsDelta('item_eof', '1}'), + terminationEvent, + ]); + const iterator = new OpenAIResponsesStreamedMessage( + controlled.iterable, + true, + )[Symbol.asyncIterator](); + + expect(await iterator.next()).toEqual({ + done: false, + value: indexedToolCall('item_eof', 'call_eof', 'add'), + }); + + let settled = false; + const fallback = iterator.next().then((result) => { + settled = true; + return result; + }); + await controlled.waitingForEOF; + expect(settled).toBe(false); + + controlled.finish(); + expect(await fallback).toEqual({ + done: false, + value: indexedArguments('item_eof', '{"a":1}'), + }); + expect(await iterator.next()).toEqual({ done: true, value: undefined }); + }, + ); + + it.each([ + ['equal', '{"a":1}', '', '{"a":1}'], + ['prefix extension', '{"a":', '', '{"a":1}'], + ])('emits one complete part for a %s function-done final', async (_label, delta, initial, final) => { + const events = [ + functionCallAdded('item_compat', 'call_compat', 'add', initial), + functionArgumentsDelta('item_compat', delta), + functionArgumentsDone('item_compat', final), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_compat', 'call_compat', 'add'), + indexedArguments('item_compat', final), + ]); + }); + + it.each([ + ['missing', {}], + ['null', { arguments: null }], + ])( + 'falls back to function-done arguments when output_item.done arguments are %s', + async (_label, outputFields) => { + const events = [ + functionCallAdded('item_null', 'call_null', 'add'), + functionArgumentsDelta('item_null', '{"draft":true}'), + functionArgumentsDone('item_null', '{"final":true}'), + { + type: 'response.output_item.done', + item: { type: 'function_call', id: 'item_null', ...outputFields }, + }, + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_null', 'call_null', 'add'), + indexedArguments('item_null', '{"final":true}'), + ]); + }, + ); + + it('keeps parallel calls isolated and preserves header order after reverse completion', async () => { + const events = [ + functionCallAdded('item_a', 'call_a', 'add', '{"draft":'), + functionCallAdded('item_b', 'call_b', 'multiply'), + functionArgumentsDelta('item_a', '1}'), + functionArgumentsDelta('item_b', '{"draft":2}'), + functionArgumentsDone('item_a', '{"a":1}'), + functionArgumentsDone('item_b', '{"b":2}'), + functionCallOutputDone('item_b', '{"b":20}'), + functionCallOutputDone('item_a', '{"a":10}'), + ]; + const rawParts: StreamedMessagePart[] = []; + + const result = await generate( + streamingProvider(events), + '', + [], + [{ role: 'user', content: [{ type: 'text', text: 'run tools' }], toolCalls: [] }], + { + onMessagePart: (part) => { + rawParts.push(part); + }, + }, + ); + + expect(rawParts).toEqual([ + indexedToolCall('item_a', 'call_a', 'add'), + indexedToolCall('item_b', 'call_b', 'multiply'), + indexedArguments('item_b', '{"b":20}'), + indexedArguments('item_a', '{"a":10}'), + ]); + expect(result.message.toolCalls).toEqual([ + { + type: 'function', + id: 'call_a', + name: 'add', + arguments: '{"a":10}', + extras: undefined, + }, + { + type: 'function', + id: 'call_b', + name: 'multiply', + arguments: '{"b":20}', + extras: undefined, + }, + ]); + }); + + it.each([ + ['delta', functionArgumentsDelta('missing_item', '{}')], + ['done', functionArgumentsDone('missing_item', '{}')], + ])('rejects %s arguments for an unknown stream index', async (_label, event) => { + const stream = new OpenAIResponsesStreamedMessage(makeAsyncIterable([event]), true); + await expect(collectStreamParts(stream)).rejects.toThrow(/unknown stream index missing_item/); + }); + + it('buffers an unindexed call as a whole call across intervening text and reasoning', async () => { + const events = [ + functionCallAdded(undefined, 'call_unindexed', 'add'), + functionArgumentsDelta(undefined, '{"draft":1}'), + { type: 'response.output_text.delta', delta: 'between' }, + { type: 'response.reasoning_summary_text.delta', delta: 'thinking' }, + functionCallOutputDone(undefined, '{"final":2}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + { type: 'text', text: 'between' }, + { type: 'think', think: 'thinking' }, + { + type: 'function', + id: 'call_unindexed', + name: 'add', + arguments: '{"final":2}', + }, + ]); + }); + + it.each([ + ['another unindexed', undefined], + ['an indexed', 'item_later'], + ])('rejects %s function-call header while an unindexed call is unresolved', async (_label, id) => { + const events = [ + functionCallAdded(undefined, 'call_unindexed', 'add'), + { type: 'response.output_text.delta', delta: 'between' }, + functionCallAdded(id, 'call_later', 'multiply'), + ]; + const { parts, error } = await collectPartsAndError( + new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true), + ); + + expect(parts).toEqual([{ type: 'text', text: 'between' }]); + expect(error).toBeInstanceOf(ChatProviderError); + expect((error as Error).message).toMatch(/unindexed function call.*unresolved/i); + }); + + it('allows a sequential unindexed call after the prior call commits', async () => { + const events = [ + functionCallAdded(undefined, 'call_first', 'add'), + functionCallOutputDone(undefined, '{"first":1}'), + functionCallAdded(undefined, 'call_second', 'multiply'), + functionCallOutputDone(undefined, '{"second":2}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + { type: 'function', id: 'call_first', name: 'add', arguments: '{"first":1}' }, + { + type: 'function', + id: 'call_second', + name: 'multiply', + arguments: '{"second":2}', + }, + ]); + }); + + it.each([ + [ + 'error', + { type: 'error', code: 'server_error', message: 'upstream failed', param: null }, + ], + [ + 'response.failed', + { + type: 'response.failed', + response: { + id: 'resp_failed', + status: 'failed', + error: { code: 'server_error', message: 'upstream failed' }, + }, + }, + ], + ['decode failure', { type: 42 }], + ])('%s does not flush buffered arguments', async (_label, failureEvent) => { + const events = [ + functionCallAdded('item_failure', 'call_failure', 'add'), + functionArgumentsDelta('item_failure', '{"partial":'), + failureEvent, + ]; + const { parts, error } = await collectPartsAndError( + new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true), + ); + + expect(parts).toEqual([indexedToolCall('item_failure', 'call_failure', 'add')]); + expect(error).toBeInstanceOf(Error); + }); + + it('does not flush buffered arguments when the source iterator throws', async () => { + async function* failingEvents(): AsyncIterable> { + yield functionCallAdded('item_throw', 'call_throw', 'add'); + yield functionArgumentsDelta('item_throw', '{"partial":'); + throw new Error('iterator boom'); + } + const { parts, error } = await collectPartsAndError( + new OpenAIResponsesStreamedMessage(failingEvents(), true), + ); + + expect(parts).toEqual([indexedToolCall('item_throw', 'call_throw', 'add')]); + expect((error as Error).message).toContain('iterator boom'); + }); + + it('does not flush buffered arguments when the consumer returns early', async () => { + const stream = new OpenAIResponsesStreamedMessage( + makeAsyncIterable([ + functionCallAdded('item_cancel', 'call_cancel', 'add', '{"initial":true}'), + functionArgumentsDelta('item_cancel', '{"partial":'), + ]), + true, + ); + const iterator = stream[Symbol.asyncIterator](); + + expect(await iterator.next()).toEqual({ + done: false, + value: indexedToolCall('item_cancel', 'call_cancel', 'add'), + }); + expect(await iterator.return?.()).toEqual({ done: true, value: undefined }); + }); +}); + +async function collectStreamParts( + stream: OpenAIResponsesStreamedMessage, +): Promise { + const parts: StreamedMessagePart[] = []; + for await (const part of stream) parts.push(part); + return parts; +} + +function makeAsyncIterable( + events: Record[], +): AsyncIterable> { + return { + [Symbol.asyncIterator](): AsyncIterator> { + let index = 0; + return { + next(): Promise>> { + if (index < events.length) { + return Promise.resolve({ value: events[index++]!, done: false }); + } + return Promise.resolve({ + value: undefined as unknown as Record, + done: true, + }); + }, + }; + }, + }; +} + +function functionCallAdded( + itemId: string | undefined, + callId: string, + name: string, + argumentsValue: string = '', +): Record { + return { + type: 'response.output_item.added', + item: { + id: itemId, + type: 'function_call', + call_id: callId, + name, + arguments: argumentsValue, + }, + }; +} + +function functionArgumentsDelta( + itemId: string | undefined, + delta: string, +): Record { + return { type: 'response.function_call_arguments.delta', item_id: itemId, delta }; +} + +function functionArgumentsDone( + itemId: string | undefined, + argumentsValue: string, +): Record { + return { + type: 'response.function_call_arguments.done', + item_id: itemId, + arguments: argumentsValue, + }; +} + +function functionCallOutputDone( + itemId: string | undefined, + argumentsValue: string, +): Record { + return { + type: 'response.output_item.done', + item: { id: itemId, type: 'function_call', arguments: argumentsValue }, + }; +} + +function indexedToolCall(index: string, id: string, name: string): ToolCall { + return { type: 'function', id, name, arguments: null, _streamIndex: index }; +} + +function indexedArguments(index: string, argumentsPart: string): StreamedMessagePart { + return { type: 'tool_call_part', argumentsPart, index }; +} + +function streamingProvider(events: Record[]): OpenAIResponsesChatProvider { + const create = vi.fn().mockResolvedValue(makeAsyncIterable(events)); + return new OpenAIResponsesChatProvider({ + model: 'gpt-5', + apiKey: '', + clientFactory: () => ({ responses: { create } }) as never, + }); +} + +async function collectPartsAndError( + stream: OpenAIResponsesStreamedMessage, +): Promise<{ parts: StreamedMessagePart[]; error: unknown }> { + const parts: StreamedMessagePart[] = []; + let caughtError: unknown; + try { + for await (const part of stream) parts.push(part); + } catch (error) { + caughtError = error; + } + return { parts, error: caughtError }; +} + +function controlledEOFIterable(events: Record[]): { + iterable: AsyncIterable>; + waitingForEOF: Promise; + finish: () => void; +} { + let index = 0; + let finish: () => void = () => undefined; + let notifyWaiting: () => void = () => undefined; + const waitingForEOF = new Promise((resolve) => { + notifyWaiting = resolve; + }); + const eof = new Promise>>((resolve) => { + finish = () => { + resolve({ + value: undefined as unknown as Record, + done: true, + }); + }; + }); + + return { + iterable: { + [Symbol.asyncIterator](): AsyncIterator> { + return { + next(): Promise>> { + if (index < events.length) { + return Promise.resolve({ value: events[index++]!, done: false }); + } + notifyWaiting(); + return eof; + }, + }; + }, + }, + waitingForEOF, + finish, + }; +} diff --git a/packages/kosong/src/providers/openai-responses.ts b/packages/kosong/src/providers/openai-responses.ts index baf8666ab..8516038bb 100644 --- a/packages/kosong/src/providers/openai-responses.ts +++ b/packages/kosong/src/providers/openai-responses.ts @@ -772,28 +772,31 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { private async *_convertStreamResponse( response: AsyncIterable, ): AsyncGenerator { - const functionCallArgumentsByIndex = new Map(); - let unindexedFunctionCallArguments: string | undefined; - - const hasFunctionCallArguments = (streamIndex: number | string | undefined): boolean => - streamIndex === undefined - ? unindexedFunctionCallArguments !== undefined - : functionCallArgumentsByIndex.has(streamIndex); + type FunctionCallState = { + toolCall: ToolCall; + accumulatedArguments: string; + functionDoneArguments?: string; + committed: boolean; + }; - const getFunctionCallArguments = (streamIndex: number | string | undefined): string => - streamIndex === undefined - ? (unindexedFunctionCallArguments as string) - : functionCallArgumentsByIndex.get(streamIndex)!; + const functionCallsByIndex = new Map(); + let unindexedFunctionCall: FunctionCallState | undefined; - const setFunctionCallArguments = ( + const getFunctionCall = ( streamIndex: number | string | undefined, - argumentsValue: string, - ): void => { - if (streamIndex === undefined) { - unindexedFunctionCallArguments = argumentsValue; - } else { - functionCallArgumentsByIndex.set(streamIndex, argumentsValue); + context: string, + ): FunctionCallState => { + const state = + streamIndex === undefined + ? unindexedFunctionCall + : functionCallsByIndex.get(streamIndex); + if (state === undefined) { + failResponsesDecode( + context, + `received function-call arguments for unknown stream index ${formatResponseStreamIndex(streamIndex)}.`, + ); } + return state; }; const appendFunctionCallArguments = ( @@ -801,57 +804,32 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { argumentsPart: string, context: string, ): void => { - if (!hasFunctionCallArguments(streamIndex)) { - failResponsesDecode( - context, - `received function-call arguments for unknown stream index ${formatResponseStreamIndex(streamIndex)}.`, - ); - } - setFunctionCallArguments( - streamIndex, - getFunctionCallArguments(streamIndex) + argumentsPart, - ); + const state = getFunctionCall(streamIndex, context); + state.accumulatedArguments += argumentsPart; }; - const yieldFinalArgumentsSuffix = function* ( + const commitFunctionCall = function* ( streamIndex: number | string | undefined, finalArguments: string, context: string, ): Generator { - if (!hasFunctionCallArguments(streamIndex)) { - failResponsesDecode( - context, - `received final function-call arguments for unknown stream index ${formatResponseStreamIndex(streamIndex)}.`, - ); - } - - const accumulatedArguments = getFunctionCallArguments(streamIndex); - if (finalArguments === accumulatedArguments) { + const state = getFunctionCall(streamIndex, context); + if (state.committed) { return; } + state.committed = true; - if (!finalArguments.startsWith(accumulatedArguments)) { - throw new ChatProviderError( - `OpenAI Responses final function-call arguments for stream index ${formatResponseStreamIndex( - streamIndex, - )} do not match the streamed argument deltas.`, - ); - } - - const suffix = finalArguments.slice(accumulatedArguments.length); - setFunctionCallArguments(streamIndex, finalArguments); - if (suffix.length === 0) { + if (streamIndex === undefined) { + unindexedFunctionCall = undefined; + yield { ...state.toolCall, arguments: finalArguments }; return; } - const part: StreamedMessagePart = { + yield { type: 'tool_call_part', - argumentsPart: suffix, + argumentsPart: finalArguments, + index: streamIndex, }; - if (streamIndex !== undefined) { - (part as { index: number | string }).index = streamIndex; - } - yield part; }; try { @@ -892,22 +870,31 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { // would clobber the real response id (or leave it undefined for // tool-call items that have no `item.id`). if (item.type === 'function_call') { - // The Responses API routes streaming argument deltas via - // `item_id`, which matches `item.id` on output_item.added. - // Preserve it so the generate loop can dispatch interleaved - // deltas across parallel function calls correctly. + if (unindexedFunctionCall !== undefined) { + failResponsesDecode( + `${type}.item`, + 'cannot start a function call while an unindexed function call is unresolved.', + ); + } const streamIndex = responseStreamIndex(item.itemId, outputIndex); - setFunctionCallArguments(streamIndex, item.arguments ?? ''); const tc: ToolCall = { type: 'function', id: functionCallId(item.callId), name: requireFunctionCallName(item), - arguments: item.arguments ?? null, + arguments: null, + }; + const state: FunctionCallState = { + toolCall: tc, + accumulatedArguments: item.arguments ?? '', + committed: false, }; if (streamIndex !== undefined) { tc._streamIndex = streamIndex; + functionCallsByIndex.set(streamIndex, state); + yield tc; + } else { + unindexedFunctionCall = state; } - yield tc; } break; } @@ -923,7 +910,7 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { yield thinkPart; } else if (item.type === 'function_call' && typeof item.arguments === 'string') { const streamIndex = responseStreamIndex(item.itemId, outputIndex); - yield* yieldFinalArgumentsSuffix(streamIndex, item.arguments, type); + yield* commitFunctionCall(streamIndex, item.arguments, type); } break; } @@ -935,15 +922,7 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { readNumberField(chunk, 'output_index'), ); const argumentsPart = requireStringField(chunk, 'delta', type); - const part: StreamedMessagePart = { - type: 'tool_call_part', - argumentsPart, - }; appendFunctionCallArguments(streamIndex, argumentsPart, type); - if (streamIndex !== undefined) { - (part as { index: number | string }).index = streamIndex; - } - yield part; break; } case 'response.function_call_arguments.done': { @@ -952,7 +931,8 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { readStringField(chunk, 'item_id'), readNumberField(chunk, 'output_index'), ); - yield* yieldFinalArgumentsSuffix(streamIndex, functionArguments, type); + const state = getFunctionCall(streamIndex, type); + state.functionDoneArguments ??= functionArguments; break; } case 'response.reasoning_summary_part.added': @@ -1009,6 +989,24 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { } catch (error: unknown) { throw convertOpenAIError(error); } + + for (const [streamIndex, state] of functionCallsByIndex) { + if (!state.committed) { + yield* commitFunctionCall( + streamIndex, + state.functionDoneArguments ?? state.accumulatedArguments, + 'OpenAI Responses stream completion', + ); + } + } + if (unindexedFunctionCall !== undefined) { + const state = unindexedFunctionCall; + yield* commitFunctionCall( + undefined, + state.functionDoneArguments ?? state.accumulatedArguments, + 'OpenAI Responses stream completion', + ); + } } } export class OpenAIResponsesChatProvider implements ChatProvider { diff --git a/packages/kosong/test/openai-responses.test.ts b/packages/kosong/test/openai-responses.test.ts index 1a8d4c752..ca9edaaef 100644 --- a/packages/kosong/test/openai-responses.test.ts +++ b/packages/kosong/test/openai-responses.test.ts @@ -1531,18 +1531,19 @@ describe('OpenAIResponsesChatProvider', () => { parts.push(part); } - // The Responses adapter absorbs `function_call_arguments.done` and - // `output_item.done` for function_call items — generate.ts infers - // completion from merge boundaries and stream end. expect(parts).toEqual([ { type: 'function', id: 'call_123', - name: 'add', arguments: '', + name: 'add', + arguments: null, _streamIndex: 'item_123', }, - { type: 'tool_call_part', argumentsPart: '{"a":', index: 'item_123' }, - { type: 'tool_call_part', argumentsPart: ' 2, "b": 3}', index: 'item_123' }, + { + type: 'tool_call_part', + argumentsPart: '{"a": 2, "b": 3}', + index: 'item_123', + }, ]); expect(stream.usage).toEqual({ @@ -1600,7 +1601,7 @@ describe('OpenAIResponsesChatProvider', () => { ]); }); - it('rejects function_call_arguments.done when it disagrees with streamed deltas', async () => { + it('uses function_call_arguments.done when it corrects streamed deltas', async () => { const events = [ { type: 'response.output_item.added', @@ -1629,9 +1630,20 @@ describe('OpenAIResponsesChatProvider', () => { const stream = new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true); - await expect(collectStreamParts(stream)).rejects.toThrow( - /final function-call arguments.*do not match/, - ); + await expect(collectStreamParts(stream)).resolves.toEqual([ + { + type: 'function', + id: 'call_mismatch', + name: 'add', + arguments: null, + _streamIndex: 'item_mismatch', + }, + { + type: 'tool_call_part', + argumentsPart: '{"a": 2}', + index: 'item_mismatch', + }, + ]); }); it('streams reasoning with encrypted_content', async () => { @@ -2094,6 +2106,297 @@ describe('OpenAIResponsesChatProvider', () => { }); }); +describe('OpenAI Responses authoritative function-call arguments', () => { + it('prefers output_item.done over function_call_arguments.done and buffered deltas', async () => { + const events = [ + functionCallAdded('item_precedence', 'call_precedence', 'add', '{"draft":'), + functionArgumentsDelta('item_precedence', '1}'), + functionArgumentsDone('item_precedence', '{"source":"function-done"}'), + functionCallOutputDone('item_precedence', '{"source":"output-item"}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_precedence', 'call_precedence', 'add'), + indexedArguments('item_precedence', '{"source":"output-item"}'), + ]); + }); + + it.each(['response.completed', 'response.incomplete'] as const)( + '%s waits for normal EOF before emitting accumulated fallback arguments', + async (terminationType) => { + const terminationEvent = + terminationType === 'response.completed' + ? { + type: terminationType, + response: { id: 'resp_eof', status: 'completed' }, + } + : { + type: terminationType, + response: { + id: 'resp_eof', + status: 'incomplete', + incomplete_details: { reason: 'max_output_tokens' }, + }, + }; + const controlled = controlledEOFIterable([ + functionCallAdded('item_eof', 'call_eof', 'add', '{"a":'), + functionArgumentsDelta('item_eof', '1}'), + terminationEvent, + ]); + const iterator = new OpenAIResponsesStreamedMessage( + controlled.iterable, + true, + )[Symbol.asyncIterator](); + + expect(await iterator.next()).toEqual({ + done: false, + value: indexedToolCall('item_eof', 'call_eof', 'add'), + }); + + let settled = false; + const fallback = iterator.next().then((result) => { + settled = true; + return result; + }); + await controlled.waitingForEOF; + expect(settled).toBe(false); + + controlled.finish(); + expect(await fallback).toEqual({ + done: false, + value: indexedArguments('item_eof', '{"a":1}'), + }); + expect(await iterator.next()).toEqual({ done: true, value: undefined }); + }, + ); + + it.each([ + ['equal', '{"a":1}', '', '{"a":1}'], + ['prefix extension', '{"a":', '', '{"a":1}'], + ])('emits one complete part for a %s function-done final', async (_label, delta, initial, final) => { + const events = [ + functionCallAdded('item_compat', 'call_compat', 'add', initial), + functionArgumentsDelta('item_compat', delta), + functionArgumentsDone('item_compat', final), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_compat', 'call_compat', 'add'), + indexedArguments('item_compat', final), + ]); + }); + + it.each([ + ['missing', {}], + ['null', { arguments: null }], + ])( + 'falls back to function-done arguments when output_item.done arguments are %s', + async (_label, outputFields) => { + const events = [ + functionCallAdded('item_null', 'call_null', 'add'), + functionArgumentsDelta('item_null', '{"draft":true}'), + functionArgumentsDone('item_null', '{"final":true}'), + { + type: 'response.output_item.done', + item: { type: 'function_call', id: 'item_null', ...outputFields }, + }, + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + indexedToolCall('item_null', 'call_null', 'add'), + indexedArguments('item_null', '{"final":true}'), + ]); + }, + ); + + it('keeps parallel calls isolated and preserves header order after reverse completion', async () => { + const events = [ + functionCallAdded('item_a', 'call_a', 'add', '{"draft":'), + functionCallAdded('item_b', 'call_b', 'multiply'), + functionArgumentsDelta('item_a', '1}'), + functionArgumentsDelta('item_b', '{"draft":2}'), + functionArgumentsDone('item_a', '{"a":1}'), + functionArgumentsDone('item_b', '{"b":2}'), + functionCallOutputDone('item_b', '{"b":20}'), + functionCallOutputDone('item_a', '{"a":10}'), + ]; + const rawParts: StreamedMessagePart[] = []; + + const result = await generate( + streamingProvider(events), + '', + [], + [{ role: 'user', content: [{ type: 'text', text: 'run tools' }], toolCalls: [] }], + { + onMessagePart: (part) => { + rawParts.push(part); + }, + }, + ); + + expect(rawParts).toEqual([ + indexedToolCall('item_a', 'call_a', 'add'), + indexedToolCall('item_b', 'call_b', 'multiply'), + indexedArguments('item_b', '{"b":20}'), + indexedArguments('item_a', '{"a":10}'), + ]); + expect(result.message.toolCalls).toEqual([ + { + type: 'function', + id: 'call_a', + name: 'add', + arguments: '{"a":10}', + extras: undefined, + }, + { + type: 'function', + id: 'call_b', + name: 'multiply', + arguments: '{"b":20}', + extras: undefined, + }, + ]); + }); + + it.each([ + ['delta', functionArgumentsDelta('missing_item', '{}')], + ['done', functionArgumentsDone('missing_item', '{}')], + ])('rejects %s arguments for an unknown stream index', async (_label, event) => { + const stream = new OpenAIResponsesStreamedMessage(makeAsyncIterable([event]), true); + await expect(collectStreamParts(stream)).rejects.toThrow(/unknown stream index missing_item/); + }); + + it('buffers an unindexed call as a whole call across intervening text and reasoning', async () => { + const events = [ + functionCallAdded(undefined, 'call_unindexed', 'add'), + functionArgumentsDelta(undefined, '{"draft":1}'), + { type: 'response.output_text.delta', delta: 'between' }, + { type: 'response.reasoning_summary_text.delta', delta: 'thinking' }, + functionCallOutputDone(undefined, '{"final":2}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + { type: 'text', text: 'between' }, + { type: 'think', think: 'thinking' }, + { + type: 'function', + id: 'call_unindexed', + name: 'add', + arguments: '{"final":2}', + }, + ]); + }); + + it.each([ + ['another unindexed', undefined], + ['an indexed', 'item_later'], + ])('rejects %s function-call header while an unindexed call is unresolved', async (_label, id) => { + const events = [ + functionCallAdded(undefined, 'call_unindexed', 'add'), + { type: 'response.output_text.delta', delta: 'between' }, + functionCallAdded(id, 'call_later', 'multiply'), + ]; + const { parts, error } = await collectPartsAndError( + new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true), + ); + + expect(parts).toEqual([{ type: 'text', text: 'between' }]); + expect(error).toBeInstanceOf(ChatProviderError); + expect((error as Error).message).toMatch(/unindexed function call.*unresolved/i); + }); + + it('allows a sequential unindexed call after the prior call commits', async () => { + const events = [ + functionCallAdded(undefined, 'call_first', 'add'), + functionCallOutputDone(undefined, '{"first":1}'), + functionCallAdded(undefined, 'call_second', 'multiply'), + functionCallOutputDone(undefined, '{"second":2}'), + ]; + + await expect( + collectStreamParts(new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true)), + ).resolves.toEqual([ + { type: 'function', id: 'call_first', name: 'add', arguments: '{"first":1}' }, + { + type: 'function', + id: 'call_second', + name: 'multiply', + arguments: '{"second":2}', + }, + ]); + }); + + it.each([ + [ + 'error', + { type: 'error', code: 'server_error', message: 'upstream failed', param: null }, + ], + [ + 'response.failed', + { + type: 'response.failed', + response: { + id: 'resp_failed', + status: 'failed', + error: { code: 'server_error', message: 'upstream failed' }, + }, + }, + ], + ['decode failure', { type: 42 }], + ])('%s does not flush buffered arguments', async (_label, failureEvent) => { + const events = [ + functionCallAdded('item_failure', 'call_failure', 'add'), + functionArgumentsDelta('item_failure', '{"partial":'), + failureEvent, + ]; + const { parts, error } = await collectPartsAndError( + new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true), + ); + + expect(parts).toEqual([indexedToolCall('item_failure', 'call_failure', 'add')]); + expect(error).toBeInstanceOf(Error); + }); + + it('does not flush buffered arguments when the source iterator throws', async () => { + async function* failingEvents(): AsyncIterable> { + yield functionCallAdded('item_throw', 'call_throw', 'add'); + yield functionArgumentsDelta('item_throw', '{"partial":'); + throw new Error('iterator boom'); + } + const { parts, error } = await collectPartsAndError( + new OpenAIResponsesStreamedMessage(failingEvents(), true), + ); + + expect(parts).toEqual([indexedToolCall('item_throw', 'call_throw', 'add')]); + expect((error as Error).message).toContain('iterator boom'); + }); + + it('does not flush buffered arguments when the consumer returns early', async () => { + const stream = new OpenAIResponsesStreamedMessage( + makeAsyncIterable([ + functionCallAdded('item_cancel', 'call_cancel', 'add', '{"initial":true}'), + functionArgumentsDelta('item_cancel', '{"partial":'), + ]), + true, + ); + const iterator = stream[Symbol.asyncIterator](); + + expect(await iterator.next()).toEqual({ + done: false, + value: indexedToolCall('item_cancel', 'call_cancel', 'add'), + }); + expect(await iterator.return?.()).toEqual({ done: true, value: undefined }); + }); +}); + async function collectStreamParts( stream: OpenAIResponsesStreamedMessage, ): Promise { @@ -2124,3 +2427,117 @@ function makeAsyncIterable( }, }; } + +function functionCallAdded( + itemId: string | undefined, + callId: string, + name: string, + argumentsValue: string = '', +): Record { + return { + type: 'response.output_item.added', + item: { + id: itemId, + type: 'function_call', + call_id: callId, + name, + arguments: argumentsValue, + }, + }; +} + +function functionArgumentsDelta( + itemId: string | undefined, + delta: string, +): Record { + return { type: 'response.function_call_arguments.delta', item_id: itemId, delta }; +} + +function functionArgumentsDone( + itemId: string | undefined, + argumentsValue: string, +): Record { + return { + type: 'response.function_call_arguments.done', + item_id: itemId, + arguments: argumentsValue, + }; +} + +function functionCallOutputDone( + itemId: string | undefined, + argumentsValue: string, +): Record { + return { + type: 'response.output_item.done', + item: { id: itemId, type: 'function_call', arguments: argumentsValue }, + }; +} + +function indexedToolCall(index: string, id: string, name: string): ToolCall { + return { type: 'function', id, name, arguments: null, _streamIndex: index }; +} + +function indexedArguments(index: string, argumentsPart: string): StreamedMessagePart { + return { type: 'tool_call_part', argumentsPart, index }; +} + +function streamingProvider(events: Record[]): OpenAIResponsesChatProvider { + const provider = createProvider(); + ((provider as any)._client.responses as unknown as Record)['create'] = vi + .fn() + .mockResolvedValue(makeAsyncIterable(events)); + return provider; +} + +async function collectPartsAndError( + stream: OpenAIResponsesStreamedMessage, +): Promise<{ parts: StreamedMessagePart[]; error: unknown }> { + const parts: StreamedMessagePart[] = []; + let caughtError: unknown; + try { + for await (const part of stream) parts.push(part); + } catch (error) { + caughtError = error; + } + return { parts, error: caughtError }; +} + +function controlledEOFIterable(events: Record[]): { + iterable: AsyncIterable>; + waitingForEOF: Promise; + finish: () => void; +} { + let index = 0; + let finish: () => void = () => undefined; + let notifyWaiting: () => void = () => undefined; + const waitingForEOF = new Promise((resolve) => { + notifyWaiting = resolve; + }); + const eof = new Promise>>((resolve) => { + finish = () => { + resolve({ + value: undefined as unknown as Record, + done: true, + }); + }; + }); + + return { + iterable: { + [Symbol.asyncIterator](): AsyncIterator> { + return { + next(): Promise>> { + if (index < events.length) { + return Promise.resolve({ value: events[index++]!, done: false }); + } + notifyWaiting(); + return eof; + }, + }; + }, + }, + waitingForEOF, + finish, + }; +}