refactor(ai): avoid repeated option and SSE frame decoding in OpenAI protocols (#57993)

packages/ai/src/protocols/open-responses.ts

@@ -774,7 +774,7 @@ export const lowerConversation = Effect.fnUntraced(function* (
 export const lowerGeneration = (request: LLMRequest, options = OpenResponsesOptions.resolve(request)) => {
   const generation = request.generation
   const cacheKey = ProviderShared.promptCacheKey(request)
-  const parallelToolCalls = resolveParallelToolCalls(request)
+  const parallelToolCalls = resolveParallelToolCalls(request, options)
   return {
     stream: true as const,
     max_output_tokens: generation?.maxTokens,
@@ -802,15 +802,15 @@ export const lowerGeneration = (request: LLMRequest, options = OpenResponsesOpti
   }
 }
 
-export const resolveParallelToolCalls = (request: LLMRequest) => {
-  const configured = OpenResponsesOptions.resolve(request).parallelToolCalls
+export const resolveParallelToolCalls = (request: LLMRequest, options = OpenResponsesOptions.resolve(request)) => {
+  const configured = options.parallelToolCalls
   if (configured !== undefined) return configured
   const disabled = request.toolChoice?.disableParallelToolUse
   return disabled === undefined ? undefined : !disabled
 }
 
-export const allowedToolChoice = (request: LLMRequest) => {
-  const allowed = OpenResponsesOptions.resolve(request).allowedTools
+export const allowedToolChoice = (request: LLMRequest, options = OpenResponsesOptions.resolve(request)) => {
+  const allowed = options.allowedTools
   if (!allowed) return undefined
   return {
     type: "allowed_tools" as const,
@@ -824,12 +824,13 @@ export const fromRequestWithAdapter = Effect.fn("OpenResponses.fromRequestWithAd
   adapter: ProviderAdapter,
 ) {
   const projected = ProviderShared.flattenToolRequest(request)
+  const options = OpenResponsesOptions.resolve(request)
   return {
     ...(yield* lowerConversation(projected.request, adapter)),
-    ...lowerGeneration(request),
+    ...lowerGeneration(request, options),
     tools: projected.tools.length === 0 ? undefined : yield* lowerTools(projected.tools, adapter),
     tool_choice:
-      allowedToolChoice(request) ??
+      allowedToolChoice(request, options) ??
       (request.toolChoice ? yield* lowerToolChoice(adapter.name, request.toolChoice) : undefined),
   }
 })

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

@@ -210,7 +210,7 @@ const fromRequest = Effect.fn("OpenAIResponses.fromRequest")(function* (request:
     tool_choice:
       request.tools.length === 0
         ? undefined
-        : (OpenResponses.allowedToolChoice(request) ??
+        : (OpenResponses.allowedToolChoice(request, options) ??
           (request.toolChoice ? yield* lowerToolChoice(request.toolChoice, request.tools) : undefined)),
   }
 })

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

@@ -1,4 +1,4 @@
-import { Effect, Schema, Stream } from "effect"
+import { Effect, Schema, SchemaGetter, Stream } from "effect"
 import { classifyProviderFailure } from "../provider-error.js"
 import { Framing } from "../route/framing.js"
 import { MediaProtocol } from "../route/media-protocol.js"
@@ -79,8 +79,16 @@ const transcriptFields = {
   usage: Schema.optional(Usage),
 }
 
-/** OpenAI may add stream event types; frames outside `EVENT_TYPES` are ignored. */
-const EventType = Schema.Struct({ type: Schema.String })
+const EVENT_TYPES = new Set(["transcript.text.delta", "transcript.text.segment", "transcript.text.done", "error"])
+/** 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 Transcription stream events are decode-only"),
+  }),
+)
 const Event = Schema.Union([
   Schema.Struct({ type: Schema.Literal("transcript.text.delta"), delta: Schema.String }),
   Schema.Struct({ type: Schema.Literal("transcript.text.segment"), ...Segment.fields }),
@@ -90,12 +98,11 @@ const Event = Schema.Union([
     message: Schema.optional(Schema.String),
     error: Schema.optional(Schema.Struct({ message: Schema.optional(Schema.String) })),
   }),
+  UnknownEvent,
 ])
-const EVENT_TYPES = new Set(["transcript.text.delta", "transcript.text.segment", "transcript.text.done", "error"])
 const Transcript = Schema.Struct(transcriptFields)
 type Transcript = Schema.Schema.Type<typeof Transcript>
 
-const decodeEventType = route.decodeFrame(EventType)
 const decodeEvent = route.decodeFrame(Event)
 const decodeTranscript = route.decodeFrame(Transcript)
 
@@ -211,8 +218,8 @@ const segment = (value: Schema.Schema.Type<typeof Segment>): TranscriptionSegmen
 })
 
 const onEvent = Effect.fnUntraced(function* (state: State, frame: string) {
-  if (!EVENT_TYPES.has((yield* decodeEventType(frame)).type)) return [state, []] as const
   const event = yield* decodeEvent(frame)
+  if (event.type === "unknown") return [state, []] as const
   if (event.type === "error")
     return yield* new AIError({
       reason: classifyProviderFailure({