diff --git a/packages/llm/src/protocols/openai-responses.ts b/packages/llm/src/protocols/openai-responses.ts index 4936d31c921b..90802407e6c6 100644 --- a/packages/llm/src/protocols/openai-responses.ts +++ b/packages/llm/src/protocols/openai-responses.ts @@ -221,6 +221,7 @@ const OpenAIResponsesEvent = Schema.Struct({ id: Schema.optional(Schema.String), service_tier: optionalNull(Schema.String), incomplete_details: optionalNull(Schema.Struct({ reason: Schema.String })), + truncated: Schema.optional(Schema.Boolean), usage: optionalNull(OpenAIResponsesUsage), error: optionalNull(OpenAIResponsesErrorPayload), }), @@ -239,6 +240,20 @@ interface ParserState { readonly lifecycle: Lifecycle.State readonly reasoningItems: Readonly> readonly store: boolean | undefined + readonly unparsedTools: ReadonlyArray +} + +/** + * A function-call item whose streamed JSON arguments never parsed into a + * complete value. The tool is never dispatched; the entry is kept so the + * terminal response event can classify whether the provider truncated the + * stream or shipped malformed JSON for a finished call. Only the id, name, + * and UTF-8 byte count are retained — never the argument contents. + */ +interface UnparsedTool { + readonly id: string + readonly name: string + readonly bytes: number } type ReasoningSummaryStatus = "active" | "can-conclude" | "concluded" @@ -817,12 +832,19 @@ const onOutputItemDone = Effect.fn("OpenAIResponses.onOutputItemDone")(function* const tools = state.tools[item.id] ? state.tools : ToolStream.start(state.tools, item.id, { id: item.call_id, name: item.name }) - const result = - item.arguments === undefined - ? yield* ToolStream.finish(ADAPTER, tools, item.id) - : yield* ToolStream.finishWithInput(ADAPTER, tools, item.id, item.arguments) + // A malformed argument buffer is not fatal by itself: truncation is only + // diagnosable once the terminal response event arrives (`truncated` / + // `max_output_tokens`), so record the call and settle it there instead of + // failing the whole stream mid-turn. + const settled = yield* (item.arguments === undefined + ? ToolStream.finish(ADAPTER, tools, item.id) + : ToolStream.finishWithInput(ADAPTER, tools, item.id, item.arguments) + ).pipe( + Effect.catchTag("LLM.Error", () => Effect.succeed(undefined)), + ) + if (settled === undefined) return unparsedToolStep(state, tools, item) const events: LLMEvent[] = [] - const resultEvents = result.events ?? [] + const resultEvents = settled.events ?? [] const lifecycle = resultEvents.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle events.push(...resultEvents) return [ @@ -830,7 +852,7 @@ const onOutputItemDone = Effect.fn("OpenAIResponses.onOutputItemDone")(function* ...state, lifecycle, hasFunctionCall: resultEvents.some(LLMEvent.is.toolCall) ? true : state.hasFunctionCall, - tools: result.tools, + tools: settled.tools, }, events, ] satisfies StepResult @@ -872,7 +894,40 @@ const onOutputItemDone = Effect.fn("OpenAIResponses.onOutputItemDone")(function* return [state, NO_EVENTS] satisfies StepResult }) +// A tool call whose arguments failed to parse: emit the input-end event so +// the lifecycle stays coherent, drop the pending accumulator (the tool is +// never dispatched), and defer the failure to the terminal response event. +const unparsedToolStep = ( + state: ParserState, + tools: ToolStream.State, + item: OpenAIResponsesStreamItem, +): StepResult => { + const tool = tools[item.id ?? ""] + const { [item.id ?? ""]: _removed, ...nextTools } = tools + const id = item.call_id ?? item.id ?? "" + const name = item.name ?? "" + const events: LLMEvent[] = [] + const lifecycle = Lifecycle.stepStart(state.lifecycle, events) + events.push(LLMEvent.toolInputEnd({ id, name, providerMetadata: tool?.providerMetadata })) + return [ + { + ...state, + lifecycle, + tools: nextTools, + unparsedTools: [ + ...state.unparsedTools, + { id, name, bytes: Buffer.byteLength(item.arguments ?? tool?.input ?? "", "utf8") }, + ], + }, + events, + ] +} + const onResponseFinish = (state: ParserState, event: OpenAIResponsesEvent): StepResult => { + if (state.unparsedTools.length > 0) { + const events = state.unparsedTools.map((tool) => unparsedToolError(event, tool)) + return [{ ...state, unparsedTools: [] }, events] + } const events: LLMEvent[] = [] const lifecycle = Lifecycle.finish(state.lifecycle, events, { reason: mapFinishReason(event, state.hasFunctionCall), @@ -888,6 +943,28 @@ const onResponseFinish = (state: ParserState, event: OpenAIResponsesEvent): Step return [{ ...state, lifecycle }, events] } +// A clean terminal event alone cannot distinguish truncation from a provider +// that finalized a malformed buffer: `response.completed` with `truncated`, +// and `response.incomplete` with `incomplete_details.reason` of +// `max_output_tokens`, mean the model ran out of budget mid-argument. +const isTruncatedResponse = (event: OpenAIResponsesEvent): boolean => + event.response?.truncated === true || + event.type === "response.incomplete" || + event.response?.incomplete_details?.reason === "max_output_tokens" + +// Surface an unparsed tool call as a `provider-error` carrying the terminal +// event, the truncation classification, and the argument byte count — without +// exposing the argument contents themselves. +const unparsedToolError = (event: OpenAIResponsesEvent, tool: UnparsedTool) => { + const truncated = isTruncatedResponse(event) + const message = truncated + ? `OpenAI Responses truncated tool call ${tool.name}: ${tool.bytes} argument bytes streamed before JSON completed (${event.type})` + : `Invalid JSON input for ${ADAPTER} tool call ${tool.name} (${tool.bytes} bytes streamed, ${event.type})` + return truncated + ? LLMEvent.providerError({ message, classification: "truncated" }) + : LLMEvent.providerError({ message }) +} + // Build a single human-readable message from whatever the provider supplied. // When both code and message are present, prefix the code so consumers see // the failure mode (e.g. `rate_limit_exceeded: Slow down`) instead of just @@ -911,15 +988,32 @@ const providerError = (event: OpenAIResponsesEvent, fallback: string) => { } const onResponseFailed = (state: ParserState, event: OpenAIResponsesEvent): StepResult => [ - state, - [providerError(event, "OpenAI Responses response failed")], + { ...state, unparsedTools: [] }, + [ + providerError(event, "OpenAI Responses response failed"), + ...state.unparsedTools.map((tool) => unparsedToolError(event, tool)), + ], ] const onError = (state: ParserState, event: OpenAIResponsesEvent): StepResult => [ - state, - [providerError(event, "OpenAI Responses stream error")], + { ...state, unparsedTools: [] }, + [ + providerError(event, "OpenAI Responses stream error"), + ...state.unparsedTools.map((tool) => unparsedToolError(event, tool)), + ], ] +// The stream ended without a terminal response event (EOF, dropped +// connection, server-side close). Any pending unparsed tool calls are +// reported here so the boundary is localizable; the terminal-event handlers +// above clear `unparsedTools`, so a normal completion never reaches this path. +const onHalt = (state: ParserState): ReadonlyArray => + state.unparsedTools.map((tool) => + LLMEvent.providerError({ + message: `OpenAI Responses stream ended with incomplete tool call ${tool.name}: ${tool.bytes} argument bytes before a terminal response event`, + }), + ) + const step = (state: ParserState, event: OpenAIResponsesEvent) => { if (event.type === "response.output_text.delta") return Effect.succeed(onOutputTextDelta(state, event)) if ( @@ -970,9 +1064,11 @@ export const protocol = Protocol.make({ lifecycle: Lifecycle.initial(), reasoningItems: {}, store: OpenAIOptions.store(request), + unparsedTools: [], }), step, terminal: (event) => TERMINAL_TYPES.has(event.type), + onHalt, }, }) diff --git a/packages/llm/src/schema/errors.ts b/packages/llm/src/schema/errors.ts index 072e4e838928..1fb5e2226204 100644 --- a/packages/llm/src/schema/errors.ts +++ b/packages/llm/src/schema/errors.ts @@ -1,7 +1,7 @@ import { Schema } from "effect" import { ModelID, ProviderID, ProviderMetadata, RouteID } from "./ids" -export const ProviderFailureClassification = Schema.Literal("context-overflow") +export const ProviderFailureClassification = Schema.Literals(["context-overflow", "truncated"]) export type ProviderFailureClassification = typeof ProviderFailureClassification.Type export class HttpRequestDetails extends Schema.Class("LLM.HttpRequestDetails")({ diff --git a/packages/llm/test/provider/openai-responses.test.ts b/packages/llm/test/provider/openai-responses.test.ts index cd8bad51af47..76c192adcef9 100644 --- a/packages/llm/test/provider/openai-responses.test.ts +++ b/packages/llm/test/provider/openai-responses.test.ts @@ -1,7 +1,7 @@ import { describe, expect } from "bun:test" import { ConfigProvider, Effect, Layer, Stream } from "effect" import { Headers, HttpClientRequest } from "effect/unstable/http" -import { LLM, LLMError, Message, Model, ToolCallPart, Usage } from "../../src" +import { LLM, LLMEvent, LLMError, Message, Model, ToolCallPart, Usage } from "../../src" import { Auth, LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route" import * as Azure from "../../src/providers/azure" import * as OpenAI from "../../src/providers/openai" @@ -1220,6 +1220,132 @@ describe("OpenAI Responses route", () => { }), ) + describe("truncated tool arguments", () => { + const truncatedRequest = LLM.updateRequest(request, { + tools: [{ name: "read", description: "Read a file", inputSchema: { type: "object" } }], + }) + const incompleteDelta = '{"path":"docs/open-code' + const deltaBytes = Buffer.byteLength(incompleteDelta, "utf8") + const truncatedToolEvents = () => + [ + { + type: "response.output_item.added", + item: { type: "function_call", id: "item_1", call_id: "call_1", name: "read", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", item_id: "item_1", delta: incompleteDelta }, + { + type: "response.output_item.done", + item: { type: "function_call", id: "item_1", call_id: "call_1", name: "read" }, + }, + ] as const + + // Regression: a tool call whose streamed arguments end mid-JSON must not + // fail the whole stream at `output_item.done`. The truncation is only + // knowable from the terminal response event, so the parser defers the + // diagnosis until `response.completed` and classifies it there. + it.effect("classifies truncation on response.completed with truncated flag", () => + Effect.gen(function* () { + const body = sseEvents( + ...truncatedToolEvents(), + { type: "response.completed", response: { truncated: true } }, + ) + const response = yield* LLMClient.generate(truncatedRequest).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events).toEqual([ + { type: "step-start", index: 0 }, + { + type: "tool-input-start", + id: "call_1", + name: "read", + providerMetadata: { openai: { itemId: "item_1" } }, + }, + { type: "tool-input-delta", id: "call_1", name: "read", text: incompleteDelta }, + { + type: "tool-input-end", + id: "call_1", + name: "read", + providerMetadata: { openai: { itemId: "item_1" } }, + }, + { + type: "provider-error", + message: `OpenAI Responses truncated tool call read: ${deltaBytes} argument bytes streamed before JSON completed (response.completed)`, + classification: "truncated", + }, + ]) + expect(response.finishReason).toBe("error") + expect(response.events.some(LLMEvent.is.toolCall)).toBe(false) + }), + ) + + it.effect("classifies truncation on response.incomplete max_output_tokens", () => + Effect.gen(function* () { + const body = sseEvents( + ...truncatedToolEvents(), + { type: "response.incomplete", response: { incomplete_details: { reason: "max_output_tokens" } } }, + ) + const response = yield* LLMClient.generate(truncatedRequest).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.filter(LLMEvent.is.providerError)).toEqual([ + { + type: "provider-error", + message: `OpenAI Responses truncated tool call read: ${deltaBytes} argument bytes streamed before JSON completed (response.incomplete)`, + classification: "truncated", + }, + ]) + expect(response.finishReason).toBe("error") + expect(response.events.some(LLMEvent.is.toolCall)).toBe(false) + }), + ) + + it.effect("reports malformed arguments when a completed response did not truncate", () => + Effect.gen(function* () { + const body = sseEvents(...truncatedToolEvents(), { type: "response.completed", response: {} }) + const response = yield* LLMClient.generate(truncatedRequest).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.filter(LLMEvent.is.providerError)).toEqual([ + { + type: "provider-error", + message: `Invalid JSON input for openai-responses tool call read (${deltaBytes} bytes streamed, response.completed)`, + }, + ]) + expect(response.finishReason).toBe("error") + }), + ) + + it.effect("appends unparsed tool diagnostics to response.failed", () => + Effect.gen(function* () { + const body = sseEvents( + ...truncatedToolEvents(), + { type: "response.failed", response: { error: { code: "server_error", message: "upstream" } } }, + ) + const response = yield* LLMClient.generate(truncatedRequest).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.filter(LLMEvent.is.providerError)).toEqual([ + { type: "provider-error", message: "server_error: upstream" }, + { + type: "provider-error", + message: `Invalid JSON input for openai-responses tool call read (${deltaBytes} bytes streamed, response.failed)`, + }, + ]) + }), + ) + + it.effect("reports unparsed tools when the stream ends without a terminal event", () => + Effect.gen(function* () { + const body = sseEvents(...truncatedToolEvents()) + const response = yield* LLMClient.generate(truncatedRequest).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.filter(LLMEvent.is.providerError)).toEqual([ + { + type: "provider-error", + message: `OpenAI Responses stream ended with incomplete tool call read: ${deltaBytes} argument bytes before a terminal response event`, + }, + ]) + expect(response.finishReason).toBe("error") + }), + ) + }) + it.effect("decodes web_search_call as provider-executed tool-call + tool-result", () => Effect.gen(function* () { const item = {