fix(ai): ignore unknown OpenAI speech stream events (#58609)

packages/ai/src/protocols/openai-speech.ts

@@ -1,4 +1,4 @@
-import { Effect, Schema } from "effect"
+import { Effect, Schema, SchemaGetter } from "effect"
 import { Framing } from "../route/framing.js"
 import { MediaProtocol } from "../route/media-protocol.js"
 import { MediaRoute } from "../route/media.js"
@@ -28,6 +28,16 @@ export type Request = SpeechRequestFor<OpenAISpeechOptions>
 // 3. Streaming event schema
 // ---------------------------------------------------------------------------
 
+const EVENT_TYPES = new Set(["speech.audio.delta", "speech.audio.done"])
+/** OpenAI may add stream event types; frames outside `EVENT_TYPES` are ignored, while a malformed known type still fails decoding. */
+const UnknownEvent = Schema.Struct({
+  type: Schema.String.check(Schema.makeFilter((type) => !EVENT_TYPES.has(type))),
+}).pipe(
+  Schema.decodeTo(Schema.Struct({ type: Schema.Literal("unknown"), rawType: Schema.String }), {
+    decode: SchemaGetter.transform((event) => ({ type: "unknown" as const, rawType: event.type })),
+    encode: SchemaGetter.forbidden(() => "Unknown OpenAI Speech stream events are decode-only"),
+  }),
+)
 const SpeechStreamEvent = Schema.Union([
   Schema.Struct({ type: Schema.Literal("speech.audio.delta"), audio: Schema.Uint8ArrayFromBase64 }),
   Schema.Struct({
@@ -40,6 +50,7 @@ const SpeechStreamEvent = Schema.Union([
       }),
     ),
   }),
+  UnknownEvent,
 ])
 
 const decodeEvent = route.decodeFrame(SpeechStreamEvent)
@@ -96,6 +107,7 @@ const isSse = (body: MediaProtocol.Body) => body.type === "json" && body.value.s
 
 const onEvent = Effect.fnUntraced(function* (state: State, frame: string) {
   const event = yield* decodeEvent(frame)
+  if (event.type === "unknown") return [state, []] as const
   if (event.type === "speech.audio.delta") return SpeechStream.delta(state, event.audio)
   const usage = event.usage
   return [

packages/ai/test/speech.test.ts

@@ -209,6 +209,24 @@ describe("Speech", () => {
     }),
   )
 
+  it.effect("ignores unknown OpenAI speech stream events", () =>
+    Effect.gen(function* () {
+      const events = yield* collect(Speech.stream({ model: openai, text: "Hi" })).pipe(
+        Effect.provide(
+          respond(
+            sseEvents(
+              { type: "speech.audio.delta", audio: "AQI=" },
+              { type: "speech.audio.future", value: 1 },
+              { type: "speech.audio.done" },
+            ),
+            "text/event-stream",
+          ),
+        ),
+      )
+      expect(events.map((event) => event.type)).toEqual(["audio-delta", "finish"])
+    }),
+  )
+
   it.effect("surfaces Gemini speech that ended without STOP instead of returning it as complete", () =>
     Effect.gen(function* () {
       const document = (finishReason?: string) =>