diff --git a/packages/ai/src/protocols/openai-chat.ts b/packages/ai/src/protocols/openai-chat.ts index a86c52ade35..3993357b95b 100644 --- a/packages/ai/src/protocols/openai-chat.ts +++ b/packages/ai/src/protocols/openai-chat.ts @@ -3,6 +3,7 @@ import { Tool } from "@opencode-ai/schema/tool" import { Route } from "../route/client.js" import { Auth } from "../route/auth.js" import { Endpoint } from "../route/endpoint.js" +import { Framing } from "../route/framing.js" import { HttpTransport } from "../route/transport/index.js" import { Protocol } from "../route/protocol.js" import { @@ -245,6 +246,8 @@ export const OpenAIChatEvent = Schema.StructWithRest( [Schema.Record(Schema.String, Schema.Unknown)], ) export type OpenAIChatEvent = Schema.Schema.Type +const DONE = "[DONE]" as const +const OpenAIChatStreamEvent = Schema.Union([Schema.Literal(DONE), Protocol.jsonEvent(OpenAIChatEvent)]) type OpenAIChatRequestMessage = LLMRequest["messages"][number] interface PendingToolDelta { @@ -1166,7 +1169,7 @@ export const protocol = Protocol.make({ from: fromRequest, }, stream: { - event: Protocol.jsonEvent(OpenAIChatEvent), + event: OpenAIChatStreamEvent, initial: (request) => ({ providerMetadataKey: request.model.route.providerMetadataKey ?? String(request.model.provider), tools: ToolStream.empty(), @@ -1180,12 +1183,14 @@ export const protocol = Protocol.make({ nextToolIndex: 0, requireFinishReason: request.model.compatibility?.requireFinishReason ?? true, }), - step, + step: (state: ParserState, event) => (event === DONE ? Effect.succeed([state, []] as const) : step(state, event)), + terminal: (event) => event === DONE, onHalt: finishEvents, }, }) -export const httpTransport = HttpTransport.sseJson.with() +export const framing = Framing.sseWithDone +export const httpTransport = HttpTransport.sseJson.with().with({ framing }) export const route = Route.make({ id: ADAPTER, diff --git a/packages/ai/src/protocols/openai-compatible-chat.ts b/packages/ai/src/protocols/openai-compatible-chat.ts index d992b3f9edb..a9749f6f759 100644 --- a/packages/ai/src/protocols/openai-compatible-chat.ts +++ b/packages/ai/src/protocols/openai-compatible-chat.ts @@ -1,6 +1,5 @@ import { Route, type RouteRoutedLanguageModelInput } from "../route/client.js" import { Endpoint } from "../route/endpoint.js" -import { Framing } from "../route/framing.js" import * as OpenAIChat from "./openai-chat.js" const ADAPTER = "openai-compatible-chat" @@ -19,7 +18,7 @@ export const route = Route.make({ providerMetadataKey: "openai", protocol: OpenAIChat.protocol, endpoint: Endpoint.path("/chat/completions"), - framing: Framing.sse, + framing: OpenAIChat.framing, }) export * as OpenAICompatibleChat from "./openai-compatible-chat.js" diff --git a/packages/ai/src/protocols/shared.ts b/packages/ai/src/protocols/shared.ts index 219da586943..9ec502de65c 100644 --- a/packages/ai/src/protocols/shared.ts +++ b/packages/ai/src/protocols/shared.ts @@ -207,15 +207,16 @@ export const errorText = (error: unknown) => { /** * `framing` step for Server-Sent Events. Decodes UTF-8, runs the SSE channel - * decoder, optionally filters named events, and drops empty / `[DONE]` - * keep-alive events so the protocol event schema sees one JSON string per - * element. Retry control events are ignored without interrupting the stream. + * decoder, optionally filters named events, and drops empty events. `[DONE]` + * is dropped by default or retained for protocols that use it as their stream + * boundary. Retry control events are ignored without interrupting the stream. * Decoder failures become provider output errors so the public error channel * stays `AIError`. */ export const sseFraming = ( bytes: Stream.Stream, events?: ReadonlySet, + includeDone = false, ): Stream.Stream => bytes.pipe( Stream.decodeText(), @@ -240,7 +241,7 @@ export const sseFraming = ( (event) => (events === undefined || events.has(event.event)) && event.data.length > 0 && - (event.data !== "[DONE]" || (events !== undefined && event.event !== "message")), + (event.data !== "[DONE]" || includeDone || (events !== undefined && event.event !== "message")), ), Stream.map((event) => event.data), ) diff --git a/packages/ai/src/providers/groq.ts b/packages/ai/src/providers/groq.ts index 0ea4fabd7f2..301bb1f49a8 100644 --- a/packages/ai/src/providers/groq.ts +++ b/packages/ai/src/providers/groq.ts @@ -5,7 +5,6 @@ import { ProviderShared } from "../protocols/shared.js" import { AuthOptions, type ProviderAuthOption } from "../route/auth-options.js" import { Route, type RouteDefaultsInput } from "../route/client.js" import { Endpoint } from "../route/endpoint.js" -import { Framing } from "../route/framing.js" import { Protocol } from "../route/protocol.js" import { ProviderID, type ModelID, type LLMRequest } from "../schema/index.js" import { profiles } from "./openai-compatible-profile.js" @@ -75,7 +74,7 @@ export const route = Route.make({ providerMetadataKey: "openai", protocol, endpoint: Endpoint.path("/chat/completions", { baseURL: profiles.groq.baseURL }), - framing: Framing.sse, + framing: OpenAIChat.framing, }) export const configure = (input: LanguageModelOptions = {}) => { diff --git a/packages/ai/src/providers/openrouter.ts b/packages/ai/src/providers/openrouter.ts index aa226b9eba4..debcc3d1357 100644 --- a/packages/ai/src/providers/openrouter.ts +++ b/packages/ai/src/providers/openrouter.ts @@ -1,7 +1,6 @@ import { Effect, Schema } from "effect" import { Route, type RouteDefaultsInput } from "../route/client.js" import { Endpoint } from "../route/endpoint.js" -import { Framing } from "../route/framing.js" import { Protocol } from "../route/protocol.js" import { AuthOptions, type ProviderAuthOption } from "../route/auth-options.js" import { ProviderID, type CacheHint, type ModelID } from "../schema/index.js" @@ -167,7 +166,7 @@ export const route = Route.make({ providerMetadataKey: "openrouter", protocol, endpoint: Endpoint.path("/chat/completions", { baseURL: profile.baseURL }), - framing: Framing.sse, + framing: OpenAIChat.framing, }) export const routes = [route] diff --git a/packages/ai/src/route/framing.ts b/packages/ai/src/route/framing.ts index 720de36374f..8b680be94f4 100644 --- a/packages/ai/src/route/framing.ts +++ b/packages/ai/src/route/framing.ts @@ -8,8 +8,8 @@ import type { AIError } from "../schema/index.js" * `Framing` is the byte-stream-shaped seam between transport and protocol: * * - SSE (`Framing.sse`) — UTF-8 decode the body, run the SSE channel decoder, - * drop empty / `[DONE]` keep-alives. Each emitted frame is the JSON `data:` - * payload of one event. + * and emit the `data:` payload of each non-empty event. The default drops + * `[DONE]`; protocols that use it as a terminal select `sseWithDone`. * - AWS event stream — length-prefixed binary frames with CRC checksums. * Each emitted frame is one parsed binary event record. * @@ -26,6 +26,12 @@ export interface Definition { /** Server-Sent Events framing. Used by every JSON-streaming HTTP provider. */ export const sse: Definition = { id: "sse", frame: ProviderShared.sseFraming } +/** Server-Sent Events framing that retains the conventional `[DONE]` sentinel. */ +export const sseWithDone: Definition = { + id: "sse", + frame: (bytes) => ProviderShared.sseFraming(bytes, undefined, true), +} + /** SSE framing restricted to protocol-recognized event names. */ export const sseEvents = (events: ReadonlySet): Definition => ({ id: "sse", diff --git a/packages/ai/test/provider/openai-chat.test.ts b/packages/ai/test/provider/openai-chat.test.ts index 8e6eca60712..9b2f420ced1 100644 --- a/packages/ai/test/provider/openai-chat.test.ts +++ b/packages/ai/test/provider/openai-chat.test.ts @@ -807,6 +807,28 @@ describe("OpenAI Chat route", () => { }), ) + it.effect("finishes at the done sentinel without waiting for response EOF", () => + Effect.gen(function* () { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode(sseEvents(deltaChunk({ content: "Hello" }), deltaChunk({}, "stop"))), + ) + }, + }) + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse(stream, { + headers: { "content-type": "text/event-stream" }, + }), + ), + ) + + expect(response.text).toBe("Hello") + expect(response.events.at(-1)?.type).toBe("finish") + }), + ) + it.effect("preserves streamed refusals as ordinary assistant text", () => Effect.gen(function* () { const response = yield* LLMClient.generate(request).pipe( diff --git a/packages/ai/test/provider/openai-compatible-chat.test.ts b/packages/ai/test/provider/openai-compatible-chat.test.ts index ab12d415d81..a37739cfeba 100644 --- a/packages/ai/test/provider/openai-compatible-chat.test.ts +++ b/packages/ai/test/provider/openai-compatible-chat.test.ts @@ -405,6 +405,19 @@ describe("OpenAI-compatible Chat route", () => { }), ) + it.effect("ignores events after the done sentinel", () => + Effect.gen(function* () { + const body = `${sseEvents( + deltaChunk({ content: "Hello" }), + deltaChunk({}, "stop"), + )}data: ${JSON.stringify(deltaChunk({ content: " late" }))}\n\n` + const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) + + expect(response.text).toBe("Hello") + expect(response.finishReason).toEqual({ normalized: "stop", raw: "stop" }) + }), + ) + it.effect("accepts nullable usage and preserves provider fields", () => Effect.gen(function* () { const response = yield* LLMClient.generate(request).pipe(