diff --git a/packages/ai/src/protocols/open-responses-continuation.ts b/packages/ai/src/protocols/open-responses-continuation.ts index 465dc0721c2..0d4284e745b 100644 --- a/packages/ai/src/protocols/open-responses-continuation.ts +++ b/packages/ai/src/protocols/open-responses-continuation.ts @@ -154,7 +154,12 @@ export const driver = (input: DriverInput): WebSocketChannelDriver => { ...observation, checkpoint: { protocol: PROTOCOL, - value: { version: VERSION, responseID, request, output: output.slice() } satisfies CheckpointValue, + value: { + version: VERSION, + responseID, + request, + output: event.response?.output ? [...event.response.output] : output.slice(), + } satisfies CheckpointValue, }, } }), diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 052f869fe54..71ca8c39277 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -296,6 +296,7 @@ export const Event = Schema.StructWithRest( id: Schema.optional(Schema.String), service_tier: optionalNull(Schema.String), incomplete_details: optionalNull(Schema.Struct({ reason: Schema.optional(Schema.String) })), + output: Schema.optional(Schema.Array(StreamItem)), usage: optionalNull(OpenResponsesUsage), error: optionalNull(OpenResponsesErrorPayload), }), @@ -1111,30 +1112,45 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( }) const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (state: ParserState, event: Event) { + const reconciled = + event.type === "response.completed" + ? yield* Effect.reduce( + event.response?.output ?? [], + () => [state, NO_EVENTS] satisfies StepResult, + ([current, events], item) => { + if (item.type !== "function_call" || !item.id || !current.tools[item.id]) + return Effect.succeed([current, events] satisfies StepResult) + return onOutputItemDone(current, { type: "response.output_item.done", item }).pipe( + Effect.map(([next, emitted]) => [next, [...events, ...emitted]] satisfies StepResult), + ) + }, + ) + : ([state, NO_EVENTS] satisfies StepResult) + const current = reconciled[0] // Some compatible providers omit output_item.done even after completing the response. const pending = event.type === "response.completed" - ? yield* ToolStream.finishAll(state.id, state.tools) - : { tools: state.tools, events: NO_EVENTS } - const events: LLMEvent[] = [...pending.events] + ? yield* ToolStream.finishAll(current.id, current.tools) + : { tools: current.tools, events: NO_EVENTS } + const events: LLMEvent[] = [...reconciled[1], ...pending.events] const hasFunctionCall = pending.events.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) || - state.hasFunctionCall - const lifecycle = Lifecycle.finish(state.lifecycle, events, { + current.hasFunctionCall + const lifecycle = Lifecycle.finish(current.lifecycle, events, { reason: { normalized: mapFinishReason(event, hasFunctionCall), raw: event.response?.incomplete_details?.reason, }, - usage: mapUsage(event.response?.usage, state.providerMetadataKey), + usage: mapUsage(event.response?.usage, current.providerMetadataKey), providerMetadata: event.response?.id || event.response?.service_tier - ? providerMetadata(state, { + ? providerMetadata(current, { responseId: event.response.id, serviceTier: event.response.service_tier, }) : undefined, }) - return [{ ...state, lifecycle, hasFunctionCall, tools: pending.tools }, events] satisfies StepResult + return [{ ...current, lifecycle, hasFunctionCall, tools: pending.tools }, events] satisfies StepResult }) // Build the prettiest summary available from whatever the provider supplied. diff --git a/packages/ai/test/provider/openai-compatible-responses.test.ts b/packages/ai/test/provider/openai-compatible-responses.test.ts index 583043e506f..18b1db7e60f 100644 --- a/packages/ai/test/provider/openai-compatible-responses.test.ts +++ b/packages/ai/test/provider/openai-compatible-responses.test.ts @@ -225,6 +225,48 @@ describe("Open Responses-compatible route", () => { }), ) + it.effect("finalizes pending function calls from completed response output", () => + Effect.gen(function* () { + const model = configure({ + apiKey: "test-key", + baseURL: "https://responses.example.test/v1", + provider: "example", + }).model("example-model") + const response = yield* LLMClient.generate(LLM.request({ model, prompt: "Look it up." })).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { + type: "response.output_item.added", + item: { type: "function_call", id: "item_1", call_id: "call_1", name: "lookup", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", item_id: "item_1", delta: '{"query":"par' }, + { + type: "response.completed", + response: { + output: [ + { + type: "function_call", + id: "item_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"complete"}', + }, + ], + }, + }, + ), + ), + ), + ) + + expect(response.events.find(LLMEvent.is.toolCall)).toMatchObject({ + input: { query: "complete" }, + providerMetadata: { openresponses: { itemId: "item_1" } }, + }) + }), + ) + it.effect("reconciles raw reasoning finals without streamed deltas", () => Effect.gen(function* () { const model = configure({ diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index 23e610650ba..a456a833420 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -518,6 +518,56 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("continues a tool call from authoritative completed response output", () => + Effect.gen(function* () { + const firstRequest = { + type: "response.create", + model: "gpt-5.2", + store: false, + input: [{ role: "user", content: [{ type: "input_text", text: "Weather?" }] }], + } + const first = continuationDriver(firstRequest) + const firstCreate = yield* first.create(undefined) + const saved = checkpoint( + yield* first.observe( + firstCreate, + ProviderShared.encodeJson({ + type: "response.completed", + response: { + id: "resp_1", + output: [ + { + type: "function_call", + id: "fc_1", + status: "completed", + call_id: "call_1", + name: "weather", + arguments: '{ "city": "Paris" }', + }, + ], + }, + }), + ), + ) + const second = continuationDriver({ + ...firstRequest, + input: [ + ...firstRequest.input, + { type: "function_call", call_id: "call_1", name: "weather", arguments: '{"city":"Paris"}' }, + { type: "function_call_output", call_id: "call_1", output: '{"temperature":22}' }, + ], + }) + + const create = yield* second.create(saved) + + expect(create.mode).toBe("incremental") + expect(ProviderShared.decodeJson(create.message)).toMatchObject({ + previous_response_id: "resp_1", + input: [{ type: "function_call_output", call_id: "call_1", output: '{"temperature":22}' }], + }) + }), + ) + it.effect("continues a promoted steer after the completed assistant output", () => Effect.gen(function* () { const firstInput = [{ role: "user", content: [{ type: "input_text", text: "First" }] }] @@ -3051,6 +3101,163 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("treats empty completed output item arguments as authoritative", () => + Effect.gen(function* () { + const body = sseEvents( + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", item_id: "fc_item_1", delta: '{"query":"streamed"}' }, + { + type: "response.output_item.done", + item: { type: "function_call", id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: "" }, + }, + { type: "response.completed", response: { id: "resp_1" } }, + ) + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.find(LLMEvent.is.toolCall)).toMatchObject({ input: {} }) + expect(response.events.filter(LLMEvent.is.toolCall)).toHaveLength(1) + }), + ) + + it.effect("uses completed response output when output item completion is missing", () => + Effect.gen(function* () { + const body = sseEvents( + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", item_id: "fc_item_1", delta: '{"query":"wea' }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [ + { + type: "function_call", + id: "fc_item_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"weather"}', + }, + ], + }, + }, + ) + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.find(LLMEvent.is.toolCall)).toMatchObject({ + id: "call_1", + input: { query: "weather" }, + providerMetadata: { openai: { itemId: "fc_item_1" } }, + }) + expect(response.events.filter(LLMEvent.is.toolInputEnd)).toHaveLength(1) + expect(response.events.filter(LLMEvent.is.toolCall)).toHaveLength(1) + expect(response.finishReason.normalized).toBe("tool-calls") + }), + ) + + it.effect("lets completed response output override arguments done", () => + Effect.gen(function* () { + const body = sseEvents( + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: "" }, + }, + { + type: "response.function_call_arguments.done", + item_id: "fc_item_1", + arguments: '{"query":"arguments-done"}', + }, + { + type: "response.completed", + response: { + output: [ + { + type: "function_call", + id: "fc_item_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"completed"}', + }, + ], + }, + }, + ) + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.find(LLMEvent.is.toolCall)).toMatchObject({ input: { query: "completed" } }) + }), + ) + + it.effect("preserves explicit empty arguments from completed response output", () => + Effect.gen(function* () { + const body = sseEvents( + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", item_id: "fc_item_1", delta: '{"query":"streamed"}' }, + { + type: "response.completed", + response: { + output: [{ type: "function_call", id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: "" }], + }, + }, + ) + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.find(LLMEvent.is.toolCall)).toMatchObject({ input: {} }) + }), + ) + + it.effect("does not repeat function calls already finalized by an output item", () => + Effect.gen(function* () { + const item = { + type: "function_call", + id: "fc_item_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"weather"}', + } + const body = sseEvents( + { type: "response.output_item.added", item: { ...item, arguments: "" } }, + { type: "response.output_item.done", item }, + { type: "response.completed", response: { output: [item] } }, + ) + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.filter(LLMEvent.is.toolInputEnd)).toHaveLength(1) + expect(response.events.filter(LLMEvent.is.toolCall)).toHaveLength(1) + }), + ) + + it.effect("does not finalize pending function calls from incomplete response output", () => + Effect.gen(function* () { + const item = { + type: "function_call", + id: "fc_item_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"partial', + } + const body = sseEvents( + { type: "response.output_item.added", item: { ...item, arguments: "" } }, + { type: "response.function_call_arguments.delta", item_id: "fc_item_1", delta: item.arguments }, + { + type: "response.incomplete", + response: { incomplete_details: { reason: "max_output_tokens" }, output: [item] }, + }, + ) + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.some(LLMEvent.is.toolCall)).toBeFalse() + expect(response.finishReason.normalized).toBe("length") + }), + ) + it.effect("finalizes a pending function call at response completion", () => Effect.gen(function* () { const body = sseEvents(