From 41c0ed4fb7d9c10bcad21edf26342930d1e1c005 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Thu, 6 Aug 2026 20:36:28 -0500 Subject: [PATCH] fix(ai): support streams without finish reasons --- packages/ai/src/protocols/openai-chat.ts | 32 +++++-- packages/ai/src/route/client.ts | 55 +++++++---- packages/ai/src/route/protocol.ts | 4 +- packages/ai/test/adapter.test.ts | 23 +++++ packages/ai/test/provider/openai-chat.test.ts | 94 ++++++++++++++++++- packages/core/test/session-runner.test.ts | 24 +++++ 6 files changed, 204 insertions(+), 28 deletions(-) diff --git a/packages/ai/src/protocols/openai-chat.ts b/packages/ai/src/protocols/openai-chat.ts index 26a2d4b19d8d..14521b63e239 100644 --- a/packages/ai/src/protocols/openai-chat.ts +++ b/packages/ai/src/protocols/openai-chat.ts @@ -241,6 +241,8 @@ export interface ParserState { readonly reasoningEmitted: boolean readonly latestToolIndex?: number readonly nextToolIndex: number + readonly outputStarted: boolean + readonly requireFinishReason: boolean } // ============================================================================= @@ -707,9 +709,7 @@ const step = (state: ParserState, event: OpenAIChatEvent) => Boolean(delta?.content) || reasoning !== undefined || (Array.isArray(delta?.reasoning_details) && delta.reasoning_details.length > 0) || - toolDeltas.some( - (tool) => Boolean(tool.id) || Boolean(tool.function?.name) || Boolean(tool.function?.arguments), - ) + toolDeltas.some((tool) => Boolean(tool.id) || Boolean(tool.function?.name) || Boolean(tool.function?.arguments)) if (state.finishReason !== undefined) { if (hasLateContent) return yield* ProviderShared.eventError(ADAPTER, "OpenAI Chat received content after the finish reason") @@ -749,8 +749,7 @@ const step = (state: ParserState, event: OpenAIChatEvent) => const fallback = toolDeltas.length > 1 ? position : (latestToolIndex ?? position) const fallbackTool = tools[fallback] ?? pendingTools[fallback] const index = - tool.index ?? matched ?? - (tool.id && fallbackTool?.id && fallbackTool.id !== tool.id ? nextToolIndex : fallback) + tool.index ?? matched ?? (tool.id && fallbackTool?.id && fallbackTool.id !== tool.id ? nextToolIndex : fallback) const current = tools[index] const pending = pendingTools[index] const id = current?.id ?? pending?.id ?? (tool.id || undefined) @@ -806,6 +805,8 @@ const step = (state: ParserState, event: OpenAIChatEvent) => reasoningEmitted, latestToolIndex, nextToolIndex, + outputStarted: state.outputStarted || hasLateContent, + requireFinishReason: state.requireFinishReason, }, events, ] as const @@ -836,6 +837,23 @@ const finishEvents = (state: ParserState): ReadonlyArray => { return events } +const onHalt = (state: ParserState) => + Effect.gen(function* () { + if (state.finishReason !== undefined || state.requireFinishReason) return finishEvents(state) + if (!state.outputStarted) return [] + if (Object.keys(state.pendingTools).length > 0) + return yield* ProviderShared.eventError(ADAPTER, "OpenAI Chat tool call delta is missing id or name") + // Chat has no per-call stop event, so an accepted EOF must finalize every + // accumulated tool input before publishing the synthetic terminal reason. + const finished = yield* ToolStream.finishAll(ADAPTER, state.tools) + return finishEvents({ + ...state, + tools: finished.tools, + toolCallEvents: finished.events, + finishReason: { normalized: "unknown" }, + }) + }) + // ============================================================================= // Protocol And OpenAI Route // ============================================================================= @@ -863,9 +881,11 @@ export const protocol = Protocol.make({ reasoningDetailsObserved: false, reasoningEmitted: false, nextToolIndex: 0, + outputStarted: false, + requireFinishReason: request.model.compatibility?.requireFinishReason ?? true, }), step, - onHalt: finishEvents, + onHalt, }, }) diff --git a/packages/ai/src/route/client.ts b/packages/ai/src/route/client.ts index 89000c15dbdf..d414c026ccf2 100644 --- a/packages/ai/src/route/client.ts +++ b/packages/ai/src/route/client.ts @@ -7,7 +7,7 @@ import { Framing } from "./framing" import { HttpTransport } from "./transport" import type { HttpMiddleware, Transport, TransportRuntime } from "./transport" import { WebSocketExecutor } from "./transport" -import type { Protocol } from "./protocol" +import type { Protocol, ProtocolStream } from "./protocol" import { applyCachePolicy } from "../cache-policy" import * as ProviderShared from "../protocols/shared" import type { ProtocolID, ProviderOptions } from "../schema" @@ -243,28 +243,56 @@ const incompleteStreamError = (route: string) => }), }) -const requireTerminalEvent = (route: string) => (events: Stream.Stream) => +const ensureTerminalEvent = (route: string, required: boolean) => (events: Stream.Stream) => Stream.suspend(() => { let terminal = false + let output = false + const fallback = Stream.suspend(() => { + if (terminal) return Stream.empty + if (required || !output) return Stream.fail(incompleteStreamError(route)) + // The compatibility override trusts a clean stream end, but it cannot + // recover the provider's omitted reason. + const reason = { normalized: "unknown" as const } + return Stream.make(LLMEvent.stepFinish({ index: 0, reason }), LLMEvent.finish({ reason })) + }) return events.pipe( Stream.mapEffect((event) => { if (terminal) return Effect.fail( ProviderShared.eventError(route, `Provider emitted ${event.type} after the terminal event`), ) + output = true if (LLMEvent.is.finish(event) || LLMEvent.is.providerError(event)) terminal = true return Effect.succeed(event) }), - Stream.onEnd( - Effect.suspend(() => - terminal - ? Effect.void - : Effect.fail(incompleteStreamError(route)), - ), - ), + Stream.concat(fallback), ) }) +type ProtocolEvent = { readonly type: "event"; readonly event: Event } | { readonly type: "halt" } + +const parseProtocolEvents = ( + events: Stream.Stream, + request: LLMRequest, + protocol: { readonly stream: ProtocolStream }, +) => + events.pipe( + Stream.map((event): ProtocolEvent => ({ type: "event", event })), + // A normal halt becomes an in-band parser input so finalization may fail. + Stream.concat(Stream.succeed({ type: "halt" } as const)), + Stream.mapAccumEffect( + () => protocol.stream.initial(request), + (state, event) => { + if (event.type === "event") return protocol.stream.step(state, event.event) + if (!protocol.stream.onHalt) return Effect.succeed([state, []] as const) + const events = protocol.stream.onHalt(state) + return Effect.isEffect(events) + ? events.pipe(Effect.map((events) => [state, events] as const)) + : Effect.succeed([state, events] as const) + }, + ), + ) + function makeFromTransport( input: MakeTransportInput, ): Route { @@ -329,14 +357,9 @@ function makeFromTransport( Stream.mapEffect(decodeEvent(route)), protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream, ) - return events.pipe( - Stream.mapAccumEffect( - () => protocol.stream.initial(request), - protocol.stream.step, - protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined, - ), + return parseProtocolEvents(events, request, protocol).pipe( Stream.catchCause((cause) => Stream.fail(streamError(route, `Failed to read ${route} stream`, cause))), - requireTerminalEvent(route), + ensureTerminalEvent(route, request.model.compatibility?.requireFinishReason ?? true), ) }, } satisfies Route diff --git a/packages/ai/src/route/protocol.ts b/packages/ai/src/route/protocol.ts index 77cc26844358..7fe133c7fe12 100644 --- a/packages/ai/src/route/protocol.ts +++ b/packages/ai/src/route/protocol.ts @@ -59,8 +59,8 @@ export interface ProtocolStream { readonly step: (state: State, event: Event) => Effect.Effect], AIError> /** Optional request-completion signal for transports that do not end naturally. */ readonly terminal?: (event: Event) => boolean - /** Optional flush emitted when the framed stream ends. */ - readonly onHalt?: (state: State) => ReadonlyArray + /** Optional flush emitted when the framed stream ends successfully. */ + readonly onHalt?: (state: State) => ReadonlyArray | Effect.Effect, AIError> } /** diff --git a/packages/ai/test/adapter.test.ts b/packages/ai/test/adapter.test.ts index 3ac7cfbb8dfc..01f41c1d4011 100644 --- a/packages/ai/test/adapter.test.ts +++ b/packages/ai/test/adapter.test.ts @@ -138,6 +138,29 @@ describe("llm route", () => { }), ) + unterminated.effect("synthesizes an unknown finish when a terminal event is not required", () => + Effect.gen(function* () { + const response = yield* (yield* LLMClient.Service).generate( + LLMRequest.update(request, { + model: updateModel(request.model, { compatibility: { requireFinishReason: false } }), + }), + ) + + expect(response.text).toBe("partial") + expect(response.finishReason).toEqual({ normalized: "unknown" }) + expect(response.events.slice(-2)).toEqual([ + { + type: "step-finish", + index: 0, + reason: { normalized: "unknown" }, + usage: undefined, + providerMetadata: undefined, + }, + { type: "finish", reason: { normalized: "unknown" }, usage: undefined }, + ]) + }), + ) + it.effect("selects routes by model route value", () => Effect.gen(function* () { const prepared = yield* compileRequest( diff --git a/packages/ai/test/provider/openai-chat.test.ts b/packages/ai/test/provider/openai-chat.test.ts index f9f226f75b38..fd21ff5ddd5e 100644 --- a/packages/ai/test/provider/openai-chat.test.ts +++ b/packages/ai/test/provider/openai-chat.test.ts @@ -40,6 +40,10 @@ const request = LLM.request({ generation: { maxTokens: 20, temperature: 0 }, }) +const optionalFinishRequest = LLMRequest.update(request, { + model: LanguageModel.update(model, { compatibility: { requireFinishReason: false } }), +}) + describe("OpenAI Chat route", () => { it.effect("prepares OpenAI Chat payload", () => Effect.gen(function* () { @@ -596,6 +600,20 @@ describe("OpenAI Chat route", () => { }), ) + it.effect("accepts text and usage without a finish reason when configured", () => + Effect.gen(function* () { + const body = sseEvents( + deltaChunk({ role: "assistant", content: "Hello" }), + usageChunk({ prompt_tokens: 5, completion_tokens: 1, total_tokens: 6 }), + ) + const response = yield* LLMClient.generate(optionalFinishRequest).pipe(Effect.provide(fixedResponse(body))) + + expect(response.text).toBe("Hello") + expect(response.finishReason).toEqual({ normalized: "unknown" }) + expect(response.usage).toMatchObject({ inputTokens: 5, outputTokens: 1, totalTokens: 6 }) + }), + ) + it.effect("parses and replays OpenAI-compatible reasoning fields", () => Effect.gen(function* () { const fields = ["reasoning_content", "reasoning", "reasoning_text"] as const @@ -1145,21 +1163,89 @@ describe("OpenAI Chat route", () => { }), ) - it.effect("fails on malformed stream events", () => + it.effect("finalizes a streamed tool call without a finish reason when configured", () => + Effect.gen(function* () { + const body = sseEvents( + deltaChunk({ + role: "assistant", + tool_calls: [{ index: 0, id: "call_1", function: { name: "lookup", arguments: '{"query"' } }], + }), + deltaChunk({ tool_calls: [{ index: 0, function: { arguments: ':"weather"}' } }] }), + ) + const response = yield* LLMClient.generate( + LLMRequest.update(optionalFinishRequest, { + tools: [ToolDefinition.make({ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } })], + }), + ).pipe(Effect.provide(fixedResponse(body))) + + expect(response.toolCalls).toMatchObject([{ id: "call_1", name: "lookup", input: { query: "weather" } }]) + expect(response.finishReason).toEqual({ normalized: "unknown" }) + }), + ) + + it.effect("settles malformed tool input without a finish reason when configured", () => + Effect.gen(function* () { + const body = sseEvents( + deltaChunk({ + role: "assistant", + tool_calls: [{ index: 0, id: "call_1", function: { name: "lookup", arguments: '{"query"' } }], + }), + ) + const response = yield* LLMClient.generate( + LLMRequest.update(optionalFinishRequest, { + tools: [ToolDefinition.make({ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } })], + }), + ).pipe(Effect.provide(fixedResponse(body))) + + expect(response.events.filter(LLMEvent.is.toolInputError)).toMatchObject([ + { id: "call_1", name: "lookup", raw: '{"query"' }, + ]) + expect(response.toolCalls).toEqual([]) + expect(response.finishReason).toEqual({ normalized: "unknown" }) + }), + ) + + it.effect("rejects incomplete tool identity without a finish reason when configured", () => + Effect.gen(function* () { + const body = sseEvents(deltaChunk({ tool_calls: [{ index: 0, id: "call_1", function: { arguments: "{}" } }] })) + const error = yield* LLMClient.generate(optionalFinishRequest).pipe( + Effect.provide(fixedResponse(body)), + Effect.flip, + ) + + expect(error.message).toContain("OpenAI Chat tool call delta is missing id or name") + }), + ) + + it.effect("rejects an empty stream when a finish reason is not required", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate(optionalFinishRequest).pipe( + Effect.provide(fixedResponse(sseEvents())), + Effect.flip, + ) + + expect(error.reason).toMatchObject({ _tag: "InvalidProviderOutput", classification: "incomplete-stream" }) + }), + ) + + it.effect("fails on malformed stream events when a finish reason is not required", () => Effect.gen(function* () { const body = sseEvents(deltaChunk({ content: 123 })) - const error = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body)), Effect.flip) + const error = yield* LLMClient.generate(optionalFinishRequest).pipe( + Effect.provide(fixedResponse(body)), + Effect.flip, + ) expect(error.message).toContain("Invalid openai/openai-chat stream event") }), ) - it.effect("surfaces transport errors that occur mid-stream", () => + it.effect("surfaces transport errors when a finish reason is not required", () => Effect.gen(function* () { const layer = truncatedStream([ `data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`, ]) - const error = yield* LLMClient.generate(request).pipe(Effect.provide(layer), Effect.flip) + const error = yield* LLMClient.generate(optionalFinishRequest).pipe(Effect.provide(layer), Effect.flip) expect(error.message).toContain("Failed to read openai/openai-chat stream") }), diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 72d15f16b181..445a7b95bde9 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -2440,6 +2440,30 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("continues after an unknown finish containing a local tool call", () => + Effect.gen(function* () { + const session = yield* setup + yield* admit(session, "Echo this") + yield* TestLLM.push( + TestLLM.complete( + { reason: { normalized: "unknown" } }, + LLMEvent.toolCall({ id: "call-echo", name: "echo", input: { text: "hello" } }), + ), + TestLLM.text("Done", "text-final"), + ) + + yield* session.resume(sessionID) + + expect(requests).toHaveLength(2) + expect(executions).toEqual(["hello"]) + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user", text: "Echo this" }, + { type: "assistant", finish: "unknown", content: [{ type: "tool", state: { status: "completed" } }] }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: "Done" }] }, + ]) + }), + ) + it.effect("reloads a model switch before a tool-driven continuation step", () => Effect.gen(function* () { const session = yield* setup