refactor(core): adapt ai sdk transport execution

packages/ai/example/tutorial.ts

@@ -1,6 +1,6 @@
 import { Config, Effect, Formatter, Layer, Schema, Stream } from "effect"
 import { LLM, LLMClient, LLMRequest, Message, ProviderID, Tool, ToolRuntime } from "@opencode-ai/ai"
-import { Route, Auth, Endpoint, Framing, Protocol, RequestExecutor, WebSocketExecutor } from "@opencode-ai/ai/route"
+import { Route, Auth, Endpoint, Framing, Protocol, RequestExecutor } from "@opencode-ai/ai/route"
 import { OpenAI } from "@opencode-ai/ai/providers"
 
 /**
@@ -214,8 +214,7 @@ const FakeEcho = {
 // enabled at a time so the tutorial can demonstrate generate, stream, or
 // tool-loop behavior without spending tokens on every example.
 const requestExecutorLayer = RequestExecutor.fetchLayer
-const llmDeps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
+const llmClientLayer = LLMClient.layer.pipe(Layer.provide(requestExecutorLayer))
-const llmClientLayer = LLMClient.layer.pipe(Layer.provide(llmDeps))
 
 const program = Effect.gen(function* () {
   // yield* generateOnce
@@ -223,6 +222,6 @@ const program = Effect.gen(function* () {
   // yield* generateStructuredObject
   // yield* generateDynamicObject.pipe(Effect.andThen((response) => Effect.sync(() => console.log(response.object))))
   yield* streamWithTools
-}).pipe(Effect.provide(Layer.mergeAll(llmDeps, llmClientLayer)))
+}).pipe(Effect.provide(Layer.mergeAll(requestExecutorLayer, llmClientLayer)))
 
 Effect.runPromise(program)

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

@@ -211,11 +211,43 @@ export type StreamItem = Schema.Schema.Type<typeof StreamItem>
 // event-level `error` envelope, so accept all three shapes here.
 // https://www.openresponses.org/specification
 const OpenResponsesErrorPayload = Schema.Struct({
+  type: optionalNull(Schema.String),
   code: optionalNull(Schema.String),
   message: optionalNull(Schema.String),
   param: optionalNull(Schema.String),
 })
 
+const WebSocketErrorHeader = Schema.Union([Schema.String, Schema.Number, Schema.Boolean])
+export const WebSocketErrorEvent = Schema.StructWithRest(
+  Schema.Struct({
+    type: Schema.tag("error"),
+    status: Schema.optional(Schema.Number),
+    status_code: Schema.optional(Schema.Number),
+    code: optionalNull(Schema.String),
+    message: Schema.optional(Schema.String),
+    param: optionalNull(Schema.String),
+    error: optionalNull(OpenResponsesErrorPayload),
+    headers: Schema.optional(Schema.Record(Schema.String, WebSocketErrorHeader)),
+  }),
+  [Schema.Record(Schema.String, Schema.Unknown)],
+)
+const decodeWebSocketErrorEvent = Schema.decodeUnknownEffect(WebSocketErrorEvent)
+
+const decodeKnownErrorEvent = (event: Event) =>
+  decodeWebSocketErrorEvent({
+    ...event,
+    status: typeof event.status === "number" ? event.status : undefined,
+    status_code: typeof event.status_code === "number" ? event.status_code : undefined,
+    headers: ProviderShared.isRecord(event.headers)
+      ? Object.fromEntries(
+          Object.entries(event.headers).filter(
+            (entry): entry is [string, string | number | boolean] =>
+              typeof entry[1] === "string" || typeof entry[1] === "number" || typeof entry[1] === "boolean",
+          ),
+        )
+      : undefined,
+  })
+
 export const Event = Schema.StructWithRest(
   Schema.Struct({
     type: Schema.String,
@@ -240,6 +272,9 @@ export const Event = Schema.StructWithRest(
     message: Schema.optional(Schema.String),
     param: optionalNull(Schema.String),
     error: optionalNull(OpenResponsesErrorPayload),
+    status: Schema.optional(Schema.Unknown),
+    status_code: Schema.optional(Schema.Unknown),
+    headers: Schema.optional(Schema.Unknown),
   }),
   [Schema.Record(Schema.String, Schema.Unknown)],
 )
@@ -632,9 +667,9 @@ export type StepResult = readonly [ParserState, ReadonlyArray<LLMEvent>]
 const NO_EVENTS: StepResult["1"] = []
 
 // `response.completed` / `response.incomplete` are clean finishes that emit a
-// `finish` event; `response.failed` is a hard failure. All three end the stream,
+// `finish` event; `response.failed` and `error` are hard failures. All four end
-// so keep this set aligned with `step` and the protocol's terminal predicate.
+// the stream, so keep this set aligned with `step` and the protocol's terminal predicate.
-const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"])
+const TERMINAL_TYPES = new Set(["error", "response.completed", "response.incomplete", "response.failed"])
 export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type)
 
 const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => {
@@ -969,10 +1004,16 @@ const providerErrorMessage = (event: Event, fallback: string): string => {
 const providerError = (state: ParserState, event: Event, fallback: string) => {
   const code = event.code || event.error?.code || event.response?.error?.code || undefined
   const message = providerErrorMessage(event, fallback)
+  const status =
+    typeof event.status === "number"
+      ? event.status
+      : typeof event.status_code === "number"
+        ? event.status_code
+        : undefined
   return new AIError({
     module: state.id,
     method: "stream",
-    reason: classifyProviderFailure({ message, code }),
+    reason: classifyProviderFailure({ message, code, status }),
   })
 }
 
@@ -1015,7 +1056,11 @@ export const step = (state: ParserState, event: Event) => {
   if (event.type === "response.completed" || event.type === "response.incomplete")
     return Effect.succeed(onResponseFinish(state, event))
   if (event.type === "response.failed") return providerError(state, event, `${state.name} response failed`)
-  if (event.type === "error") return providerError(state, event, `${state.name} stream error`)
+  if (event.type === "error")
+    return decodeKnownErrorEvent(event).pipe(
+      Effect.mapError(() => ProviderShared.eventError(state.id, `${state.name} returned a malformed error event`)),
+      Effect.flatMap(() => providerError(state, event, `${state.name} stream error`)),
+    )
   return Effect.succeed<StepResult>([state, NO_EVENTS])
 }
 

packages/ai/src/provider-error.ts

@@ -67,6 +67,7 @@ const SERVER_CODES = new Set([
   "overloaded_error",
   "server_error",
   "server_is_overloaded",
+  "slow_down",
   "serviceunavailableexception",
 ])
 const INVALID_REQUEST_CODES = new Set(["invalid_prompt", "invalid_request_error", "validationexception"])

packages/ai/src/route/client.ts

@@ -1,12 +1,10 @@
 import { Cause, Context, Effect, Layer, Schema, Stream } from "effect"
-import * as Option from "effect/Option"
 import { Auth } from "./auth"
 import { Endpoint, type EndpointPatch } from "./endpoint"
 import { RequestExecutor } from "./executor"
 import { Framing } from "./framing"
 import { HttpTransport } from "./transport"
-import type { HttpMiddleware, Transport, TransportRuntime } from "./transport"
+import type { HttpMiddleware, Transport, TransportRuntime, WebSocketChannelExecutor } from "./transport"
-import { WebSocketExecutor } from "./transport"
 import type { Protocol } from "./protocol"
 import { applyCachePolicy } from "../cache-policy"
 import * as ProviderShared from "../protocols/shared"
@@ -58,6 +56,7 @@ export interface Route<Body, Prepared = unknown> {
     prepared: Prepared,
     request: LLMRequest,
     runtime: TransportRuntime,
+    options?: StreamOptions,
   ) => Stream.Stream<LLMEvent, AIError>
 }
 
@@ -157,6 +156,7 @@ export interface Interface {
 
 export interface StreamOptions {
   readonly http?: HttpMiddleware
+  readonly webSocket?: WebSocketChannelExecutor
 }
 
 export interface StreamMethod {
@@ -255,13 +255,7 @@ const requireTerminalEvent = (route: string) => (events: Stream.Stream<LLMEvent,
         if (LLMEvent.is.finish(event) || LLMEvent.is.providerError(event)) terminal = true
         return Effect.succeed(event)
       }),
-      Stream.onEnd(
+      Stream.onEnd(Effect.suspend(() => (terminal ? Effect.void : Effect.fail(incompleteStreamError(route))))),
-        Effect.suspend(() =>
-          terminal
-            ? Effect.void
-            : Effect.fail(incompleteStreamError(route)),
-        ),
-      ),
     )
   })
 
@@ -321,15 +315,16 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
           headers: routeInput.headers,
           middleware: options?.http,
         }),
-      streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => {
+      streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime, options?: StreamOptions) => {
         const route = `${request.model.provider}/${request.model.route.id}`
-        const events = routeInput.transport
+        return Stream.unwrap(
-          .frames(prepared, request, runtime)
+          routeInput.transport.execute(prepared, request, runtime, options).pipe(
-          .pipe(
+            Effect.map((execution) => {
+              const events = execution.frames.pipe(
                 Stream.mapEffect(decodeEvent(route)),
                 protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
               )
-        return events.pipe(
+              const stream = events.pipe(
                 Stream.mapAccumEffect(
                   () => protocol.stream.initial(request),
                   protocol.stream.step,
@@ -338,6 +333,10 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
                 Stream.catchCause((cause) => Stream.fail(streamError(route, `Failed to read ${route} stream`, cause))),
                 requireTerminalEvent(route),
               )
+              return execution.complete ? stream.pipe(Stream.onEnd(execution.complete)) : stream
+            }),
+          ),
+        )
       },
     } satisfies Route<Body, Prepared>
     return route
@@ -419,7 +418,7 @@ const streamRequestWith = (runtime: TransportRuntime) => (request: LLMRequest, o
   Stream.unwrap(
     Effect.gen(function* () {
       const compiled = yield* compile(request, options)
-      return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime)
+      return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime, options)
     }),
   )
 
@@ -457,7 +456,6 @@ export const layer: Layer.Layer<Service, never, RequestExecutor.Service> = Layer
   Effect.gen(function* () {
     const stream = streamRequestWith({
       http: yield* RequestExecutor.Service,
-      webSocket: Option.getOrUndefined(yield* Effect.serviceOption(WebSocketExecutor.Service)),
     })
     return Service.of({ stream, generate: generateWith(stream) })
   }),

packages/ai/src/route/index.ts

@@ -16,11 +16,28 @@ export { AuthOptions } from "./auth-options"
 export { Endpoint } from "./endpoint"
 export { Framing } from "./framing"
 export { Protocol } from "./protocol"
-export { HttpTransport, WebSocketExecutor, WebSocketTransport } from "./transport"
+export { HttpTransport, WebSocketTransport } from "./transport"
 export * as Transport from "./transport"
 export type { Definition as AuthShape, AuthInput, Credential, CredentialError } from "./auth"
 export type { ApiKeyMode, AuthOverride, ProviderAuthOption } from "./auth-options"
 export type { Definition as EndpointFn, EndpointInput } from "./endpoint"
 export type { Definition as FramingDef } from "./framing"
 export type { Protocol as ProtocolDef } from "./protocol"
-export type { HttpHandler, HttpMiddleware, Transport as TransportDef, TransportRuntime } from "./transport"
+export type {
+  ChannelCheckpoint,
+  ChannelCreate,
+  ChannelObservation,
+  HttpHandler,
+  HttpMiddleware,
+  Transport as TransportDef,
+  TransportExecuteOptions,
+  TransportExecution,
+  TransportRuntime,
+  WebSocketConnection,
+  WebSocketChannelDriver,
+  WebSocketChannelExchange,
+  WebSocketChannelExecution,
+  WebSocketChannelExecutor,
+  WebSocketConnector,
+  WebSocketRequest,
+} from "./transport"

packages/ai/src/route/transport/http.ts

@@ -86,8 +86,9 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
         middleware: prepareInput.middleware,
       }
     }),
-  frames: (prepared, request, runtime) =>
+  execute: (prepared, request, runtime) =>
-    Stream.unwrap(
+    Effect.succeed({
+      frames: Stream.unwrap(
         runtime.http
           .execute(prepared.request, prepared.middleware)
           .pipe(
@@ -106,6 +107,7 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
             ),
           ),
       ),
+    }),
 })
 
 export const sseJson = {

packages/ai/src/route/transport/index.ts

@@ -1,19 +1,33 @@
-import type { Effect, Stream } from "effect"
+import type { Effect, Scope, Stream } from "effect"
 import { Endpoint } from "../endpoint"
 import { Auth } from "../auth"
 import type { HttpMiddleware, Interface as RequestExecutorInterface } from "../executor"
-import type { Interface as WebSocketExecutorInterface } from "./websocket"
+import type { WebSocketChannelExecutor } from "./websocket-channel"
 import type { AIError, LLMRequest } from "../../schema"
 
 export interface TransportRuntime {
   readonly http: RequestExecutorInterface
-  readonly webSocket?: WebSocketExecutorInterface
+}
+
+export interface TransportExecution<Frame> {
+  readonly frames: Stream.Stream<Frame, AIError>
+  /** Optional successful-consumption acknowledgement. HTTP leaves this absent. */
+  readonly complete?: Effect.Effect<void>
+}
+
+export interface TransportExecuteOptions {
+  readonly webSocket?: WebSocketChannelExecutor
 }
 
 export interface Transport<Body, Prepared, Frame> {
   readonly id: string
   readonly prepare: (input: TransportPrepareInput<Body>) => Effect.Effect<Prepared, AIError>
-  readonly frames: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => Stream.Stream<Frame, AIError>
+  readonly execute: (
+    prepared: Prepared,
+    request: LLMRequest,
+    runtime: TransportRuntime,
+    options?: TransportExecuteOptions,
+  ) => Effect.Effect<TransportExecution<Frame>, AIError, Scope.Scope>
 }
 
 export interface TransportPrepareInput<Body> {
@@ -28,4 +42,14 @@ export interface TransportPrepareInput<Body> {
 
 export * as HttpTransport from "./http"
 export type { HttpHandler, HttpMiddleware } from "../executor"
-export { WebSocketExecutor, WebSocketTransport } from "./websocket"
+export type {
+  ChannelCheckpoint,
+  ChannelCreate,
+  ChannelObservation,
+  WebSocketChannelDriver,
+  WebSocketChannelExchange,
+  WebSocketChannelExecution,
+  WebSocketChannelExecutor,
+} from "./websocket-channel"
+export type { WebSocketConnection, WebSocketConnector, WebSocketRequest } from "./websocket"
+export { WebSocketTransport } from "./websocket"

packages/ai/src/route/transport/websocket-channel.ts

@@ -0,0 +1,48 @@
+import type { Effect, Scope, Stream } from "effect"
+import type { Headers } from "effect/unstable/http"
+import type { AIError } from "../../schema"
+
+export interface WebSocketChannelExecutor {
+  readonly execute: (
+    exchange: WebSocketChannelExchange,
+  ) => Effect.Effect<WebSocketChannelExecution, AIError, Scope.Scope>
+}
+
+export interface WebSocketChannelExecution {
+  readonly frames: Stream.Stream<string, AIError>
+  /** Commits staged state after the decoded Route stream ends successfully. */
+  readonly complete: Effect.Effect<void>
+}
+
+export interface WebSocketChannelExchange {
+  readonly id: string
+  readonly connect: {
+    readonly url: string
+    readonly headers: Headers.Headers
+  }
+  readonly fallback: () => Stream.Stream<string, AIError>
+  readonly driver: WebSocketChannelDriver
+}
+
+export interface WebSocketChannelDriver {
+  readonly create: (checkpoint: ChannelCheckpoint | undefined) => Effect.Effect<ChannelCreate, AIError>
+  readonly observe: (create: ChannelCreate, frame: string) => Effect.Effect<ChannelObservation, AIError>
+}
+
+export interface ChannelCreate {
+  readonly message: string
+  readonly mode: "full" | "incremental"
+}
+
+export type ChannelObservation =
+  | { readonly type: "frame"; readonly frame: string }
+  | { readonly type: "completed"; readonly frame: string; readonly checkpoint?: ChannelCheckpoint }
+  | { readonly type: "incomplete"; readonly frame: string }
+  | { readonly type: "provider-failure"; readonly error: AIError }
+  | { readonly type: "rejected"; readonly error: AIError; readonly recovery: "retry-full" }
+  | { readonly type: "rejected"; readonly error: AIError; readonly recovery: "rotate-and-retry-full" }
+
+export interface ChannelCheckpoint {
+  readonly protocol: string
+  readonly value: unknown
+}

packages/ai/src/route/transport/websocket.ts

@@ -1,8 +1,15 @@
-import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
+import { Cause, Effect, Queue, Stream } from "effect"
 import { Headers } from "effect/unstable/http"
+import { Socket } from "effect/unstable/socket"
 import { AIError, TransportReason } from "../../schema"
 import * as HttpTransport from "./http"
 import type { Transport } from "./index"
+import type {
+  ChannelObservation,
+  WebSocketChannelDriver,
+  WebSocketChannelExchange,
+  WebSocketChannelExecutor,
+} from "./websocket-channel"
 
 export interface WebSocketRequest {
   readonly url: string
@@ -15,28 +22,57 @@ export interface WebSocketConnection {
   readonly close: Effect.Effect<void, never>
 }
 
-export interface Interface {
+export interface WebSocketConnector {
   readonly open: (input: WebSocketRequest) => Effect.Effect<WebSocketConnection, AIError>
 }
 
-type WebSocketConstructorWithHeaders = new (
+type WebSocketConstructorWithHeaders = (
   url: string,
   options?: { readonly headers?: Headers.Headers },
 ) => globalThis.WebSocket
 
-export class Service extends Context.Service<Service, Interface>()("@opencode/AI/WebSocketExecutor") {}
-
 const transportError = (
   method: string,
   message: string,
-  input: { readonly url?: string; readonly kind?: string } = {},
+  input: {
+    readonly url?: string
+    readonly kind?: string
+    readonly phase?: TransportReason["phase"]
+    readonly delivery?: TransportReason["delivery"]
+  } = {},
 ) =>
   new AIError({
-    module: "WebSocketExecutor",
+    module: "WebSocketConnector",
     method,
-    reason: new TransportReason({ message, url: input.url, kind: input.kind }),
+    reason: new TransportReason({
+      message,
+      url: input.url,
+      kind: input.kind,
+      phase: input.phase,
+      delivery: input.delivery,
+    }),
   })
 
+const annotateTransportError = (
+  error: AIError,
+  input: { readonly phase: TransportReason["phase"]; readonly delivery: TransportReason["delivery"] },
+) =>
+  error.reason._tag === "Transport"
+    ? new AIError({
+        module: error.module,
+        method: error.method,
+        reason: new TransportReason({
+          message: error.reason.message,
+          kind: error.reason.kind,
+          url: error.reason.url,
+          http: error.reason.http,
+          phase: input.phase,
+          delivery: input.delivery,
+          recovery: error.reason.recovery,
+        }),
+      })
+    : error
+
 const eventMessage = (event: Event) => {
   if ("message" in event && typeof event.message === "string") return event.message
   return event.type
@@ -56,6 +92,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
       transportError("open", `WebSocket closed before opening (state ${ws.readyState})`, {
         url: input.url,
         kind: "open",
+        phase: "connect",
+        delivery: "not-sent",
       }),
     )
   }
@@ -79,7 +117,12 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
       cleanup()
       resume(
         Effect.fail(
-          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
+          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
+            url: input.url,
+            kind: "open",
+            phase: "connect",
+            delivery: "not-sent",
+          }),
         ),
       )
     }
@@ -90,6 +133,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
           transportError("open", `WebSocket closed before opening with code ${event.code}`, {
             url: input.url,
             kind: "open",
+            phase: "connect",
+            delivery: "not-sent",
           }),
         ),
       )
@@ -119,21 +164,31 @@ const webSocketUrl = (value: string) =>
       transportError("prepare", error instanceof Error ? error.message : "Invalid WebSocket URL", {
         url: value,
         kind: "websocket",
+        phase: "prepare",
+        delivery: "not-sent",
       }),
   })
 
 export const open = (input: WebSocketRequest) =>
-  Effect.try({
+  Effect.gen(function* () {
+    const constructor = yield* Socket.WebSocketConstructor
+    const ws = yield* Effect.try({
       try: () =>
-      new (globalThis.WebSocket as unknown as WebSocketConstructorWithHeaders)(input.url, { headers: input.headers }),
+        // Platform implementations may extend Effect's browser-compatible constructor with handshake options.
+        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
+        (constructor as unknown as WebSocketConstructorWithHeaders)(input.url, {
+          headers: input.headers,
+        }),
       catch: (error) =>
         transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
           url: input.url,
           kind: "open",
+          phase: "connect",
+          delivery: "not-sent",
         }),
-  }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
+    })
-
+    return yield* fromWebSocket(ws, input)
-export const layer: Layer.Layer<Service> = Layer.succeed(Service, Service.of({ open }))
+  })
 
 export const fromWebSocket = (
   ws: globalThis.WebSocket,
@@ -150,7 +205,11 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", "Unsupported WebSocket message payload", { url: input.url, kind: "message" }),
+          transportError("message", "Unsupported WebSocket message payload", {
+            url: input.url,
+            kind: "message",
+            phase: "receive",
+          }),
         ),
       )
     }
@@ -158,16 +217,23 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, kind: "message" }),
+          transportError("message", `WebSocket error: ${eventMessage(event)}`, {
+            url: input.url,
+            kind: "message",
+            phase: "receive",
+          }),
         ),
       )
     }
     const onClose = (event: CloseEvent) => {
-      if (event.code === 1000 || event.code === 1005) return Queue.endUnsafe(messages)
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket closed with code ${event.code}`, { url: input.url, kind: "close" }),
+          transportError("message", `WebSocket closed with code ${event.code}`, {
+            url: input.url,
+            kind: "close",
+            phase: "close",
+          }),
         ),
       )
     }
@@ -189,6 +255,8 @@ export const fromWebSocket = (
             transportError("sendText", error instanceof Error ? error.message : "Failed to send WebSocket message", {
               url: input.url,
               kind: "write",
+              phase: "send",
+              delivery: "not-sent",
             }),
         }),
       messages: Stream.fromQueue(messages),
@@ -206,6 +274,57 @@ export const fromWebSocket = (
 export const messageText = (message: string | Uint8Array, decoder: TextDecoder) =>
   typeof message === "string" ? message : decoder.decode(message)
 
+const observationFrame = (observation: ChannelObservation) => {
+  if (observation.type === "frame" || observation.type === "completed" || observation.type === "incomplete")
+    return Effect.succeed(observation.frame)
+  return Effect.fail(observation.error)
+}
+
+const observationTerminal = (observation: ChannelObservation) => observation.type !== "frame"
+
+export const makeDirect = (connector: WebSocketConnector): WebSocketChannelExecutor => ({
+  execute: (exchange) =>
+    Effect.gen(function* () {
+      const connection = yield* Effect.acquireRelease(
+        connector
+          .open(exchange.connect)
+          .pipe(Effect.mapError((error) => annotateTransportError(error, { phase: "connect", delivery: "not-sent" }))),
+        (connection) => connection.close,
+      )
+      const create = yield* exchange.driver.create(undefined)
+      yield* connection.sendText(create.message)
+      const decoder = new TextDecoder()
+      let observed = false
+      return {
+        frames: connection.messages.pipe(
+          Stream.map((message) => {
+            observed = true
+            return messageText(message, decoder)
+          }),
+          Stream.mapError((error) =>
+            annotateTransportError(error, {
+              phase: error.reason._tag === "Transport" && error.reason.phase === "close" ? "close" : "receive",
+              delivery: observed ? "accepted" : "ambiguous",
+            }),
+          ),
+          Stream.mapEffect((frame) => exchange.driver.observe(create, frame)),
+          Stream.takeUntil(observationTerminal),
+          Stream.mapEffect(observationFrame),
+        ),
+        complete: Effect.void,
+      }
+    }),
+})
+
+export const direct: Effect.Effect<WebSocketChannelExecutor, never, Socket.WebSocketConstructor> = Effect.gen(
+  function* () {
+    const constructor = yield* Socket.WebSocketConstructor
+    return makeDirect({
+      open: (input) => open(input).pipe(Effect.provideService(Socket.WebSocketConstructor, constructor)),
+    })
+  },
+)
+
 export interface JsonPrepared {
   readonly url: string
   readonly headers: Headers.Headers
@@ -237,27 +356,37 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
         message: input.encodeMessage(yield* input.toMessage(parts.jsonBody)),
       }
     }),
-  frames: (prepared, _request, runtime) => {
+  execute: (prepared, request, _runtime, options) => {
-    const webSocket = runtime.webSocket
+    const webSocket = options?.webSocket
     if (!webSocket) {
-      return Stream.fail(
+      return Effect.fail(
-        transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
+        transportError("json", "WebSocket JSON transport requires StreamOptions.webSocket", {
           url: prepared.url,
           kind: "websocket",
+          phase: "prepare",
+          delivery: "not-sent",
         }),
       )
     }
-    const decoder = new TextDecoder()
+    const driver: WebSocketChannelDriver = {
-    return Stream.unwrap(
+      create: () => Effect.succeed({ message: prepared.message, mode: "full" }),
-      Effect.gen(function* () {
+      observe: (_create, frame) => Effect.succeed({ type: "frame", frame }),
-        const connection = yield* Effect.acquireRelease(
+    }
-          webSocket.open({ url: prepared.url, headers: prepared.headers }),
+    const exchange: WebSocketChannelExchange = {
-          (connection) => connection.close,
+      id: request.id ?? "request",
-        )
+      connect: { url: prepared.url, headers: prepared.headers },
-        yield* connection.sendText(prepared.message)
+      fallback: () =>
-        return connection.messages.pipe(Stream.map((message) => messageText(message, decoder)))
+        Stream.fail(
+          transportError("fallback", "WebSocket JSON transport does not provide HTTP fallback", {
+            url: prepared.url,
+            kind: "websocket",
+            phase: "fallback",
+            delivery: "not-sent",
           }),
-    )
+        ),
+      driver,
+    }
+    return webSocket.execute(exchange)
   },
 })
 
@@ -266,15 +395,12 @@ export const jsonTransport = {
   with: json,
 } as const
 
-export const WebSocketExecutor = {
+export const WebSocketTransport = {
-  Service,
+  json,
-  layer,
+  jsonTransport,
+  direct,
+  makeDirect,
   open,
   fromWebSocket,
   messageText,
 } as const
-
-export const WebSocketTransport = {
-  json,
-  jsonTransport,
-} as const

packages/ai/src/schema/errors.ts

@@ -98,6 +98,13 @@ export class TransportReason extends Schema.Class<TransportReason>("AI.Error.Tra
   kind: Schema.optional(Schema.String),
   url: Schema.optional(Schema.String),
   http: Schema.optional(HttpContext),
+  phase: Schema.optional(
+    Schema.Literals(["prepare", "queue", "connect", "send", "receive", "decode", "complete", "fallback", "close"]),
+  ),
+  delivery: Schema.optional(Schema.Literals(["not-sent", "rejected", "ambiguous", "accepted"])),
+  recovery: Schema.optional(
+    Schema.Literals(["retry-connect", "retry-full", "rotate-and-retry-full", "fallback-http", "fail"]),
+  ),
 }) {}
 
 export class InvalidProviderOutputReason extends Schema.Class<InvalidProviderOutputReason>(

packages/ai/test/executor.test.ts

@@ -1,10 +1,11 @@
 import { describe, expect } from "bun:test"
-import { Effect, Layer, Ref } from "effect"
+import { Deferred, Effect, Fiber, Layer, Ref, Stream } from "effect"
 import { Headers, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 import { LLM, AIError } from "../src"
-import { LLMClient, RequestExecutor } from "../src/route"
+import { LLMClient, RequestExecutor, WebSocketTransport, type WebSocketChannelExecutor } from "../src/route"
 import * as OpenAIChat from "../src/protocols/openai-chat"
-import { dynamicResponse } from "./lib/http"
+import * as OpenAI from "../src/providers/openai"
+import { dynamicResponse, fixedResponse } from "./lib/http"
 import { deltaChunk } from "./lib/openai-chunks"
 import { sseRaw } from "./lib/sse"
 import { it } from "./lib/effect"
@@ -413,3 +414,127 @@ describe("RequestExecutor", () => {
     }),
   )
 })
+
+describe("WebSocket channel execution", () => {
+  const model = OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+    "gpt-4.1-mini",
+  )
+  const request = LLM.request({ model, prompt: "Say hello." })
+  const frames = [
+    JSON.stringify({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }),
+    JSON.stringify({ type: "response.completed", response: { id: "resp_1" } }),
+  ]
+
+  it.effect("runs a channel driver through the direct executor", () =>
+    Effect.gen(function* () {
+      const sent = yield* Ref.make("")
+      const closed = yield* Ref.make(false)
+      const observed = yield* Ref.make(0)
+      const webSocket = WebSocketTransport.makeDirect({
+        open: () =>
+          Effect.succeed({
+            sendText: (message) => Ref.set(sent, message),
+            messages: Stream.make("one", "done", "late"),
+            close: Ref.set(closed, true),
+          }),
+      })
+      const received = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const execution = yield* webSocket.execute({
+            id: "exchange_1",
+            connect: { url: "wss://api.openai.test/v1/responses", headers: Headers.empty },
+            fallback: () => Stream.empty,
+            driver: {
+              create: () => Effect.succeed({ message: "create", mode: "full" }),
+              observe: (_create, frame) =>
+                Ref.update(observed, (value) => value + 1).pipe(
+                  Effect.as(
+                    frame === "done" ? { type: "completed" as const, frame } : { type: "frame" as const, frame },
+                  ),
+                ),
+            },
+          })
+          return yield* Stream.runCollect(execution.frames)
+        }),
+      )
+
+      expect(Array.from(received)).toEqual(["one", "done"])
+      expect(yield* Ref.get(sent)).toBe("create")
+      expect(yield* Ref.get(observed)).toBe(2)
+      expect(yield* Ref.get(closed)).toBe(true)
+    }),
+  )
+
+  it.effect("requires a per-call WebSocket executor", () =>
+    Effect.gen(function* () {
+      const error = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse("")), Effect.flip)
+
+      expect(error.reason).toMatchObject({
+        _tag: "Transport",
+        phase: "prepare",
+        delivery: "not-sent",
+      })
+      expect(error.message).toContain("StreamOptions.webSocket")
+    }),
+  )
+
+  it.effect("commits channel execution only after complete consumption", () =>
+    Effect.gen(function* () {
+      const commits = yield* Ref.make(0)
+      const executor = (input: Stream.Stream<string, AIError>): WebSocketChannelExecutor => ({
+        execute: () =>
+          Effect.succeed({
+            frames: input,
+            complete: Ref.update(commits, (value) => value + 1),
+          }),
+      })
+
+      const response = yield* LLMClient.generate(request, {
+        webSocket: executor(Stream.fromArray(frames)),
+      }).pipe(Effect.provide(fixedResponse("")))
+      expect(response.text).toBe("Hi")
+      expect(yield* Ref.get(commits)).toBe(1)
+
+      yield* LLMClient.generate(request, { webSocket: executor(Stream.make("not-json")) }).pipe(
+        Effect.provide(fixedResponse("")),
+        Effect.flip,
+      )
+      expect(yield* Ref.get(commits)).toBe(1)
+
+      yield* LLMClient.stream(request, { webSocket: executor(Stream.fromArray(frames)) }).pipe(
+        Stream.take(1),
+        Stream.runDrain,
+        Effect.provide(fixedResponse("")),
+      )
+      expect(yield* Ref.get(commits)).toBe(1)
+    }),
+  )
+
+  it.effect("does not commit interrupted channel execution", () =>
+    Effect.gen(function* () {
+      const commits = yield* Ref.make(0)
+      const started = yield* Deferred.make<void>()
+      const executor: WebSocketChannelExecutor = {
+        execute: () =>
+          Effect.succeed({
+            frames: Stream.fromEffect(
+              Deferred.succeed(started, undefined).pipe(
+                Effect.as(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })),
+              ),
+            ).pipe(Stream.concat(Stream.never)),
+            complete: Ref.update(commits, (value) => value + 1),
+          }),
+      }
+      const fiber = yield* LLMClient.stream(request, { webSocket: executor }).pipe(
+        Stream.runDrain,
+        Effect.provide(fixedResponse("")),
+        Effect.forkChild({ startImmediately: true }),
+      )
+
+      yield* Deferred.await(started)
+      yield* Fiber.interrupt(fiber)
+
+      expect(yield* Ref.get(commits)).toBe(0)
+    }),
+  )
+})

packages/ai/test/exports.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, test } from "bun:test"
 import { AIError, ImageInput, LanguageModel, LLM, LLMClient, Provider } from "@opencode-ai/ai"
-import { Route, Protocol } from "@opencode-ai/ai/route"
+import { Route, Protocol, WebSocketTransport } from "@opencode-ai/ai/route"
 import { Provider as ProviderSubpath } from "@opencode-ai/ai/provider"
 import {
   CloudflareAIGateway,
@@ -37,6 +37,7 @@ describe("public exports", () => {
   test("route barrel exposes route-authoring APIs", () => {
     expect(Route.make).toBeFunction()
     expect(Protocol.make).toBeFunction()
+    expect(WebSocketTransport.makeDirect).toBeFunction()
   })
 
   test("provider barrels expose user-facing facades", async () => {

packages/ai/test/lib/http.ts

@@ -1,9 +1,8 @@
 import { Effect, Layer, Ref } from "effect"
 import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
-import { LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route"
+import { LLMClient, RequestExecutor } from "../../src/route"
 import type { Service as LLMClientService } from "../../src/route/client"
 import type { Service as RequestExecutorService } from "../../src/route/executor"
-import type { Service as WebSocketExecutorService } from "../../src/route/transport/websocket"
 
 export type HandlerInput = {
   readonly request: HttpClientRequest.HttpClientRequest
@@ -32,13 +31,12 @@ const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
     ),
   )
 
-export type RuntimeEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService
+export type RuntimeEnv = RequestExecutorService | LLMClientService
 
 export const runtimeLayer = (layer: Layer.Layer<HttpClient.HttpClient>): Layer.Layer<RuntimeEnv> => {
   const requestExecutorLayer = RequestExecutor.layer.pipe(Layer.provide(layer))
-  const deps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
+  const llmClientLayer = LLMClient.layer.pipe(Layer.provide(requestExecutorLayer))
-  const llmClientLayer = LLMClient.layer.pipe(Layer.provide(deps))
+  return Layer.mergeAll(requestExecutorLayer, llmClientLayer)
-  return Layer.mergeAll(deps, llmClientLayer)
 }
 
 const SSE_HEADERS = { "content-type": "text/event-stream" } as const

packages/ai/test/provider-error.test.ts

@@ -69,10 +69,10 @@ describe("provider error classification", () => {
 
   test("classifies V1 overloaded provider codes", () => {
     expect(
-      ['{"code":"resource_exhausted"}', '{"code":"service_unavailable"}'].map(
+      ['{"code":"resource_exhausted"}', '{"code":"service_unavailable"}', '{"code":"slow_down"}'].map(
         (message) => classifyProviderFailure({ message })._tag,
       ),
-    ).toEqual(["ProviderInternal", "ProviderInternal"])
+    ).toEqual(["ProviderInternal", "ProviderInternal", "ProviderInternal"])
   })
 
   test("classifies transient client statuses as provider internal", () => {

packages/ai/test/provider/openai-responses.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { ConfigProvider, Effect, Layer, Stream } from "effect"
+import { ConfigProvider, Effect, Layer, Ref, Stream } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import {
   LLM,
@@ -11,9 +11,10 @@ import {
   ToolCallPart,
   ToolDefinition,
   ToolResultPart,
+  TransportReason,
   Usage,
 } from "../../src"
-import { Auth, LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route"
+import { Auth, LLMClient, RequestExecutor, WebSocketTransport } from "../../src/route"
 import { compileRequest } from "../../src/route/client"
 import * as Azure from "../../src/providers/azure"
 import * as OpenAI from "../../src/providers/openai"
@@ -238,16 +239,13 @@ describe("OpenAI Responses route", () => {
       const sent: string[] = []
       const opened: Array<{ readonly url: string; readonly authorization: string | undefined }> = []
       let closed = false
-      const deps = Layer.mergeAll(
+      const deps = Layer.succeed(
-        Layer.succeed(
         RequestExecutor.Service,
         RequestExecutor.Service.of({
           execute: () => Effect.die("unexpected HTTP request"),
         }),
-        ),
+      )
-        Layer.succeed(
+      const webSocket = WebSocketTransport.makeDirect({
-          WebSocketExecutor.Service,
-          WebSocketExecutor.Service.of({
         open: (input) =>
           Effect.succeed({
             sendText: (message) =>
@@ -263,9 +261,7 @@ describe("OpenAI Responses route", () => {
               closed = true
             }),
           }),
-          }),
+      })
-        ),
-      )
       const response = yield* LLMClient.generate(
         LLM.request({
           model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
@@ -273,6 +269,7 @@ describe("OpenAI Responses route", () => {
           ),
           prompt: "Say hello.",
         }),
+        { webSocket },
       ).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))))
 
       expect(response.text).toBe("Hi")
@@ -288,15 +285,158 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("closes a direct WebSocket execution after partial consumption", () =>
+    Effect.gen(function* () {
+      const closed = yield* Ref.make(false)
+      const webSocket = WebSocketTransport.makeDirect({
+        open: () =>
+          Effect.succeed({
+            sendText: () => Effect.void,
+            messages: Stream.fromArray([
+              ProviderShared.encodeJson({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }),
+              ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_ws" } }),
+            ]),
+            close: Ref.set(closed, true),
+          }),
+      })
+
+      yield* LLMClient.stream(
+        LLM.request({
+          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+            "gpt-4.1-mini",
+          ),
+          prompt: "Say hello.",
+        }),
+        { webSocket },
+      ).pipe(
+        Stream.take(1),
+        Stream.runDrain,
+        Effect.provide(
+          LLMClient.layer.pipe(
+            Layer.provide(
+              Layer.succeed(
+                RequestExecutor.Service,
+                RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+              ),
+            ),
+          ),
+        ),
+      )
+
+      expect(yield* Ref.get(closed)).toBe(true)
+    }),
+  )
+
+  it.effect("terminates WebSocket control events without waiting for the socket to close", () =>
+    Effect.gen(function* () {
+      const events = [
+        { type: "error", error: { code: "slow_down", message: "Try later" } },
+        {
+          type: "error",
+          status_code: 429,
+          message: "Rate limited",
+          headers: { "retry-after": 1, "x-request-id": "request", cached: false, invalid: [] },
+        },
+        {
+          type: "response.failed",
+          response: { error: { code: "server_error", message: "Unavailable" } },
+        },
+        { type: "error", status: "not-a-status", message: "Malformed status" },
+      ]
+
+      const errors = yield* Effect.forEach(events, (event) =>
+        LLMClient.generate(
+          LLM.request({
+            model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+              "gpt-4.1-mini",
+            ),
+            prompt: "Say hello.",
+          }),
+          {
+            webSocket: WebSocketTransport.makeDirect({
+              open: () =>
+                Effect.succeed({
+                  sendText: () => Effect.void,
+                  messages: Stream.make(ProviderShared.encodeJson(event)).pipe(Stream.concat(Stream.never)),
+                  close: Effect.void,
+                }),
+            }),
+          },
+        ).pipe(
+          Effect.provide(
+            LLMClient.layer.pipe(
+              Layer.provide(
+                Layer.succeed(
+                  RequestExecutor.Service,
+                  RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+                ),
+              ),
+            ),
+          ),
+          Effect.flip,
+        ),
+      )
+
+      expect(errors.map((error) => error.reason._tag)).toEqual([
+        "ProviderInternal",
+        "RateLimit",
+        "ProviderInternal",
+        "UnknownProvider",
+      ])
+    }),
+  )
+
+  it.effect("marks post-send WebSocket failures with delivery state", () =>
+    Effect.gen(function* () {
+      const failure = new AIError({
+        module: "test",
+        method: "receive",
+        reason: new TransportReason({ message: "socket closed", phase: "close" }),
+      })
+      const streams = [
+        Stream.fail(failure),
+        Stream.make(ProviderShared.encodeJson({ type: "response.created" })).pipe(Stream.concat(Stream.fail(failure))),
+      ]
+      const deps = Layer.succeed(
+        RequestExecutor.Service,
+        RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+      )
+      const webSocket = WebSocketTransport.makeDirect({
+        open: () =>
+          Effect.succeed({
+            sendText: () => Effect.void,
+            messages: streams.shift() ?? Stream.die("unexpected WebSocket open"),
+            close: Effect.void,
+          }),
+      })
+      const model = OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+        "gpt-4.1-mini",
+      )
+
+      const errors = yield* Effect.forEach(["first", "second"], (prompt) =>
+        LLMClient.generate(LLM.request({ model, prompt }), { webSocket }).pipe(
+          Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))),
+          Effect.flip,
+        ),
+      )
+
+      expect(errors.map((error) => error.reason)).toEqual([
+        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "ambiguous" }),
+        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "accepted" }),
+      ])
+    }),
+  )
+
   it.effect("fails immediately when WebSocket is already closed", () =>
     Effect.gen(function* () {
-      const error = yield* WebSocketExecutor.fromWebSocket(
+      const error = yield* WebSocketTransport.fromWebSocket(
         // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- fromWebSocket reads readyState before touching WebSocket methods on this branch.
         { readyState: globalThis.WebSocket.CLOSED } as globalThis.WebSocket,
         { url: "wss://api.openai.test/v1/responses", headers: Headers.empty },
       ).pipe(Effect.flip)
 
       expect(error.message).toContain("closed before opening")
+      expect(error.reason).toMatchObject({ _tag: "Transport", phase: "connect", delivery: "not-sent" })
     }),
   )
 

packages/ai/test/recorded-test.ts

@@ -2,12 +2,11 @@ import { HttpRecorder } from "@opencode-ai/http-recorder"
 import { Layer } from "effect"
 import * as path from "node:path"
 import { fileURLToPath } from "node:url"
-import { LLMClient, RequestExecutor, WebSocketExecutor } from "../src/route"
+import { LLMClient, RequestExecutor } from "../src/route"
 import { ImageClient } from "../src/image-client"
 import type { Service as ImageClientService } from "../src/image-client"
 import type { Service as LLMClientService } from "../src/route/client"
 import type { Service as RequestExecutorService } from "../src/route/executor"
-import type { Service as WebSocketExecutorService } from "../src/route/transport/websocket"
 import {
   recordedEffectGroup,
   type RecordedCaseOptions as RunnerCaseOptions,
@@ -17,7 +16,7 @@ import {
 const __dirname = path.dirname(fileURLToPath(import.meta.url))
 const FIXTURES_DIR = path.resolve(__dirname, "fixtures", "recordings")
 
-type RecordedEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService | ImageClientService
+type RecordedEnv = RequestExecutorService | LLMClientService | ImageClientService
 
 type RecordedTestsOptions = RecordedGroupOptions & {
   readonly options?: HttpRecorder.RecorderOptions
@@ -82,11 +81,10 @@ export const recordedTests = (options: RecordedTestsOptions) =>
           }),
         ),
       )
-      const deps = Layer.mergeAll(requestExecutor, WebSocketExecutor.layer)
       return Layer.mergeAll(
-        deps,
+        requestExecutor,
-        LLMClient.layer.pipe(Layer.provide(deps)),
+        LLMClient.layer.pipe(Layer.provide(requestExecutor)),
-        ImageClient.layer.pipe(Layer.provide(deps)),
+        ImageClient.layer.pipe(Layer.provide(requestExecutor)),
       )
     },
   })

packages/ai/test/schema.test.ts

@@ -11,6 +11,7 @@ import {
   LanguageModel,
   ModelID,
   ProviderID,
+  TransportReason,
   Usage,
 } from "../src/schema"
 import { ProviderShared } from "../src/protocols/shared"
@@ -108,3 +109,21 @@ test("AI errors expose the shared runtime tag", async () => {
     await Effect.runPromise(Effect.fail(error).pipe(Effect.catchTag("AI.Error", () => Effect.succeed("caught")))),
   ).toBe("caught")
 })
+
+test("transport errors serialize execution facts", () => {
+  const reason = new TransportReason({
+    message: "connection closed",
+    phase: "receive",
+    delivery: "ambiguous",
+    recovery: "fail",
+  })
+
+  expect(Schema.encodeSync(TransportReason)(reason)).toEqual({
+    _tag: "Transport",
+    message: "connection closed",
+    phase: "receive",
+    delivery: "ambiguous",
+    recovery: "fail",
+  })
+  expect(Schema.decodeUnknownSync(TransportReason)(Schema.encodeSync(TransportReason)(reason))).toEqual(reason)
+})

packages/core/src/aisdk.ts

@@ -319,7 +319,7 @@ function modelFromLanguage(info: Info, language: LanguageModelV3) {
     transport: {
       id: "ai-sdk",
       prepare: (input) => Effect.succeed(input.body),
-      frames: () => Stream.empty,
+      execute: () => Effect.succeed({ frames: Stream.empty }),
     },
     defaults: {
       headers: info.headers,

packages/core/src/session/runner/retry.ts

@@ -18,8 +18,9 @@ export function isRetryable(error: AIError) {
   switch (error.reason._tag) {
     case "RateLimit":
     case "ProviderInternal":
-    case "Transport":
       return true
+    case "Transport":
+      return error.reason.delivery === undefined || error.reason.delivery === "not-sent"
     case "InvalidProviderOutput":
       return error.reason.classification === "incomplete-stream"
     case "Authentication":

packages/core/test/session-error.test.ts

@@ -110,4 +110,26 @@ describe("toSessionError", () => {
     expect(eligible.map(SessionRunnerRetry.isRetryable)).toEqual([true, true, true])
     expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false, false, false, false, false, false])
   })
+
+  test("retries transport failures only when delivery is absent or not sent", () => {
+    const retryable = [
+      llm(new TransportReason({ message: "http transport" })),
+      llm(new TransportReason({ message: "connect failed", delivery: "not-sent", phase: "connect" })),
+    ]
+    const ineligible = [
+      llm(new TransportReason({ message: "send uncertain", delivery: "ambiguous", phase: "send" })),
+      llm(new TransportReason({ message: "response interrupted", delivery: "accepted", phase: "receive" })),
+      llm(
+        new TransportReason({
+          message: "continuation rejected",
+          delivery: "rejected",
+          recovery: "retry-full",
+          phase: "receive",
+        }),
+      ),
+    ]
+
+    expect(retryable.map(SessionRunnerRetry.isRetryable)).toEqual([true, true])
+    expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false, false])
+  })
 })