fix(ai): finish chat streams at done sentinel (#45850)

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<typeof OpenAIChatEvent>
+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<number>(),
@@ -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<OpenAIChatBody>()
+export const framing = Framing.sseWithDone
+export const httpTransport = HttpTransport.sseJson.with<OpenAIChatBody>().with({ framing })
 
 export const route = Route.make({
   id: ADAPTER,

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"

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<Uint8Array, AIError>,
   events?: ReadonlySet<string>,
+  includeDone = false,
 ): Stream.Stream<string, AIError> =>
   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),
   )

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 = {}) => {

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]

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<Frame> {
 /** Server-Sent Events framing. Used by every JSON-streaming HTTP provider. */
 export const sse: Definition<string> = { id: "sse", frame: ProviderShared.sseFraming }
 
+/** Server-Sent Events framing that retains the conventional `[DONE]` sentinel. */
+export const sseWithDone: Definition<string> = {
+  id: "sse",
+  frame: (bytes) => ProviderShared.sseFraming(bytes, undefined, true),
+}
+
 /** SSE framing restricted to protocol-recognized event names. */
 export const sseEvents = (events: ReadonlySet<string>): Definition<string> => ({
   id: "sse",

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<Uint8Array>({
+        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(

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(