From 8f3d1170c470fa3893a2652075901b5a8c3fab53 Mon Sep 17 00:00:00 2001 From: kvnloo Date: Sun, 27 Sep 2026 17:04:31 +0000 Subject: [PATCH] fix(opencode): ignore non-object SSE keep-alive frames Drop empty/[DONE]/non-object JSON keep-alive SSE payloads so decodeChunk sees one object event per element (gateways that inject data: null). Origin tip: anomalyco/opencode#50508 Origin SHA: 7dd74c0601fd Fork-only tip leaf; github_writes_origin=0. --- packages/llm/src/protocols/shared.ts | 33 ++++++++++++++--- .../provider/openai-compatible-chat.test.ts | 37 ++++++++++++++++++- 2 files changed, 63 insertions(+), 7 deletions(-) diff --git a/packages/llm/src/protocols/shared.ts b/packages/llm/src/protocols/shared.ts index 173dc511bb03..9bdc999fc37e 100644 --- a/packages/llm/src/protocols/shared.ts +++ b/packages/llm/src/protocols/shared.ts @@ -1,5 +1,6 @@ import { Buffer } from "node:buffer" import { Effect, Schema, Stream } from "effect" +import { isSome } from "effect/Option" import * as Sse from "effect/unstable/encoding/Sse" import { Headers, HttpClientRequest } from "effect/unstable/http" import { @@ -18,6 +19,7 @@ export { isRecord } export const Json = Schema.fromJsonString(Schema.Unknown) export const decodeJson = Schema.decodeUnknownSync(Json) +export const decodeJsonOption = Schema.decodeUnknownOption(Json) export const encodeJson = Schema.encodeSync(Json) const isJson = Schema.is(Schema.Json) export const JsonObject = Schema.Record(Schema.String, Schema.Unknown) @@ -233,21 +235,40 @@ export const errorText = (error: unknown) => { /** * `framing` step for Server-Sent Events. Decodes UTF-8, runs the SSE channel - * decoder, and drops empty / `[DONE]` keep-alive events so the downstream - * `decodeChunk` sees one JSON string per element. The SSE channel emits a - * `Retry` control event on its error channel; we drop it here (we don't - * implement client-driven retries) so the public error channel stays - * `LLMError`. + * decoder, and drops keep-alive events — empty, `[DONE]`, and JSON payloads + * that are not objects — so the downstream `decodeChunk` sees one JSON string + * per element. The SSE channel emits a `Retry` control event on its error + * channel; we drop it here (we don't implement client-driven retries) so the + * public error channel stays `LLMError`. */ export const sseFraming = (bytes: Stream.Stream): Stream.Stream => bytes.pipe( Stream.decodeText(), Stream.pipeThroughChannel(Sse.decode()), Stream.catchTag("Retry", () => Stream.empty), - Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"), + Stream.filter((event) => isEventPayload(event.data)), Stream.map((event) => event.data), ) +/** + * Keep-alive test for one `data:` payload. + * + * A non-object payload carries no event for any route: every protocol event + * schema is a struct, so passing one through makes the decoder fail and the + * whole run abort. Some OpenAI-compatible gateways interleave a bare + * `data: null` between real deltas, which is what turns that into a lost step + * (issue #50314) — it is dropped here for the same reason `[DONE]` is. + * + * Only JSON that decodes to a non-object is dropped. Corrupt JSON keeps + * reaching the decoder, and so does an object that fails the event schema, so + * a genuine provider mismatch still fails with the offending payload attached. + */ +const isEventPayload = (data: string) => { + if (data.length === 0 || data === "[DONE]") return false + const decoded = decodeJsonOption(data) + return !isSome(decoded) || isRecord(decoded.value) +} + /** * Canonical invalid-request constructor. Lift one-line `const invalid = * (message) => invalidRequest(message)` aliases out of every diff --git a/packages/llm/test/provider/openai-compatible-chat.test.ts b/packages/llm/test/provider/openai-compatible-chat.test.ts index 43ae283e9f7c..23955d62d02e 100644 --- a/packages/llm/test/provider/openai-compatible-chat.test.ts +++ b/packages/llm/test/provider/openai-compatible-chat.test.ts @@ -6,7 +6,7 @@ import { Auth, LLMClient } from "../../src/route" import * as OpenAICompatible from "../../src/providers/openai-compatible" import * as OpenAICompatibleChat from "../../src/protocols/openai-compatible-chat" import { it } from "../lib/effect" -import { dynamicResponse } from "../lib/http" +import { dynamicResponse, fixedResponse } from "../lib/http" import { sseEvents } from "../lib/sse" const Json = Schema.fromJsonString(Schema.Unknown) @@ -235,4 +235,39 @@ describe("OpenAI-compatible Chat route", () => { expect(response.events.at(-1)).toMatchObject({ type: "finish", reason: "stop" }) }), ) + + it.effect("ignores non-object keep-alive frames injected between deltas", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + dynamicResponse((input) => + Effect.succeed( + input.respond( + sseEvents( + deltaChunk({ role: "assistant", content: "Hello" }), + "null", + deltaChunk({ content: "!" }), + "null", + deltaChunk({}, "stop"), + ), + { headers: { "content-type": "text/event-stream" } }, + ), + ), + ), + ), + ) + + expect(response.text).toBe("Hello!") + expect(response.events.at(-1)).toMatchObject({ type: "finish", reason: "stop" }) + }), + ) + + it.effect("fails on an object frame that is not a stream event", () => + Effect.gen(function* () { + const body = sseEvents(deltaChunk({ role: "assistant", content: "Hello" }), { choices: "nope" }) + const error = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body)), Effect.flip) + + expect(error.message).toContain("Invalid deepseek/openai-compatible-chat stream event") + }), + ) })