From 27cdf5fbb08b7b702af43bd6c3b69c3518657668 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:01:58 -0500 Subject: [PATCH] refactor(ai): avoid repeated option and SSE frame decoding in OpenAI protocols (#57993) --- packages/ai/src/protocols/open-responses.ts | 15 ++++++++------- packages/ai/src/protocols/openai-responses.ts | 2 +- .../ai/src/protocols/openai-transcription.ts | 19 +++++++++++++------ 3 files changed, 22 insertions(+), 14 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index ca5be36690a..13289e06a07 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/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), } }) diff --git a/packages/ai/src/protocols/openai-responses.ts b/packages/ai/src/protocols/openai-responses.ts index 55437c25426..468a7e11ae9 100644 --- a/packages/ai/src/protocols/openai-responses.ts +++ b/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)), } }) diff --git a/packages/ai/src/protocols/openai-transcription.ts b/packages/ai/src/protocols/openai-transcription.ts index 8530ce35eae..83b127184b0 100644 --- a/packages/ai/src/protocols/openai-transcription.ts +++ b/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 -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): 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({