refactor(ai): decode WebSocket frames once and derive provider option types from schemas (#58001)

packages/ai/src/experimental/evaluation.ts

@@ -1,8 +1,8 @@
 import { Effect, Schema } from "effect"
+import { tryRequest } from "../media-model.js"
 import {
   AIError,
   HttpOptions,
-  InvalidRequestError,
   ModelID,
   ProviderID,
   ProviderMetadata,
@@ -223,16 +223,7 @@ export function run<const Model extends EvaluationModel, const Questions extends
 ): Effect.Effect<EvaluationResponseFor<Questions>, AIError, Service>
 export function run(input: EvaluationRequest): Effect.Effect<EvaluationResponse, AIError, Service>
 export function run(input: EvaluationRequest | EvaluationRequestInput) {
-  return Effect.try({
-    try: () => request(input),
-    catch: (cause) =>
-      new AIError({
-        reason: new InvalidRequestError({
-          message: cause instanceof Error ? cause.message : String(cause),
-          cause,
-        }),
-      }),
-  }).pipe(Effect.flatMap(EvaluationClient.evaluate))
+  return tryRequest(() => request(input)).pipe(Effect.flatMap(EvaluationClient.evaluate))
 }
 
 export const Evaluation = {

packages/ai/src/protocols/mistral-chat.ts

@@ -102,28 +102,19 @@ const MistralTool = Schema.Struct({
 })
 type MistralTool = Schema.Schema.Type<typeof MistralTool>
 
+const ReasoningEffort = ProviderShared.knownString<"none" | "minimal" | "low" | "medium" | "high" | "xhigh">()
+export type ReasoningEffort = typeof ReasoningEffort.Type
+
 const MistralOptions = Schema.Struct({
   safePrompt: Schema.optional(Schema.Boolean),
   documentImageLimit: Schema.optional(Schema.Number),
   documentPageLimit: Schema.optional(Schema.Number),
   parallelToolCalls: Schema.optional(Schema.Boolean),
-  reasoningEffort: Schema.optional(Schema.String),
+  reasoningEffort: Schema.optional(ReasoningEffort),
   promptMode: Schema.optional(Schema.Literal("reasoning")),
   promptCacheKey: Schema.optional(Schema.String),
 })
-
-export type ReasoningEffort = "none" | "minimal" | "low" | "medium" | "high" | "xhigh" | (string & {})
-
-export type ProviderOptionsInput = {
-  readonly safePrompt?: boolean
-  readonly documentImageLimit?: number
-  readonly documentPageLimit?: number
-  readonly parallelToolCalls?: boolean
-  readonly reasoningEffort?: ReasoningEffort
-  readonly promptMode?: "reasoning"
-  readonly promptCacheKey?: string
-  readonly [key: string]: unknown
-}
+export type ProviderOptionsInput = typeof MistralOptions.Type & { readonly [key: string]: unknown }
 
 export const MistralBody = Schema.Struct({
   model: Schema.String,

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

@@ -1,10 +1,12 @@
 import { Effect, Schema, Stream } from "effect"
 import { Headers } from "effect/http"
 import { Framing } from "../route/framing.js"
-import type { HttpContext } from "../schema/index.js"
+import type { AIError, HttpContext } from "../schema/index.js"
 import {
   HttpTransport,
   WebSocketTransport,
+  type ChannelCreate,
+  type ChannelObservation,
   type Transport,
   type WebSocketChannelDriver,
   type WebSocketChannelExchange,
@@ -48,9 +50,81 @@ const message = (body: unknown) =>
     return { request: decoded, message: encodeMessage(decoded) }
   })
 
-const driver = (options: Options, body: string): WebSocketChannelDriver => {
+const driver = (options: Options, body: string): OpenResponsesContinuation.BaseDriver => {
   let responseID: string | undefined
   let terminal = false
+  const observeEvent = Effect.fnUntraced(function* (
+    _create: ChannelCreate,
+    frame: string,
+    event: OpenResponses.Event,
+  ): Effect.fn.Return<ChannelObservation, AIError> {
+    if (terminal)
+      return yield* ProviderShared.eventError(
+        options.id,
+        `${options.name} emitted ${event.type} after a terminal event`,
+        frame,
+      )
+    if (event.type === "error") {
+      terminal = true
+      return {
+        type: "provider-failure",
+        error: OpenResponses.providerFailure(event, `${options.name} stream error`, frame),
+      }
+    }
+    if (event.type === "response.failed") {
+      terminal = true
+      if (responseID && event.response?.id && event.response.id !== responseID)
+        return yield* ProviderShared.eventError(
+          options.id,
+          `${options.name} response ID changed during execution`,
+          frame,
+        )
+      return {
+        type: "provider-failure",
+        error: OpenResponses.providerFailure(event, `${options.name} response failed`, frame),
+      }
+    }
+    if (event.type === "response.created") {
+      const created = event.response?.id
+      if (responseID)
+        return yield* ProviderShared.eventError(
+          options.id,
+          `${options.name} emitted duplicate response.created`,
+          frame,
+        )
+      if (!created)
+        return yield* ProviderShared.eventError(
+          options.id,
+          `${options.name} response.created is missing response.id`,
+          frame,
+        )
+      responseID = created
+      return { type: "frame", frame }
+    }
+    // Keepalives and provider notifications carry no response state and may precede response.created.
+    if (!event.type.startsWith("response.")) return { type: "frame", frame }
+    if (!responseID)
+      return yield* ProviderShared.eventError(
+        options.id,
+        `${options.name} emitted ${event.type} before response.created`,
+        frame,
+      )
+    if (event.response?.id && event.response.id !== responseID)
+      return yield* ProviderShared.eventError(
+        options.id,
+        `${options.name} response ID changed during execution`,
+        frame,
+      )
+    if (event.type === "response.completed") {
+      terminal = true
+      return { type: "completed", frame }
+    }
+    if (event.type === "response.incomplete") {
+      terminal = true
+      return { type: "incomplete", frame }
+    }
+    return { type: "frame", frame }
+  })
   return {
     create: () =>
       Effect.sync(() => {
@@ -58,80 +132,14 @@ const driver = (options: Options, body: string): WebSocketChannelDriver => {
         terminal = false
         return { message: body, mode: "full" }
       }),
-    observe: (_create, frame) =>
-      Effect.gen(function* () {
-        const event = yield* OpenResponses.decodeChannelEvent(frame).pipe(
-          Effect.mapError((cause) =>
-            ProviderShared.eventError(options.id, `Invalid ${options.name} WebSocket event`, frame, cause),
-          ),
-        )
-        if (terminal)
-          return yield* ProviderShared.eventError(
-            options.id,
-            `${options.name} emitted ${event.type} after a terminal event`,
-            frame,
-          )
-        if (event.type === "error") {
-          terminal = true
-          return {
-            type: "provider-failure",
-            error: OpenResponses.providerFailure(event, `${options.name} stream error`, frame),
-          }
-        }
-        if (event.type === "response.failed") {
-          terminal = true
-          if (responseID && event.response?.id && event.response.id !== responseID)
-            return yield* ProviderShared.eventError(
-              options.id,
-              `${options.name} response ID changed during execution`,
-              frame,
-            )
-          return {
-            type: "provider-failure",
-            error: OpenResponses.providerFailure(event, `${options.name} response failed`, frame),
-          }
-        }
-        if (event.type === "response.created") {
-          const created = event.response?.id
-          if (responseID)
-            return yield* ProviderShared.eventError(
-              options.id,
-              `${options.name} emitted duplicate response.created`,
-              frame,
-            )
-          if (!created)
-            return yield* ProviderShared.eventError(
-              options.id,
-              `${options.name} response.created is missing response.id`,
-              frame,
-            )
-          responseID = created
-          return { type: "frame", frame }
-        }
-        // Keepalives and provider notifications carry no response state and may precede response.created.
-        if (!event.type.startsWith("response.")) return { type: "frame", frame }
-        if (!responseID)
-          return yield* ProviderShared.eventError(
-            options.id,
-            `${options.name} emitted ${event.type} before response.created`,
-            frame,
-          )
-        if (event.response?.id && event.response.id !== responseID)
-          return yield* ProviderShared.eventError(
-            options.id,
-            `${options.name} response ID changed during execution`,
-            frame,
-          )
-        if (event.type === "response.completed") {
-          terminal = true
-          return { type: "completed", frame }
-        }
-        if (event.type === "response.incomplete") {
-          terminal = true
-          return { type: "incomplete", frame }
-        }
-        return { type: "frame", frame }
-      }),
+    observeEvent,
+    observe: (create, frame) =>
+      OpenResponses.decodeChannelEvent(frame).pipe(
+        Effect.mapError((cause) =>
+          ProviderShared.eventError(options.id, `Invalid ${options.name} WebSocket event`, frame, cause),
+        ),
+        Effect.flatMap((event) => observeEvent(create, frame, event)),
+      ),
   }
 }
 

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

@@ -1,5 +1,10 @@
 import { AIError, TransportError } from "../schema/index.js"
-import type { ChannelCheckpoint, ChannelObservation, WebSocketChannelDriver } from "../route/transport/index.js"
+import type {
+  ChannelCheckpoint,
+  ChannelCreate,
+  ChannelObservation,
+  WebSocketChannelDriver,
+} from "../route/transport/index.js"
 import { Effect, Option, Schema } from "effect"
 import * as ProviderShared from "./shared.js"
 import { OpenResponses } from "./open-responses.js"
@@ -20,12 +25,20 @@ interface CheckpointValue {
  */
 export type Shape = (request: Readonly<Record<string, unknown>>) => Readonly<Record<string, unknown>> | undefined
 
+export interface BaseDriver extends WebSocketChannelDriver {
+  readonly observeEvent?: (
+    create: ChannelCreate,
+    frame: string,
+    event: OpenResponses.Event,
+  ) => Effect.Effect<ChannelObservation, AIError>
+}
+
 export interface DriverInput {
   readonly id: string
   readonly name: string
   readonly request: Readonly<Record<string, unknown>>
   readonly message: string
-  readonly base: WebSocketChannelDriver
+  readonly base: BaseDriver
   readonly continuation?: Shape
 }
 
@@ -157,7 +170,9 @@ export const driver = (input: DriverInput): WebSocketChannelDriver => {
             ProviderShared.eventError(input.id, `Invalid ${input.name} WebSocket event`, frame, cause),
           ),
         )
-        const observation = yield* input.base.observe(create, frame)
+        const observation = yield* input.base.observeEvent
+          ? input.base.observeEvent(create, frame, event)
+          : input.base.observe(create, frame)
         if (event.type === "response.output_item.done" && event.item) output.push(event.item)
         if (observation.type === "provider-failure") {
           const rejection = code(event)

packages/ai/src/protocols/venice-chat.ts

@@ -5,15 +5,16 @@ import { classifyProviderFailure } from "../provider-error.js"
 import { OpenAIChat } from "./openai-chat.js"
 import { JsonObject, optionalArray, optionalNull, ProviderShared } from "./shared.js"
 import { cacheControl } from "./utils/cache.js"
+import { OpenResponsesOptions } from "./utils/open-responses-options.js"
 
 // ---------------------------------------------------------------------------
 // Public options and request body
 // ---------------------------------------------------------------------------
 
-export type ReasoningEffort = "none" | "minimal" | "low" | "medium" | "high" | "xhigh" | "max" | (string & {})
+export type ReasoningEffort = OpenResponsesOptions.ReasoningEffort
 
 const Parameters = Schema.Struct({
-  enableWebSearch: Schema.optional(Schema.String),
+  enableWebSearch: Schema.optional(ProviderShared.knownString<"off" | "on" | "auto">()),
   enableWebScraping: Schema.optional(Schema.Boolean),
   enableWebCitations: Schema.optional(Schema.Boolean),
   enableXSearch: Schema.optional(Schema.Boolean),
@@ -26,43 +27,17 @@ const Parameters = Schema.Struct({
 })
 
 const Reasoning = Schema.Struct({
-  effort: Schema.optional(Schema.String),
+  effort: Schema.optional(OpenResponsesOptions.ReasoningEffort),
   enabled: Schema.optional(Schema.Boolean),
-  summary: Schema.optional(Schema.String),
+  summary: Schema.optional(ProviderShared.knownString<"auto" | "concise" | "detailed">()),
 })
 
-export type OptionsInput = {
-  readonly reasoningEffort?: ReasoningEffort
-  readonly reasoning?: {
-    readonly effort?: ReasoningEffort
-    readonly enabled?: boolean
-    readonly summary?: "auto" | "concise" | "detailed" | (string & {})
-  }
-  readonly veniceParameters?: Omit<typeof Parameters.Type, "enableWebSearch"> & {
-    readonly enableWebSearch?: "off" | "on" | "auto" | (string & {})
-  }
-  readonly promptCacheKey?: string
-  readonly promptCacheRetention?: "default" | "extended" | "24h" | (string & {})
-  readonly parallelToolCalls?: boolean
-  readonly maxCompletionTokens?: number
-  readonly maxTokens?: number
-  readonly minP?: number
-  readonly repetitionPenalty?: number
-  readonly stopTokenIds?: readonly number[]
-  readonly logprobs?: boolean
-  readonly topLogprobs?: number
-  readonly maxTemp?: number
-  readonly minTemp?: number
-  readonly responseFormat?: Readonly<Record<string, unknown>>
-  readonly user?: string
-}
-
 const Options = Schema.Struct({
-  reasoningEffort: Schema.optional(Schema.String),
+  reasoningEffort: Schema.optional(OpenResponsesOptions.ReasoningEffort),
   reasoning: Schema.optional(Reasoning),
   veniceParameters: Schema.optional(Parameters),
   promptCacheKey: Schema.optional(Schema.String),
-  promptCacheRetention: Schema.optional(Schema.String),
+  promptCacheRetention: Schema.optional(ProviderShared.knownString<"default" | "extended" | "24h">()),
   parallelToolCalls: Schema.optional(Schema.Boolean),
   maxCompletionTokens: Schema.optional(Schema.Number),
   maxTokens: Schema.optional(Schema.Number),
@@ -76,6 +51,7 @@ const Options = Schema.Struct({
   responseFormat: Schema.optional(JsonObject),
   user: Schema.optional(Schema.String),
 })
+export type OptionsInput = typeof Options.Type
 
 const ToolCall = Schema.Struct({
   ...OpenAIChat.OpenAIChatAssistantToolCall.fields,

packages/ai/src/protocols/zai-chat.ts

@@ -3,34 +3,28 @@ import { Protocol } from "../route/protocol.js"
 import type { LanguageModelCompatibility, LLMRequest } from "../schema/index.js"
 import { OpenAIChat } from "./openai-chat.js"
 import { ProviderShared } from "./shared.js"
+import { OpenResponsesOptions } from "./utils/open-responses-options.js"
 
-export type ReasoningEffort = "none" | "minimal" | "low" | "medium" | "high" | "xhigh" | "max" | (string & {})
-
-export type OptionsInput = {
-  readonly reasoningEffort?: ReasoningEffort
-  readonly thinking?: {
-    readonly type?: "enabled" | "disabled" | (string & {})
-    /** False retains historical reasoning; omission preserves the endpoint's default. */
-    readonly clear_thinking?: boolean
-  }
-  readonly toolStream?: boolean
-  readonly doSample?: boolean
-  readonly responseFormat?: { readonly type: "text" | "json_object" | (string & {}) }
-  readonly requestID?: string
-  readonly userID?: string
-}
+export type ReasoningEffort = OpenResponsesOptions.ReasoningEffort
 
 const Options = Schema.Struct({
-  reasoningEffort: Schema.optional(Schema.String),
+  reasoningEffort: Schema.optional(OpenResponsesOptions.ReasoningEffort),
   thinking: Schema.optional(
-    Schema.Struct({ type: Schema.optional(Schema.String), clear_thinking: Schema.optional(Schema.Boolean) }),
+    Schema.Struct({
+      type: Schema.optional(ProviderShared.knownString<"enabled" | "disabled">()),
+      /** False retains historical reasoning; omission preserves the endpoint's default. */
+      clear_thinking: Schema.optional(Schema.Boolean),
+    }),
   ),
   toolStream: Schema.optional(Schema.Boolean),
   doSample: Schema.optional(Schema.Boolean),
-  responseFormat: Schema.optional(Schema.Struct({ type: Schema.String })),
+  responseFormat: Schema.optional(
+    Schema.Struct({ type: ProviderShared.knownString<"text" | "json_object">() }),
+  ),
   requestID: Schema.optional(Schema.String),
   userID: Schema.optional(Schema.String),
 })
+export type OptionsInput = typeof Options.Type
 
 const Body = Schema.Struct({
   ...OpenAIChat.bodyFields,

packages/ai/src/protocols/zai-messages.ts

@@ -3,17 +3,15 @@ import { Protocol } from "../route/protocol.js"
 import { LLMRequest } from "../schema/index.js"
 import { AnthropicMessages } from "./anthropic-messages.js"
 import { ProviderShared } from "./shared.js"
-import type { ZAIChat } from "./zai-chat.js"
-
-export type OptionsInput = {
-  readonly effort?: ZAIChat.ReasoningEffort
-  readonly thinking?: { readonly type: "enabled" | "adaptive" | "disabled" | (string & {}) }
-}
+import { OpenResponsesOptions } from "./utils/open-responses-options.js"
 
 const Options = Schema.Struct({
-  effort: Schema.optional(Schema.String),
-  thinking: Schema.optional(Schema.Struct({ type: Schema.String })),
+  effort: Schema.optional(OpenResponsesOptions.ReasoningEffort),
+  thinking: Schema.optional(
+    Schema.Struct({ type: ProviderShared.knownString<"enabled" | "adaptive" | "disabled">() }),
+  ),
 })
+export type OptionsInput = typeof Options.Type
 const Body = Schema.Struct({
   ...AnthropicMessages.AnthropicMessagesBody.fields,
   thinking: Options.fields.thinking,