refactor(ai): unify OpenAI Responses transport

packages/ai/AGENTS.md

@@ -80,7 +80,7 @@ Route defaults are request-shaping defaults such as `headers`, `limits`, `genera
 
 The four-axis decomposition is the reason DeepSeek, TogetherAI, Cerebras, Baseten, Fireworks, and DeepInfra all reuse `OpenAIChat.protocol` verbatim — each provider deployment is a 5-15 line `Route.make(...)` call instead of a 300-400 line route clone. Bug fixes in one protocol propagate to every consumer of that protocol in a single commit.
 
-When a provider ships a non-HTTP transport (OpenAI's WebSocket Responses backend, hypothetical bidirectional streaming APIs), the seam is `Transport` — `WebSocketTransport.jsonTransport.with(...)` constructs an IO template whose `prepare` receives the route endpoint/auth at compile time, builds a WebSocket URL and message, and whose `frames` yields decoded text from the socket. Same protocol and endpoint source, different transport.
+When a provider supports multiple physical transports, selection remains execution policy below its semantic route. OpenAI Responses uses a purpose-built hybrid transport that prepares one final request, executes HTTP by default, and passes a generic channel exchange to a per-call `WebSocketChannelExecutor` when supplied. `Route.streamPrepared` owns decoding and acknowledges channel completion only after successful full consumption.
 
 ### URL Construction
 
@@ -106,7 +106,7 @@ const proxied = gateway.model("openai/gpt-4o-mini")
 Keep provider facades small and explicit:
 
 - Use branded `ProviderID.make(...)` and `ModelID.make(...)` where ids are constructed directly.
-- Use `model` for the default API path and named methods for provider-native alternatives such as OpenAI `responses`, `responsesWebSocket`, and `chat`.
+- Use `model` for the default API path and named methods for provider-native alternatives such as OpenAI `responses` and `chat`.
 - Put provider-specific setup on `.configure(...)`; do not add `model(id, overrides)` as a duplicate construction path.
 - Export lower-level `routes` arrays separately only when advanced internal wiring needs them.
 - Prefer `apiKey` as provider-specific sugar and `auth` as the explicit override; keep them mutually exclusive in provider option types with `ProviderAuthOption`.
@@ -124,11 +124,10 @@ import { model } from "@opencode-ai/ai/providers/openai/responses"
 
 const selected = model("gpt-5", {
   apiKey,
-  transport: "websocket",
 })
 ```
 
-Keep semantic APIs as separate entrypoints, such as OpenAI `chat` and `responses`. Keep transport choices inside the semantic entrypoint settings, so OpenAI Responses HTTP and WebSocket share one entrypoint. Provider facades may still expose named selectors such as `responsesWebSocket` for direct typed call sites; the package-like contract maps its settings to those selectors before returning an executable `LanguageModel`.
+Keep semantic APIs as separate entrypoints, such as OpenAI `chat` and `responses`. Transport is execution policy: OpenAI Responses uses HTTP by default and may receive a per-call WebSocket channel executor through `StreamOptions` without changing model or route identity.
 
 Do not expose `Route` in provider package settings. Route composition stays an implementation detail behind `model(...)`.
 
@@ -154,9 +153,10 @@ packages/ai/src/
     auth-options.ts         ProviderAuthOption shape, AuthOptions.bearer, AtLeastOne helper
     framing.ts              Framing type + Framing.sse
     transport/              transport implementations
-      index.ts              Transport type + HttpTransport / WebSocketTransport namespaces
+      index.ts              Transport execution types + HttpTransport / WebSocketTransport namespaces
+      websocket-channel.ts  generic sequential channel executor/driver contract
       http.ts               HttpTransport.httpJson — POST + framing
-      websocket.ts          WebSocketTransport.json + WebSocketExecutor service
+      websocket.ts          direct one-request channel executor + raw socket adapter
   protocols/
     shared.ts               ProviderShared toolkit used inside protocol impls
     openai-chat.ts          protocol + route (compose OpenAIChat.protocol)

packages/ai/README.md

@@ -315,7 +315,6 @@ import { model } from "@opencode-ai/ai/providers/openai/responses"
 
 const selected = model("gpt-5", {
   apiKey: process.env.OPENAI_API_KEY,
-  transport: "websocket",
   headers: { "x-application": "opencode" },
   limits: { context: 200_000, output: 64_000 },
 })
@@ -332,7 +331,7 @@ OpenAI Chat and OpenAI Responses are separate semantic entrypoints:
 - `@opencode-ai/ai/providers/google-vertex/responses`
 - `@opencode-ai/ai/providers/google-vertex/messages`
 
-Responses HTTP versus WebSocket is a scoped `transport` setting on the OpenAI Responses entrypoint, not another entrypoint. Azure follows the same Chat/Responses split at `providers/azure/chat` and `providers/azure/responses`. Generic OpenAI-compatible Chat remains at `providers/openai-compatible`; the Responses adapter at `providers/openai-compatible/responses` uses the provider-neutral Open Responses protocol. OpenAI Responses extends that baseline with OpenAI tools, event variants, metadata, defaults, and transports. Generic Anthropic Messages-compatible providers use `providers/anthropic-compatible`, which the named Anthropic provider composes. Google Gemini and Amazon Bedrock expose their single native API through their existing provider paths.
+OpenAI Responses has one semantic route and uses HTTP by default. Advanced callers may supply a per-call WebSocket channel executor through `StreamOptions`; transport policy does not change provider settings, model identity, or route identity. Azure follows the same Chat/Responses split at `providers/azure/chat` and `providers/azure/responses`. Generic OpenAI-compatible Chat remains at `providers/openai-compatible`; the Responses adapter at `providers/openai-compatible/responses` uses the provider-neutral Open Responses protocol. OpenAI Responses extends that baseline with OpenAI tools, event variants, metadata, and defaults. Generic Anthropic Messages-compatible providers use `providers/anthropic-compatible`, which the named Anthropic provider composes. Google Gemini and Amazon Bedrock expose their single native API through their existing provider paths.
 
 Vertex Gemini, Vertex Chat, Vertex Responses, and Vertex Messages are separate API entrypoints. All accept `project`, `location`, and an optional `accessToken`; when no explicit token or auth override is supplied they lazily use Google Application Default Credentials. Vertex Gemini instead selects express mode when `apiKey` or `GOOGLE_VERTEX_API_KEY` is present. Vertex Chat targets MaaS models through the OpenAI-compatible Chat Completions endpoint, while Vertex Responses targets Grok models and defaults `store` to `false` as required by Vertex. `providers/google-vertex` remains the default alias for `providers/google-vertex/gemini`.
 

packages/ai/STATUS.md

@@ -1,6 +1,6 @@
 # LLM Provider Parity Status
 
-Last reviewed: 2026-07-24
+Last reviewed: 2026-08-07
 
 This file tracks the gap between the native `@opencode-ai/ai` package and the AI SDK provider packages that opencode still depends on for many catalog/runtime paths.
 
@@ -16,8 +16,7 @@ This file tracks the gap between the native `@opencode-ai/ai` package and the AI
 | Native slice                       | Source                                                                                                                            | Current state                                                                                                                                                            | Main gaps                                                                                                                                                                                                                                                 |
 | ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | OpenAI Chat                        | `src/protocols/openai-chat.ts`, `src/providers/openai.ts`                                                                         | Usable. Streams text, reasoning deltas, tool calls, usage, images, and common generation controls.                                                                       | No typed structured-output / `response_format` path. Limited typed OpenAI option surface compared with SDK escape hatches.                                                                                                                                |
-| OpenAI Responses HTTP              | `src/protocols/open-responses.ts`, `src/protocols/openai-responses.ts`, `src/providers/openai.ts`                                 | Usable. Extends the Open Responses baseline with hosted-tool event surfacing, reasoning replay metadata, GPT-5 defaults, and cache usage.                                | No explicit `previous_response_id` path. Typed options cover only a subset of Responses fields. Structured output is still mostly synthetic-tool based.                                                                                                   |
+| OpenAI Responses                   | `src/protocols/open-responses.ts`, `src/protocols/openai-responses.ts`, `src/providers/openai.ts`                                 | Usable over HTTP by default, with optional per-call WebSocket channel execution on the same model and route identity.                                                    | No incremental `previous_response_id` path or persistent Session channel manager yet. Typed options cover only a subset of Responses fields. Structured output is still mostly synthetic-tool based.                                                      |
-| OpenAI Responses WebSocket         | `src/protocols/openai-responses.ts`, `src/route/transport/websocket.ts`                                                           | Present as `OpenAI.responsesWebSocket(...)`.                                                                                                                             | Runner/catalog support explicitly must not downgrade WebSocket routes; broader runtime selection is not complete.                                                                                                                                         |
 | OpenAI-compatible Chat             | `src/protocols/openai-compatible-chat.ts`, `src/providers/openai-compatible.ts`                                                   | Usable for generic Chat and several profiles: Baseten, Cerebras, DeepInfra, DeepSeek, Fireworks, Groq, TogetherAI.                                                       | Family quirks are mostly endpoint defaults, not full typed behavior.                                                                                                                                                                                      |
 | Open Responses-compatible          | `src/protocols/open-responses.ts`, `src/protocols/openai-compatible-responses.ts`, `src/providers/openai-compatible-responses.ts` | Usable for deployments that implement the provider-neutral Open Responses protocol. The deployment adapter does not inherit OpenAI tools, events, metadata, or defaults. | No named family profiles or recorded deployment coverage yet.                                                                                                                                                                                             |
 | Anthropic-compatible Messages      | `src/protocols/anthropic-messages.ts`, `src/providers/anthropic-compatible.ts`                                                    | Usable for deployments that implement the Anthropic Messages wire protocol. Named Anthropic composes this base; MiniMax M3 has recorded text and tool-loop coverage.     | No named compatible family profiles yet.                                                                                                                                                                                                                  |
@@ -48,19 +47,19 @@ Other `aisdk:` packages, including Google Vertex, Azure, and Bedrock, currently
 
 ## AI SDK Package Parity Matrix
 
-| AI SDK package                    | Intended native target                                         | Status           | Biggest gaps                                                                                                                                                                                                        |
+| AI SDK package                    | Intended native target                                    | Status           | Biggest gaps                                                                                                                                                                                                        |
-| --------------------------------- | -------------------------------------------------------------- | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| --------------------------------- | --------------------------------------------------------- | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `@ai-sdk/openai`                  | `OpenAI.chat`, `OpenAI.responses`, `OpenAI.responsesWebSocket` | Partial / usable | Add complete typed option coverage, structured output strategy, explicit Responses continuation support, and runner route selection between Chat/Responses/WebSocket.                                               |
+| `@ai-sdk/openai`                  | `OpenAI.chat`, `OpenAI.responses`                         | Partial / usable | Add complete typed option coverage, structured output strategy, explicit Responses continuation support, and runner execution policy for optional WebSocket channels.                                               |
-| `@ai-sdk/openai-compatible`       | Generic OpenAI-compatible Chat and Responses                   | Partial / usable | Decide per-family namespace/profile behavior and runner API selection for providers that support Responses versus Chat only.                                                                                        |
+| `@ai-sdk/openai-compatible`       | Generic OpenAI-compatible Chat and Responses              | Partial / usable | Decide per-family namespace/profile behavior and runner API selection for providers that support Responses versus Chat only.                                                                                        |
-| `@ai-sdk/anthropic`               | `AnthropicMessages`                                            | Partial / usable | Finish Messages API parity for headers/betas/metadata/newer fields and document hosted-tool continuation expectations.                                                                                              |
+| `@ai-sdk/anthropic`               | `AnthropicMessages`                                       | Partial / usable | Finish Messages API parity for headers/betas/metadata/newer fields and document hosted-tool continuation expectations.                                                                                              |
-| `@ai-sdk/google`                  | Gemini Developer API                                           | Partial / usable | Add typed options for safety, response schema/modalities, cached content, grounding/search/code execution, and non-text output modes where supported.                                                               |
+| `@ai-sdk/google`                  | Gemini Developer API                                      | Partial / usable | Add typed options for safety, response schema/modalities, cached content, grounding/search/code execution, and non-text output modes where supported.                                                               |
-| `@ai-sdk/google-vertex`           | Vertex Gemini namespace/facade                                 | Partial / usable | Add runner/catalog mapping, recorded coverage, and broader provider-option parity.                                                                                                                                  |
+| `@ai-sdk/google-vertex`           | Vertex Gemini namespace/facade                            | Partial / usable | Add runner/catalog mapping, recorded coverage, and broader provider-option parity.                                                                                                                                  |
-| `@ai-sdk/google-vertex/anthropic` | Anthropic Messages over Vertex namespace/facade                | Partial / usable | Add runner/catalog mapping, recorded coverage, and Vertex-specific hosted-tool parity.                                                                                                                              |
+| `@ai-sdk/google-vertex/anthropic` | Anthropic Messages over Vertex namespace/facade           | Partial / usable | Add runner/catalog mapping, recorded coverage, and Vertex-specific hosted-tool parity.                                                                                                                              |
-| `@ai-sdk/google-vertex/maas`      | Vertex Chat                                                    | Partial / usable | Add runner/catalog mapping, recorded coverage, and MaaS family-specific request parity.                                                                                                                             |
+| `@ai-sdk/google-vertex/maas`      | Vertex Chat                                               | Partial / usable | Add runner/catalog mapping, recorded coverage, and MaaS family-specific request parity.                                                                                                                             |
-| `@ai-sdk/google-vertex/xai`       | Vertex Chat / Responses                                        | Partial / usable | Decide Chat/Responses selection for catalog models, add runner mapping and recorded coverage, and review xAI-specific request options.                                                                              |
+| `@ai-sdk/google-vertex/xai`       | Vertex Chat / Responses                                   | Partial / usable | Decide Chat/Responses selection for catalog models, add runner mapping and recorded coverage, and review xAI-specific request options.                                                                              |
-| `@ai-sdk/azure`                   | Azure OpenAI Chat/Responses facade                             | Partial          | Map runner/catalog metadata to native Azure, handle resourceName/baseURL/apiVersion variants, add AAD/token auth story, and verify Chat vs Responses deployment selection.                                          |
+| `@ai-sdk/azure`                   | Azure OpenAI Chat/Responses facade                        | Partial          | Map runner/catalog metadata to native Azure, handle resourceName/baseURL/apiVersion variants, add AAD/token auth story, and verify Chat vs Responses deployment selection.                                          |
-| `@ai-sdk/amazon-bedrock`          | Bedrock Converse                                               | Partial          | Add default AWS credential chain/profile support, region/inference-profile model ID handling, provider option parity via `additionalModelRequestFields`, guardrails/performance config, and runner/catalog mapping. |
+| `@ai-sdk/amazon-bedrock`          | Bedrock Converse                                          | Partial          | Add default AWS credential chain/profile support, region/inference-profile model ID handling, provider option parity via `additionalModelRequestFields`, guardrails/performance config, and runner/catalog mapping. |
-| `@ai-sdk/amazon-bedrock/mantle`   | Bedrock Mantle OpenAI-compatible Chat/Responses namespace      | Partial / usable | Add default AWS credential chain/profile support; native catalog mapping currently requires bearer auth or explicit static credentials.                                                                             |
+| `@ai-sdk/amazon-bedrock/mantle`   | Bedrock Mantle OpenAI-compatible Chat/Responses namespace | Partial / usable | Add default AWS credential chain/profile support; native catalog mapping currently requires bearer auth or explicit static credentials.                                                                             |
 
 ## Highest-Risk Gaps
 

packages/ai/example/call-sites.md

@@ -67,7 +67,6 @@ Examples:
 ```ts
 OpenAI.responses("gpt-4o")
 OpenAI.chat("gpt-4o")
-OpenAI.responsesWebSocket("gpt-4o")
 
 Azure.configure({ resourceName, apiKey }).responses("my-deployment")
 AmazonBedrock.configure({ region, credentials }).model("anthropic.claude-3-5-sonnet-20241022-v2:0")
@@ -250,11 +249,6 @@ const openAIChat = Route.make({
   auth: Auth.envBearer("OPENAI_API_KEY"),
 })
 
-const openAIResponsesWebSocket = openAIResponses.with({
-  id: "openai-responses-websocket",
-  transport: WebSocketTransport.json,
-})
-
 const openAIConfig = (input: OpenAIConfig) => ({
   endpoint: input.endpoint,
   auth: input.auth ?? (input.apiKey ? Auth.bearer(input.apiKey) : undefined),
@@ -266,13 +260,11 @@ const openAIConfig = (input: OpenAIConfig) => ({
 
 const configureOpenAI = (input: OpenAIConfig = {}) => {
   const responses = openAIResponses.with(openAIConfig(input))
-  const responsesWebSocket = openAIResponsesWebSocket.with(openAIConfig(input))
   const chat = openAIChat.with(openAIConfig(input))
 
   return {
     id: openAIProvider,
     responses: responses.model,
-    responsesWebSocket: responsesWebSocket.model,
     chat: chat.model,
     model: responses.model,
     configure: configureOpenAI,
@@ -342,22 +334,19 @@ const response =
   )
 ```
 
-For direct provider-facade calls, HTTP versus WebSocket is represented as named
+For direct provider-facade calls, Responses has one semantic model and route:
-route selectors, not as model or request overrides. Same protocol, different
-transport, different route:
 
 ```ts
 OpenAI.responses("gpt-4o")
-OpenAI.responsesWebSocket("gpt-4o")
 ```
 
-The package-like OpenAI Responses entrypoint instead keeps transport scoped to
+The package-like OpenAI Responses entrypoint has the same transport-neutral
-Responses settings while preserving the same `model(...)` contract:
+`model(...)` contract:
 
 ```ts
 import { model } from "@opencode-ai/ai/providers/openai/responses"
 
-model("gpt-4o", { apiKey, transport: "websocket" })
+model("gpt-4o", { apiKey })
 ```
 
 Vertex keeps Gemini, Chat, Responses, and Messages as separate package-like entrypoints,
@@ -499,16 +488,13 @@ generic dynamic resolver:
 const model =
   providerID === "azure"
     ? Azure.configure(resolvedAzureConfig).responses(apiModelID)
-    : endpoint.websocket
+    : OpenAI.responses(apiModelID)
-      ? OpenAI.responsesWebSocket(apiModelID)
-      : OpenAI.responses(apiModelID)
 ```
 
 That boundary can branch on durable config/catalog metadata and call typed
-provider APIs directly. A direct provider-facade boundary maps metadata like
+provider APIs directly. Transport selection remains execution policy: a Session
-`endpoint.websocket` to `OpenAI.responsesWebSocket(apiModelID)`. A package-loading
+or other caller may pass a WebSocket channel executor per call without changing
-boundary passes `transport: "websocket"` to the OpenAI Responses entrypoint.
+the model constructed by this boundary.
-The client runtime only executes the route carried by the resulting model.
 
 ## Competitive Shape
 
@@ -544,9 +530,8 @@ App boundary = explicit durable-config -> typed-provider call
   id.
 - No `model(id, overrides)` escape hatch. Model selection takes the model id;
   endpoint/auth/deployment customization happens by configuring the route first.
-- No transport override on an executable model or request. Direct provider
+- No transport setting on a provider or executable model. OpenAI Responses uses
-  facades use `responses` versus `responsesWebSocket`; the package-like Responses
+  HTTP by default and accepts an optional per-call channel executor as execution policy.
-  entrypoint maps its scoped `transport` setting before constructing the model.
 - No separate public `LLMClient.layerWithWebSocket`. The runtime should expose one
   client layer with the available transport capabilities.
 - No executable `ModelRef`. The executable handle is `LanguageModel`; durable model
@@ -580,12 +565,10 @@ App boundary = explicit durable-config -> typed-provider call
 - [x] Make unconfigured transports reusable constants such as
       `HttpTransport.sseJson`; keep transport functions only for configured/fresh
       state construction.
-- [x] Collapse the public WebSocket runtime split so one `LLMClient.layer`
+- [x] Collapse the public WebSocket runtime split so one `LLMClient.layer` accepts
-      exposes available transport capabilities and selected routes fail with typed
+      optional per-call channel execution without changing route identity.
-      transport config errors when a required capability is missing.
 - [x] Convert OpenAI provider APIs to provider-facade shape:
-      `OpenAI.configure(config).responses(id)`, `.chat(id)`, and
+      `OpenAI.configure(config).responses(id)` and `.chat(id)`.
-      `.responsesWebSocket(id)`.
 - [x] Convert Azure to a configured facade where resource/base URL/api version
       setup happens before selecting deployment ids.
 - [x] Split Cloudflare products into separate facades such as

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)
+
+export 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 => {
@@ -966,16 +1001,24 @@ const providerErrorMessage = (event: Event, fallback: string): string => {
   return message || code || fallback
 }
 
-const providerError = (state: ParserState, event: Event, fallback: string) => {
+export const providerFailure = (id: string, 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,
+    module: id,
     method: "stream",
-    reason: classifyProviderFailure({ message, code }),
+    reason: classifyProviderFailure({ message, code, status }),
   })
 }
 
+const providerError = (state: ParserState, event: Event, fallback: string) => providerFailure(state.id, event, fallback)
+
 export const step = (state: ParserState, event: Event) => {
   if (event.type === "response.output_text.delta" || event.type === "response.output_text.done") {
     if (!event.item_id) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
@@ -1015,7 +1058,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/protocols/openai-responses-channel.ts

@@ -0,0 +1,37 @@
+import { Effect, Schema } from "effect"
+import type { WebSocketChannelDriver } from "../route/transport"
+import * as ProviderShared from "./shared"
+import { OpenResponses } from "./open-responses"
+
+const ADAPTER = "openai-responses"
+const NAME = "OpenAI Responses"
+const decodeEvent = Schema.decodeUnknownEffect(OpenResponses.protocol.stream.event)
+
+export const make = (message: string): WebSocketChannelDriver => ({
+  create: () => Effect.succeed({ message, mode: "full" }),
+  observe: (_create, frame) =>
+    Effect.gen(function* () {
+      const event = yield* decodeEvent(frame).pipe(
+        Effect.mapError(() => ProviderShared.eventError(ADAPTER, "Invalid OpenAI Responses WebSocket event", frame)),
+      )
+      if (event.type === "response.completed") return { type: "completed", frame }
+      if (event.type === "response.incomplete") return { type: "incomplete", frame }
+      if (event.type === "response.failed")
+        return {
+          type: "provider-failure",
+          error: OpenResponses.providerFailure(ADAPTER, event, `${NAME} response failed`),
+        }
+      if (event.type === "error") {
+        yield* OpenResponses.decodeKnownErrorEvent(event).pipe(
+          Effect.mapError(() => ProviderShared.eventError(ADAPTER, `${NAME} returned a malformed error event`, frame)),
+        )
+        return {
+          type: "provider-failure",
+          error: OpenResponses.providerFailure(ADAPTER, event, `${NAME} stream error`),
+        }
+      }
+      return { type: "frame", frame }
+    }),
+})
+
+export const OpenAIResponsesChannel = { make } as const

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

@@ -1,15 +1,24 @@
-import { Effect, Encoding, Schema } from "effect"
+import { Effect, Encoding, Schema, Stream } from "effect"
+import { Headers } from "effect/unstable/http"
 import { Route } from "../route/client"
 import { Auth } from "../route/auth"
 import { Endpoint } from "../route/endpoint"
+import { Framing } from "../route/framing"
 import { Protocol } from "../route/protocol"
-import { HttpTransport, WebSocketTransport } from "../route/transport"
+import {
+  HttpTransport,
+  WebSocketTransport,
+  type Transport,
+  type WebSocketChannelDriver,
+  type WebSocketChannelExchange,
+} from "../route/transport"
 import { LLMEvent, LLMRequest, type JsonSchema, type ToolDefinition } from "../schema"
 import { OpenResponses } from "./open-responses"
 import { optionalArray, ProviderShared } from "./shared"
 import { Lifecycle } from "./utils/lifecycle"
 import { OpenAIImage } from "./utils/openai-image"
 import { ToolSchemaProjection } from "./utils/tool-schema"
+import { OpenAIResponsesChannel } from "./openai-responses-channel"
 
 const ADAPTER = "openai-responses"
 const NAME = "OpenAI Responses"
@@ -250,17 +259,6 @@ const auth = Auth.none
 
 export const httpTransport = HttpTransport.sseJson.with<OpenAIResponsesBody>()
 
-export const route = Route.make({
-  id: ADAPTER,
-  provider: "openai",
-  providerMetadataKey: "openai",
-  protocol,
-  endpoint,
-  auth,
-  transport: httpTransport,
-  defaults: { providerOptions: { openai: { store: false } } },
-})
-
 const decodeWebSocketMessage = ProviderShared.validateWith(Schema.decodeUnknownEffect(OpenAIResponsesWebSocketMessage))
 
 const webSocketMessage = (body: OpenAIResponsesBody | Record<string, unknown>) =>
@@ -271,22 +269,58 @@ const webSocketMessage = (body: OpenAIResponsesBody | Record<string, unknown>) =
     return yield* decodeWebSocketMessage({ ...message, type: "response.create" })
   })
 
-export const webSocketTransport = WebSocketTransport.jsonTransport.with<
+export interface OpenAIResponsesPrepared {
-  OpenAIResponsesBody,
+  readonly http: HttpTransport.HttpPrepared<string>
-  OpenAIResponsesWebSocketMessage
+  readonly channel?: {
->({
+    readonly url: string
-  toMessage: webSocketMessage,
+    readonly headers: Headers.Headers
-  encodeMessage: encodeWebSocketMessage,
+    readonly driver: WebSocketChannelDriver
-})
+  }
+}
 
-export const webSocketRoute = Route.make({
+export const transport: Transport<OpenAIResponsesBody, OpenAIResponsesPrepared, string> = {
-  id: `${ADAPTER}-websocket`,
+  id: httpTransport.id,
+  prepare: (input) =>
+    Effect.gen(function* () {
+      const parts = yield* HttpTransport.jsonRequestParts(input)
+      return {
+        http: {
+          request: ProviderShared.jsonPost({ url: parts.url, body: parts.bodyText, headers: parts.headers }),
+          framing: Framing.sse,
+          middleware: input.middleware,
+        },
+        channel: input.webSocket
+          ? {
+              url: yield* WebSocketTransport.toWebSocketUrl(parts.url),
+              headers: parts.headers,
+              driver: OpenAIResponsesChannel.make(encodeWebSocketMessage(yield* webSocketMessage(parts.jsonBody))),
+            }
+          : undefined,
+      }
+    }),
+  execute: (prepared, request, runtime, options) => {
+    if (!options?.webSocket || !prepared.channel) return httpTransport.execute(prepared.http, request, runtime)
+    const exchange: WebSocketChannelExchange = {
+      id: request.id ?? "request",
+      connect: { url: prepared.channel.url, headers: prepared.channel.headers },
+      fallback: () =>
+        Stream.unwrap(
+          httpTransport.execute(prepared.http, request, runtime).pipe(Effect.map((execution) => execution.frames)),
+        ),
+      driver: prepared.channel.driver,
+    }
+    return options.webSocket.execute(exchange)
+  },
+}
+
+export const route = Route.make({
+  id: ADAPTER,
   provider: "openai",
   providerMetadataKey: "openai",
   protocol,
   endpoint,
   auth,
-  transport: webSocketTransport,
+  transport,
   defaults: { providerOptions: { openai: { store: false } } },
 })
 

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/providers/openai.ts

@@ -12,7 +12,7 @@ export type { OpenAIImageOptions } from "../protocols/openai-images"
 
 export const id = ProviderID.make("openai")
 
-export const routes = [OpenAIResponses.route, OpenAIResponses.webSocketRoute, OpenAIChat.route]
+export const routes = [OpenAIResponses.route, OpenAIChat.route]
 
 // This provider facade wraps the lower-level Responses and Chat model factories
 // with OpenAI-specific conveniences: typed options, API-key sugar, env fallback,
@@ -63,7 +63,6 @@ export interface Settings extends ProviderPackage.Settings {
   readonly organization?: string
   readonly project?: string
   readonly queryParams?: Readonly<Record<string, string>>
-  readonly transport?: "http" | "websocket"
   readonly providerOptions?: OpenAIProviderOptionsInput
 }
 
@@ -82,17 +81,12 @@ const configuredRoute = <Body, Prepared>(route: Route<Body, Prepared>, input: Co
 
 export const configure = (input: Config = {}) => {
   const responsesRoute = configuredRoute(OpenAIResponses.route, input)
-  const responsesWebSocketRoute = configuredRoute(OpenAIResponses.webSocketRoute, input)
   const chatRoute = configuredRoute(OpenAIChat.route, input)
   const modelDefaults = defaults(input)
   const responses = (id: string | ModelID) =>
     responsesRoute
       .with(withOpenAIOptions(id, modelDefaults, { textVerbosity: true }))
       .model<OpenAIProviderOptionsInput>({ id })
-  const responsesWebSocket = (id: string | ModelID) =>
-    responsesWebSocketRoute
-      .with(withOpenAIOptions(id, modelDefaults, { textVerbosity: true }))
-      .model<OpenAIProviderOptionsInput>({ id })
   const chat = (id: string | ModelID) =>
     chatRoute.with(withOpenAIOptions(id, modelDefaults)).model<OpenAIProviderOptionsInput>({ id })
   const image = (modelID: string | ModelID) =>
@@ -111,7 +105,6 @@ export const configure = (input: Config = {}) => {
     id,
     model: responses,
     responses,
-    responsesWebSocket,
     chat,
     image,
     configure,
@@ -138,10 +131,7 @@ const config = (settings: Settings): Config => {
 }
 
 export const model: ProviderPackage.Definition<Settings, OpenAIProviderOptionsInput>["model"] = (modelID, settings) => {
-  const configured = configure(config(settings))
+  return configure(config(settings)).responses(modelID)
-  if (settings.transport === undefined || settings.transport === "http") return configured.responses(modelID)
-  if (settings.transport === "websocket") return configured.responsesWebSocket(modelID)
-  throw new Error(`Unsupported OpenAI Responses transport: ${String(settings.transport)}`)
 }
 
 export const chatModel: ProviderPackage.Definition<Settings, OpenAIProviderOptionsInput>["model"] = (
@@ -149,6 +139,5 @@ export const chatModel: ProviderPackage.Definition<Settings, OpenAIProviderOptio
   settings,
 ) => configure(config(settings)).chat(modelID)
 export const responses = provider.responses
-export const responsesWebSocket = provider.responsesWebSocket
 export const chat = provider.chat
 export const image = provider.image

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)),
-        ),
-      ),
     )
   })
 
@@ -320,23 +314,29 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
           encodeBody,
           headers: routeInput.headers,
           middleware: options?.http,
+          webSocket: options?.webSocket,
         }),
-      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) => {
-            Stream.mapEffect(decodeEvent(route)),
+              const events = execution.frames.pipe(
-            protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
+                Stream.mapEffect(decodeEvent(route)),
-          )
+                protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
-        return events.pipe(
+              )
-          Stream.mapAccumEffect(
+              const stream = events.pipe(
-            () => protocol.stream.initial(request),
+                Stream.mapAccumEffect(
-            protocol.stream.step,
+                  () => protocol.stream.initial(request),
-            protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined,
+                  protocol.stream.step,
+                  protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined,
+                ),
+                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
+            }),
           ),
-          Stream.catchCause((cause) => Stream.fail(streamError(route, `Failed to read ${route} stream`, cause))),
-          requireTerminalEvent(route),
         )
       },
     } satisfies Route<Body, Prepared>
@@ -419,7 +419,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 +457,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,26 +86,28 @@ 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({
-      runtime.http
+      frames: Stream.unwrap(
-        .execute(prepared.request, prepared.middleware)
+        runtime.http
-        .pipe(
+          .execute(prepared.request, prepared.middleware)
-          Effect.map((response) =>
+          .pipe(
-            prepared.framing.frame(
+            Effect.map((response) =>
-              response.stream.pipe(
+              prepared.framing.frame(
-                Stream.mapError((error) =>
+                response.stream.pipe(
-                  ProviderShared.eventError(
+                  Stream.mapError((error) =>
-                    `${request.model.provider}/${request.model.route.id}`,
+                    ProviderShared.eventError(
-                    `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
+                      `${request.model.provider}/${request.model.route.id}`,
-                    ProviderShared.errorText(error),
+                      `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
+                      ProviderShared.errorText(error),
+                    ),
                   ),
                 ),
               ),
             ),
           ),
-        ),
+      ),
-    ),
+    }),
 })
 
 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> {
@@ -24,8 +38,19 @@ export interface TransportPrepareInput<Body> {
   readonly encodeBody: (body: Body) => string
   readonly headers?: (input: { readonly request: LLMRequest }) => Record<string, string>
   readonly middleware?: HttpMiddleware
+  readonly webSocket?: WebSocketChannelExecutor
 }
 
 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",
           }),
         ),
       )
@@ -101,7 +146,7 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
   })
 }
 
-const webSocketUrl = (value: string) =>
+export const toWebSocketUrl = (value: string) =>
   Effect.try({
     try: () => {
       const url = new URL(value)
@@ -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* () {
-    try: () =>
+    const constructor = yield* Socket.WebSocketConstructor
-      new (globalThis.WebSocket as unknown as WebSocketConstructorWithHeaders)(input.url, { headers: input.headers }),
+    const ws = yield* Effect.try({
-    catch: (error) =>
+      try: () =>
-      transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
+        // Platform implementations may extend Effect's browser-compatible constructor with handshake options.
-        url: input.url,
+        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
-        kind: "open",
+        (constructor as unknown as WebSocketConstructorWithHeaders)(input.url, {
-      }),
+          headers: input.headers,
-  }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
+        }),
-
+      catch: (error) =>
-export const layer: Layer.Layer<Service> = Layer.succeed(Service, Service.of({ open }))
+        transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
+          url: input.url,
+          kind: "open",
+          phase: "connect",
+          delivery: "not-sent",
+        }),
+    })
+    return yield* fromWebSocket(ws, input)
+  })
 
 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
@@ -232,32 +351,42 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
         ...prepareInput,
       })
       return {
-        url: yield* webSocketUrl(parts.url),
+        url: yield* toWebSocketUrl(parts.url),
         headers: parts.headers,
         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,13 @@ export const jsonTransport = {
   with: json,
 } as const
 
-export const WebSocketExecutor = {
-  Service,
-  layer,
-  open,
-  fromWebSocket,
-  messageText,
-} as const
-
 export const WebSocketTransport = {
   json,
   jsonTransport,
+  direct,
+  makeDirect,
+  open,
+  fromWebSocket,
+  messageText,
+  toWebSocketUrl,
 } 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,125 @@ describe("RequestExecutor", () => {
     }),
   )
 })
+
+describe("WebSocket channel execution", () => {
+  const model = OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("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 () => {
@@ -44,7 +45,6 @@ describe("public exports", () => {
 
     expect(OpenAI.model).toBeFunction()
     expect(OpenAI.provider.responses).toBe(OpenAI.responses)
-    expect(OpenAI.provider.responsesWebSocket).toBe(OpenAI.responsesWebSocket)
     expect(OpenAI.configure({ apiKey: "fixture" }).responses).toBeFunction()
     expect(OpenAICompatible.deepseek.model).toBeFunction()
     expect(
@@ -86,7 +86,6 @@ describe("public exports", () => {
     expect(OpenAICompatibleResponses.route.id).toBe("openai-compatible-responses")
     expect(OpenAICompatibleResponses.route.protocol).toBe("open-responses")
     expect(OpenAIResponses.route.id).toBe("openai-responses")
-    expect(OpenAIResponses.webSocketRoute.id).toBe("openai-responses-websocket")
     expect(AnthropicMessages.route.id).toBe("anthropic-messages")
   })
 })

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-options/openai.types.ts

@@ -1,13 +1,18 @@
 import { LLM } from "../../src"
 import { OpenAI } from "../../src/providers"
 
-const model = OpenAI.responses("gpt-5")
+const selected = OpenAI.responses("gpt-5")
 
-LLM.request({ model, prompt: "Hello", providerOptions: { openai: { reasoningEffort: "high" } } })
+LLM.request({ model: selected, prompt: "Hello", providerOptions: { openai: { reasoningEffort: "high" } } })
 
 LLM.request({
-  model,
+  model: selected,
   prompt: "Hello",
   // @ts-expect-error OpenAI reasoning effort must be a string.
   providerOptions: { openai: { reasoningEffort: 1 } },
 })
+
+OpenAI.configure({
+  // @ts-expect-error Transport is execution policy, not provider configuration.
+  transport: "websocket",
+})

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

@@ -80,11 +80,6 @@ describe("provider package entrypoints", () => {
     expect(selected.route.defaults.limits).toEqual({ context: 200_000, output: 64_000 })
   })
 
-  test("selects transport without changing the semantic API", () => {
-    expect(model("gpt-5", { apiKey: "fixture" }).route.id).toBe("openai-responses")
-    expect(model("gpt-5", { apiKey: "fixture", transport: "websocket" }).route.id).toBe("openai-responses-websocket")
-  })
-
   test("maps OpenAI-compatible Responses settings onto the executable model", async () => {
     const OpenAICompatibleResponses = await import("@opencode-ai/ai/providers/openai-compatible/responses")
     const selected = OpenAICompatibleResponses.model("custom-model", {

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

@@ -1,9 +1,10 @@
 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,
   AIError,
+  HttpOptions,
   LLMEvent,
   LLMRequest,
   Message,
@@ -11,9 +12,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"
@@ -216,19 +218,19 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
-  it.effect("prepares OpenAI Responses WebSocket target", () =>
+  it.effect("prepares one OpenAI Responses route for either transport", () =>
     Effect.gen(function* () {
       const prepared = yield* compileRequest(
         LLMRequest.update(request, {
-          model: OpenAIResponses.webSocketRoute
+          model: OpenAIResponses.route
             .with({ endpoint: { baseURL: "https://api.openai.test/v1/" }, auth: Auth.bearer("test") })
             .model({ id: "gpt-4.1-mini" }),
         }),
       )
 
-      expect(prepared.route).toBe("openai-responses-websocket")
+      expect(prepared.route).toBe("openai-responses")
       expect(prepared.protocol).toBe("openai-responses")
-      expect(prepared.metadata).toEqual({ transport: "websocket-json" })
+      expect(prepared.metadata).toEqual({ transport: "http-json" })
       expect(prepared.body).toMatchObject({ model: "gpt-4.1-mini", store: false, stream: true })
     }),
   )
@@ -238,41 +240,35 @@ 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,
+        RequestExecutor.Service.of({
-          RequestExecutor.Service.of({
+          execute: () => Effect.die("unexpected HTTP request"),
-            execute: () => Effect.die("unexpected HTTP request"),
+        }),
-          }),
-        ),
-        Layer.succeed(
-          WebSocketExecutor.Service,
-          WebSocketExecutor.Service.of({
-            open: (input) =>
-              Effect.succeed({
-                sendText: (message) =>
-                  Effect.sync(() => {
-                    opened.push({ url: input.url, authorization: input.headers.authorization })
-                    sent.push(message)
-                  }),
-                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: Effect.sync(() => {
-                  closed = true
-                }),
-              }),
-          }),
-        ),
       )
+      const webSocket = WebSocketTransport.makeDirect({
+        open: (input) =>
+          Effect.succeed({
+            sendText: (message) =>
+              Effect.sync(() => {
+                opened.push({ url: input.url, authorization: input.headers.authorization })
+                sent.push(message)
+              }),
+            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: Effect.sync(() => {
+              closed = true
+            }),
+          }),
+      })
       const response = yield* LLMClient.generate(
         LLM.request({
-          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("gpt-4.1-mini"),
-            "gpt-4.1-mini",
-          ),
           prompt: "Say hello.",
         }),
+        { webSocket },
       ).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))))
 
       expect(response.text).toBe("Hi")
@@ -288,15 +284,235 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("builds WebSocket and HTTP fallback from the same final request", () =>
+    Effect.gen(function* () {
+      const attempts = yield* Ref.make(0)
+      const message = yield* Ref.make("")
+      const body = yield* Ref.make("")
+      const response = yield* LLMClient.generate(
+        LLM.request({
+          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("gpt-4.1-mini"),
+          prompt: "Say hello.",
+          http: {
+            body: { model: "overlaid-model", metadata: { source: "overlay" } },
+            headers: { "x-request": "request" },
+            query: { mode: "test" },
+          },
+        }),
+        {
+          webSocket: {
+            execute: (exchange) =>
+              Effect.gen(function* () {
+                yield* exchange.driver
+                  .create(undefined)
+                  .pipe(Effect.flatMap((create) => Ref.set(message, create.message)))
+                return { frames: exchange.fallback(), complete: Effect.void }
+              }),
+          },
+        },
+      ).pipe(
+        Effect.provide(
+          dynamicResponse((input) =>
+            Effect.gen(function* () {
+              yield* Ref.update(attempts, (value) => value + 1)
+              yield* Ref.set(body, input.text)
+              expect(input.request.url).toBe("https://api.openai.test/v1/responses?mode=test")
+              expect(input.request.headers.authorization).toBe("Bearer test")
+              expect(input.request.headers["x-request"]).toBe("request")
+              return input.respond(sseEvents({ type: "response.completed", response: {} }), {
+                headers: { "content-type": "text/event-stream" },
+              })
+            }),
+          ),
+        ),
+      )
+
+      const httpBody = JSON.parse(yield* Ref.get(body))
+      const { stream: _stream, ...shared } = httpBody
+      expect(response.finishReason?.normalized).toBe("stop")
+      expect(yield* Ref.get(attempts)).toBe(1)
+      expect(JSON.parse(yield* Ref.get(message))).toEqual({ type: "response.create", ...shared })
+      expect(httpBody).toMatchObject({
+        model: "overlaid-model",
+        metadata: { source: "overlay" },
+        stream: true,
+      })
+    }),
+  )
+
+  it.effect("uses exactly one HTTP request when no WebSocket executor is supplied", () =>
+    Effect.gen(function* () {
+      const attempts = yield* Ref.make(0)
+      yield* LLMClient.generate(
+        LLMRequest.update(request, { http: new HttpOptions({ body: { input: "raw-http-input" } }) }),
+      ).pipe(
+        Effect.provide(
+          dynamicResponse((input) =>
+            Effect.gen(function* () {
+              yield* Ref.update(attempts, (value) => value + 1)
+              expect(JSON.parse(input.text).input).toBe("raw-http-input")
+              return input.respond(sseEvents({ type: "response.completed", response: {} }), {
+                headers: { "content-type": "text/event-stream" },
+              })
+            }),
+          ),
+        ),
+      )
+
+      expect(yield* Ref.get(attempts)).toBe(1)
+    }),
+  )
+
+  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" }).responses("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" }).responses(
+              "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" }).responses(
+        "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])
+  })
 })