refactor(core): adapt ai sdk transport execution

refactor(ai): add sequential websocket channel seam
fix(ai): harden websocket error contracts
2026-08-09 10:59:49 -04:00 · 2026-08-07 17:18:48 +05:30 · 2026-08-07 17:17:25 +05:30 · 2026-08-05 23:04:53 +05:30
33 changed files with 959 additions and 375 deletions
@@ -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(llmDeps))
+const llmClientLayer = LLMClient.layer.pipe(Layer.provide(requestExecutorLayer))

 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)
@@ -237,7 +237,7 @@ export type AnthropicMessagesBody = Schema.Schema.Type<typeof AnthropicMessagesB

 const AnthropicUsage = Schema.StructWithRest(
  Schema.Struct({
-    input_tokens: optionalNull(Schema.Number),
+    input_tokens: Schema.optional(Schema.Number),
    output_tokens: Schema.optional(Schema.Number),
    cache_creation_input_tokens: optionalNull(Schema.Number),
    cache_read_input_tokens: optionalNull(Schema.Number),
@@ -684,7 +684,7 @@ const mapFinishReason = (reason: string | null | undefined): FinishReason => {
 // expose that subset through `output_tokens_details.thinking_tokens`.
 const mapUsage = (usage: AnthropicUsage | undefined): Usage | undefined => {
  if (!usage) return undefined
-  const nonCached = usage.input_tokens ?? undefined
+  const nonCached = usage.input_tokens
  const cacheRead = usage.cache_read_input_tokens ?? undefined
  const cacheWrite = usage.cache_creation_input_tokens ?? undefined
  const inputTokens = ProviderShared.sumTokens(nonCached, cacheRead, cacheWrite)
@@ -20,7 +20,7 @@ import {
 } from "../schema"
 import { BedrockEventStream } from "./bedrock-event-stream"
 import { classifyProviderFailure } from "../provider-error"
-import { JsonObject, optionalArray, optionalNull, ProviderShared } from "./shared"
+import { JsonObject, optionalArray, ProviderShared } from "./shared"
 import { BedrockAuth } from "./utils/bedrock-auth"
 import { BedrockCache } from "./utils/bedrock-cache"
 import { BedrockMedia } from "./utils/bedrock-media"
@@ -150,8 +150,8 @@ const BedrockUsageSchema = Schema.Struct({
  inputTokens: Schema.optional(Schema.Number),
  outputTokens: Schema.optional(Schema.Number),
  totalTokens: Schema.optional(Schema.Number),
-  cacheReadInputTokens: optionalNull(Schema.Number),
-  cacheWriteInputTokens: optionalNull(Schema.Number),
+  cacheReadInputTokens: Schema.optional(Schema.Number),
+  cacheWriteInputTokens: Schema.optional(Schema.Number),
 })
 type BedrockUsageSchema = Schema.Schema.Type<typeof BedrockUsageSchema>

@@ -206,9 +206,9 @@ const BedrockEvent = Schema.Struct({
      additionalModelResponseFields: Schema.optional(Schema.Unknown),
    }),
  ),
-  metadata: optionalNull(
+  metadata: Schema.optional(
    Schema.Struct({
-      usage: optionalNull(BedrockUsageSchema),
+      usage: Schema.optional(BedrockUsageSchema),
      metrics: Schema.optional(Schema.Unknown),
    }),
  ),
@@ -464,21 +464,19 @@ const mapFinishReason = (reason: string): FinishReason => {

 // AWS reports inputTokens separately from cache reads and writes.
 // Bedrock does not break reasoning out of outputTokens for current models.
-const mapUsage = (usage: BedrockUsageSchema | null | undefined): Usage | undefined => {
+const mapUsage = (usage: BedrockUsageSchema | undefined): Usage | undefined => {
  if (!usage) return undefined
-  const cacheRead = usage.cacheReadInputTokens ?? undefined
-  const cacheWrite = usage.cacheWriteInputTokens ?? undefined
  const inputTokens = ProviderShared.sumTokens(
    usage.inputTokens,
-    cacheRead,
-    cacheWrite,
+    usage.cacheReadInputTokens,
+    usage.cacheWriteInputTokens,
  )
  return new Usage({
    inputTokens,
    outputTokens: usage.outputTokens,
    nonCachedInputTokens: usage.inputTokens,
-    cacheReadInputTokens: cacheRead,
-    cacheWriteInputTokens: cacheWrite,
+    cacheReadInputTokens: usage.cacheReadInputTokens,
+    cacheWriteInputTokens: usage.cacheWriteInputTokens,
    totalTokens: ProviderShared.totalTokens(inputTokens, usage.outputTokens, usage.totalTokens),
    providerMetadata: { bedrock: usage },
  })
@@ -76,12 +76,12 @@ const consumeFrames = (route: string) => (state: FrameBufferState, chunk: Uint8A
      // before handing the object to the chunk schema. JSON decode goes
      // through the shared Schema-driven codec to satisfy the package rule
      // against ad-hoc `JSON.parse` calls.
-      const parsed = yield* ProviderShared.parseJson(
+      const parsed = (yield* ProviderShared.parseJson(
        route,
        payload,
        "Failed to parse Bedrock Converse event-stream payload",
-      )
-      if (ProviderShared.isRecord(parsed)) delete parsed.p
+      )) as Record<string, unknown>
+      delete parsed.p
      out.push({ [eventType]: parsed })
    }
    return [cursor, out] as const
@@ -18,7 +18,7 @@ import {
  type ToolCallPart,
  type ToolDefinition,
 } from "../schema"
-import { JsonObject, optionalArray, optionalNull, ProviderShared } from "./shared"
+import { JsonObject, optionalArray, ProviderShared } from "./shared"
 import { GeminiToolSchema } from "./utils/gemini-tool-schema"
 import { Lifecycle } from "./utils/lifecycle"
 import { ToolSchemaProjection } from "./utils/tool-schema"
@@ -162,16 +162,13 @@ const GeminiBodyFields = {
 const GeminiBody = Schema.Struct(GeminiBodyFields)
 export type GeminiBody = Schema.Schema.Type<typeof GeminiBody>

-const GeminiUsage = Schema.StructWithRest(
-  Schema.Struct({
-    cachedContentTokenCount: optionalNull(Schema.Number),
-    thoughtsTokenCount: optionalNull(Schema.Number),
-    promptTokenCount: optionalNull(Schema.Number),
-    candidatesTokenCount: optionalNull(Schema.Number),
-    totalTokenCount: optionalNull(Schema.Number),
-  }),
-  [Schema.Record(Schema.String, Schema.Unknown)],
-)
+const GeminiUsage = Schema.Struct({
+  cachedContentTokenCount: Schema.optional(Schema.Number),
+  thoughtsTokenCount: Schema.optional(Schema.Number),
+  promptTokenCount: Schema.optional(Schema.Number),
+  candidatesTokenCount: Schema.optional(Schema.Number),
+  totalTokenCount: Schema.optional(Schema.Number),
+})
 type GeminiUsage = Schema.Schema.Type<typeof GeminiUsage>

 const GeminiCandidate = Schema.Struct({
@@ -181,7 +178,7 @@ const GeminiCandidate = Schema.Struct({

 const GeminiEvent = Schema.Struct({
  candidates: optionalArray(GeminiCandidate),
-  usageMetadata: optionalNull(GeminiUsage),
+  usageMetadata: Schema.optional(GeminiUsage),
 })
 type GeminiEvent = Schema.Schema.Type<typeof GeminiEvent>

@@ -425,25 +422,23 @@ const fromRequest = Effect.fn("Gemini.fromRequest")(function* (request: LLMReque
 // `cachedContentTokenCount` subset. `candidatesTokenCount` is *exclusive*
 // of `thoughtsTokenCount` — visible-only, not a total — so we sum the two
 // to produce the inclusive `outputTokens` the rest of the contract expects.
-const mapUsage = (usage: GeminiUsage | null | undefined) => {
+const mapUsage = (usage: GeminiUsage | undefined) => {
  if (!usage) return undefined
-  const input = usage.promptTokenCount ?? undefined
-  const cached = input === undefined ? undefined : (usage.cachedContentTokenCount ?? undefined)
-  const visible = usage.candidatesTokenCount ?? undefined
-  const thoughts = visible === undefined ? undefined : (usage.thoughtsTokenCount ?? undefined)
-  const nonCached = ProviderShared.subtractTokens(input, cached)
+  const cached = usage.cachedContentTokenCount
+  const nonCached = ProviderShared.subtractTokens(usage.promptTokenCount, cached)
  // `candidatesTokenCount` is visible-only; sum with thoughts to produce the
  // inclusive `outputTokens` the contract expects. Only compute the total
  // when the visible component is reported — otherwise we'd fabricate an
  // inclusive number from a partial breakdown.
-  const outputTokens = visible === undefined ? undefined : visible + (thoughts ?? 0)
+  const outputTokens =
+    usage.candidatesTokenCount !== undefined ? usage.candidatesTokenCount + (usage.thoughtsTokenCount ?? 0) : undefined
  return new Usage({
-    inputTokens: input,
+    inputTokens: usage.promptTokenCount,
    outputTokens,
    nonCachedInputTokens: nonCached,
    cacheReadInputTokens: cached,
-    reasoningTokens: thoughts,
-    totalTokens: ProviderShared.totalTokens(input, outputTokens, usage.totalTokenCount ?? undefined),
+    reasoningTokens: usage.thoughtsTokenCount,
+    totalTokens: ProviderShared.totalTokens(usage.promptTokenCount, outputTokens, usage.totalTokenCount),
    providerMetadata: { google: usage },
  })
 }
@@ -183,12 +183,12 @@ const OpenResponsesUsage = Schema.Struct({
  input_tokens: Schema.optional(Schema.Number),
  input_tokens_details: optionalNull(
    Schema.Struct({
-      cached_tokens: optionalNull(Schema.Number),
-      cache_write_tokens: optionalNull(Schema.Number),
+      cached_tokens: Schema.optional(Schema.Number),
+      cache_write_tokens: Schema.optional(Schema.Number),
    }),
  ),
  output_tokens: Schema.optional(Schema.Number),
-  output_tokens_details: optionalNull(Schema.Struct({ reasoning_tokens: optionalNull(Schema.Number) })),
+  output_tokens_details: optionalNull(Schema.Struct({ reasoning_tokens: Schema.optional(Schema.Number) })),
  total_tokens: Schema.optional(Schema.Number),
 })
 type OpenResponsesUsage = Schema.Schema.Type<typeof OpenResponsesUsage>
@@ -211,11 +211,43 @@ export type StreamItem = Schema.Schema.Type<typeof StreamItem>
 // event-level `error` envelope, so accept all three shapes here.
 // https://www.openresponses.org/specification
 const OpenResponsesErrorPayload = Schema.Struct({
+  type: optionalNull(Schema.String),
  code: optionalNull(Schema.String),
  message: optionalNull(Schema.String),
  param: optionalNull(Schema.String),
 })

+const WebSocketErrorHeader = Schema.Union([Schema.String, Schema.Number, Schema.Boolean])
+export const WebSocketErrorEvent = Schema.StructWithRest(
+  Schema.Struct({
+    type: Schema.tag("error"),
+    status: Schema.optional(Schema.Number),
+    status_code: Schema.optional(Schema.Number),
+    code: optionalNull(Schema.String),
+    message: Schema.optional(Schema.String),
+    param: optionalNull(Schema.String),
+    error: optionalNull(OpenResponsesErrorPayload),
+    headers: Schema.optional(Schema.Record(Schema.String, WebSocketErrorHeader)),
+  }),
+  [Schema.Record(Schema.String, Schema.Unknown)],
+)
+const decodeWebSocketErrorEvent = Schema.decodeUnknownEffect(WebSocketErrorEvent)
+
+const decodeKnownErrorEvent = (event: Event) =>
+  decodeWebSocketErrorEvent({
+    ...event,
+    status: typeof event.status === "number" ? event.status : undefined,
+    status_code: typeof event.status_code === "number" ? event.status_code : undefined,
+    headers: ProviderShared.isRecord(event.headers)
+      ? Object.fromEntries(
+          Object.entries(event.headers).filter(
+            (entry): entry is [string, string | number | boolean] =>
+              typeof entry[1] === "string" || typeof entry[1] === "number" || typeof entry[1] === "boolean",
+          ),
+        )
+      : undefined,
+  })
+
 export const Event = Schema.StructWithRest(
  Schema.Struct({
    type: Schema.String,
@@ -240,6 +272,9 @@ export const Event = Schema.StructWithRest(
    message: Schema.optional(Schema.String),
    param: optionalNull(Schema.String),
    error: optionalNull(OpenResponsesErrorPayload),
+    status: Schema.optional(Schema.Unknown),
+    status_code: Schema.optional(Schema.Unknown),
+    headers: Schema.optional(Schema.Unknown),
  }),
  [Schema.Record(Schema.String, Schema.Unknown)],
 )
@@ -592,11 +627,9 @@ export const fromRequest = Effect.fn("OpenResponses.fromRequest")(function* (req
 // non-cached breakdown.
 const mapUsage = (usage: OpenResponsesUsage | null | undefined, providerMetadataKey: string) => {
  if (!usage) return undefined
-  const cached = usage.input_tokens === undefined ? undefined : (usage.input_tokens_details?.cached_tokens ?? undefined)
-  const cacheWrite =
-    usage.input_tokens === undefined ? undefined : (usage.input_tokens_details?.cache_write_tokens ?? undefined)
-  const reasoning =
-    usage.output_tokens === undefined ? undefined : (usage.output_tokens_details?.reasoning_tokens ?? undefined)
+  const cached = usage.input_tokens_details?.cached_tokens
+  const cacheWrite = usage.input_tokens_details?.cache_write_tokens
+  const reasoning = usage.output_tokens_details?.reasoning_tokens
  const nonCached = ProviderShared.subtractTokens(usage.input_tokens, ProviderShared.sumTokens(cached, cacheWrite))
  return new Usage({
    inputTokens: usage.input_tokens,
@@ -634,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,
-// so keep this set aligned with `step` and the protocol's terminal predicate.
-const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"])
+// `finish` event; `response.failed` and `error` are hard failures. All four end
+// the stream, so keep this set aligned with `step` and the protocol's terminal predicate.
+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 => {
@@ -971,10 +1004,16 @@ const providerErrorMessage = (event: Event, fallback: string): string => {
 const providerError = (state: ParserState, event: Event, fallback: string) => {
  const code = event.code || event.error?.code || event.response?.error?.code || undefined
  const message = providerErrorMessage(event, fallback)
+  const status =
+    typeof event.status === "number"
+      ? event.status
+      : typeof event.status_code === "number"
+        ? event.status_code
+        : undefined
  return new AIError({
    module: state.id,
    method: "stream",
-    reason: classifyProviderFailure({ message, code }),
+    reason: classifyProviderFailure({ message, code, status }),
  })
 }

@@ -1017,7 +1056,11 @@ export const step = (state: ParserState, event: Event) => {
  if (event.type === "response.completed" || event.type === "response.incomplete")
    return Effect.succeed(onResponseFinish(state, event))
  if (event.type === "response.failed") return providerError(state, event, `${state.name} response failed`)
-  if (event.type === "error") return providerError(state, event, `${state.name} stream error`)
+  if (event.type === "error")
+    return decodeKnownErrorEvent(event).pipe(
+      Effect.mapError(() => ProviderShared.eventError(state.id, `${state.name} returned a malformed error event`)),
+      Effect.flatMap(() => providerError(state, event, `${state.name} stream error`)),
+    )
  return Effect.succeed<StepResult>([state, NO_EVENTS])
 }

@@ -146,18 +146,18 @@ export type OpenAIChatBody = Schema.Schema.Type<typeof OpenAIChatBody>
 // byte stream into strings, then `Protocol.jsonEvent` decodes each string into
 // this provider-native event shape.
 const OpenAIChatUsage = Schema.Struct({
-  prompt_tokens: optionalNull(Schema.Number),
-  completion_tokens: optionalNull(Schema.Number),
-  total_tokens: optionalNull(Schema.Number),
+  prompt_tokens: Schema.optional(Schema.Number),
+  completion_tokens: Schema.optional(Schema.Number),
+  total_tokens: Schema.optional(Schema.Number),
  prompt_tokens_details: optionalNull(
    Schema.Struct({
-      cached_tokens: optionalNull(Schema.Number),
-      cache_write_tokens: optionalNull(Schema.Number),
+      cached_tokens: Schema.optional(Schema.Number),
+      cache_write_tokens: Schema.optional(Schema.Number),
    }),
  ),
  completion_tokens_details: optionalNull(
    Schema.Struct({
-      reasoning_tokens: optionalNull(Schema.Number),
+      reasoning_tokens: Schema.optional(Schema.Number),
    }),
  ),
 })
@@ -168,7 +168,7 @@ const OpenAIChatToolCallDeltaFunction = Schema.Struct({
 })

 const OpenAIChatToolCallDelta = Schema.Struct({
-  index: optionalNull(Schema.Number),
+  index: Schema.Number,
  id: optionalNull(Schema.String),
  function: optionalNull(OpenAIChatToolCallDeltaFunction),
 })
@@ -559,20 +559,18 @@ const mapFinishReason = (reason: string | null | undefined): FinishReason => {
 // satisfied on both sides.
 const mapUsage = (usage: OpenAIChatEvent["usage"]): Usage | undefined => {
  if (!usage) return undefined
-  const input = usage.prompt_tokens ?? undefined
-  const output = usage.completion_tokens ?? undefined
-  const cached = input === undefined ? undefined : (usage.prompt_tokens_details?.cached_tokens ?? undefined)
-  const cacheWrite = input === undefined ? undefined : (usage.prompt_tokens_details?.cache_write_tokens ?? undefined)
-  const reasoning = output === undefined ? undefined : (usage.completion_tokens_details?.reasoning_tokens ?? undefined)
-  const nonCached = ProviderShared.subtractTokens(input, ProviderShared.sumTokens(cached, cacheWrite))
+  const cached = usage.prompt_tokens_details?.cached_tokens
+  const cacheWrite = usage.prompt_tokens_details?.cache_write_tokens
+  const reasoning = usage.completion_tokens_details?.reasoning_tokens
+  const nonCached = ProviderShared.subtractTokens(usage.prompt_tokens, ProviderShared.sumTokens(cached, cacheWrite))
  return new Usage({
-    inputTokens: input,
-    outputTokens: output,
+    inputTokens: usage.prompt_tokens,
+    outputTokens: usage.completion_tokens,
    nonCachedInputTokens: nonCached,
    cacheReadInputTokens: cached,
    cacheWriteInputTokens: cacheWrite,
    reasoningTokens: reasoning,
-    totalTokens: ProviderShared.totalTokens(input, output, usage.total_tokens ?? undefined),
+    totalTokens: ProviderShared.totalTokens(usage.prompt_tokens, usage.completion_tokens, usage.total_tokens),
    providerMetadata: { openai: usage },
  })
 }
@@ -696,25 +694,24 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
      lifecycle = Lifecycle.textDelta(lifecycle, events, "text-0", delta.content)
    }

-    for (const [position, tool] of toolDeltas.entries()) {
-      const index = tool.index ?? position
-      const current = tools[index]
-      const pending = pendingTools[index]
+    for (const tool of toolDeltas) {
+      const current = tools[tool.index]
+      const pending = pendingTools[tool.index]
      const id = current?.id ?? pending?.id ?? (tool.id || undefined)
      const name = current?.name ?? pending?.name ?? (tool.function?.name || undefined)
      const text = `${pending?.input ?? ""}${tool.function?.arguments ?? ""}`
      if (!current && (!id || !name)) {
-        pendingTools = { ...pendingTools, [index]: { id: id || undefined, name: name || undefined, input: text } }
+        pendingTools = { ...pendingTools, [tool.index]: { id: id || undefined, name: name || undefined, input: text } }
        continue
      }
      if (pending) {
        pendingTools = { ...pendingTools }
-        delete pendingTools[index]
+        delete pendingTools[tool.index]
      }
      const result = ToolStream.appendOrStart(
        ADAPTER,
        tools,
-        index,
+        tool.index,
        { id: id || undefined, name: name || undefined, text },
        "OpenAI Chat tool call delta is missing id or name",
      )
@@ -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"])
@@ -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 { WebSocketExecutor } from "./transport"
+import type { HttpMiddleware, Transport, TransportRuntime, WebSocketChannelExecutor } 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(
-        Effect.suspend(() =>
-          terminal
-            ? Effect.void
-            : Effect.fail(incompleteStreamError(route)),
-        ),
-      ),
+      Stream.onEnd(Effect.suspend(() => (terminal ? Effect.void : Effect.fail(incompleteStreamError(route))))),
    )
  })

@@ -321,22 +315,27 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
          headers: routeInput.headers,
          middleware: options?.http,
        }),
-      streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => {
+      streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime, options?: StreamOptions) => {
        const route = `${request.model.provider}/${request.model.route.id}`
-        const events = routeInput.transport
-          .frames(prepared, request, runtime)
-          .pipe(
-            Stream.mapEffect(decodeEvent(route)),
-            protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
-          )
-        return events.pipe(
-          Stream.mapAccumEffect(
-            () => protocol.stream.initial(request),
-            protocol.stream.step,
-            protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined,
+        return Stream.unwrap(
+          routeInput.transport.execute(prepared, request, runtime, options).pipe(
+            Effect.map((execution) => {
+              const events = execution.frames.pipe(
+                Stream.mapEffect(decodeEvent(route)),
+                protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
+              )
+              const stream = events.pipe(
+                Stream.mapAccumEffect(
+                  () => protocol.stream.initial(request),
+                  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 +418,7 @@ const streamRequestWith = (runtime: TransportRuntime) => (request: LLMRequest, o
  Stream.unwrap(
    Effect.gen(function* () {
      const compiled = yield* compile(request, options)
-      return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime)
+      return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime, options)
    }),
  )

@@ -457,7 +456,6 @@ export const layer: Layer.Layer<Service, never, RequestExecutor.Service> = Layer
  Effect.gen(function* () {
    const stream = streamRequestWith({
      http: yield* RequestExecutor.Service,
-      webSocket: Option.getOrUndefined(yield* Effect.serviceOption(WebSocketExecutor.Service)),
    })
    return Service.of({ stream, generate: generateWith(stream) })
  }),
@@ -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"
@@ -86,26 +86,28 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
        middleware: prepareInput.middleware,
      }
    }),
-  frames: (prepared, request, runtime) =>
-    Stream.unwrap(
-      runtime.http
-        .execute(prepared.request, prepared.middleware)
-        .pipe(
-          Effect.map((response) =>
-            prepared.framing.frame(
-              response.stream.pipe(
-                Stream.mapError((error) =>
-                  ProviderShared.eventError(
-                    `${request.model.provider}/${request.model.route.id}`,
-                    `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
-                    ProviderShared.errorText(error),
+  execute: (prepared, request, runtime) =>
+    Effect.succeed({
+      frames: Stream.unwrap(
+        runtime.http
+          .execute(prepared.request, prepared.middleware)
+          .pipe(
+            Effect.map((response) =>
+              prepared.framing.frame(
+                response.stream.pipe(
+                  Stream.mapError((error) =>
+                    ProviderShared.eventError(
+                      `${request.model.provider}/${request.model.route.id}`,
+                      `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
+                      ProviderShared.errorText(error),
+                    ),
                  ),
                ),
              ),
            ),
          ),
-        ),
-    ),
+      ),
+    }),
 })

 export const sseJson = {
@@ -1,19 +1,33 @@
-import type { Effect, Stream } from "effect"
+import type { Effect, Scope, Stream } from "effect"
 import { Endpoint } from "../endpoint"
 import { Auth } from "../auth"
 import type { HttpMiddleware, Interface as RequestExecutorInterface } from "../executor"
-import type { Interface as WebSocketExecutorInterface } from "./websocket"
+import type { WebSocketChannelExecutor } from "./websocket-channel"
 import type { AIError, LLMRequest } from "../../schema"

 export interface TransportRuntime {
  readonly http: RequestExecutorInterface
-  readonly webSocket?: WebSocketExecutorInterface
+}
+
+export interface TransportExecution<Frame> {
+  readonly frames: Stream.Stream<Frame, AIError>
+  /** Optional successful-consumption acknowledgement. HTTP leaves this absent. */
+  readonly complete?: Effect.Effect<void>
+}
+
+export interface TransportExecuteOptions {
+  readonly webSocket?: WebSocketChannelExecutor
 }

 export interface Transport<Body, Prepared, Frame> {
  readonly id: string
  readonly prepare: (input: TransportPrepareInput<Body>) => Effect.Effect<Prepared, AIError>
-  readonly frames: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => Stream.Stream<Frame, AIError>
+  readonly execute: (
+    prepared: Prepared,
+    request: LLMRequest,
+    runtime: TransportRuntime,
+    options?: TransportExecuteOptions,
+  ) => Effect.Effect<TransportExecution<Frame>, AIError, Scope.Scope>
 }

 export interface TransportPrepareInput<Body> {
@@ -28,4 +42,14 @@ export interface TransportPrepareInput<Body> {

 export * as HttpTransport from "./http"
 export type { HttpHandler, HttpMiddleware } from "../executor"
-export { WebSocketExecutor, WebSocketTransport } from "./websocket"
+export type {
+  ChannelCheckpoint,
+  ChannelCreate,
+  ChannelObservation,
+  WebSocketChannelDriver,
+  WebSocketChannelExchange,
+  WebSocketChannelExecution,
+  WebSocketChannelExecutor,
+} from "./websocket-channel"
+export type { WebSocketConnection, WebSocketConnector, WebSocketRequest } from "./websocket"
+export { WebSocketTransport } from "./websocket"
@@ -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
+}
@@ -1,8 +1,15 @@
-import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
+import { Cause, Effect, Queue, Stream } from "effect"
 import { Headers } from "effect/unstable/http"
+import { Socket } from "effect/unstable/socket"
 import { AIError, TransportReason } from "../../schema"
 import * as HttpTransport from "./http"
 import type { Transport } from "./index"
+import type {
+  ChannelObservation,
+  WebSocketChannelDriver,
+  WebSocketChannelExchange,
+  WebSocketChannelExecutor,
+} from "./websocket-channel"

 export interface WebSocketRequest {
  readonly url: string
@@ -15,28 +22,57 @@ export interface WebSocketConnection {
  readonly close: Effect.Effect<void, never>
 }

-export interface Interface {
+export interface WebSocketConnector {
  readonly open: (input: WebSocketRequest) => Effect.Effect<WebSocketConnection, AIError>
 }

-type WebSocketConstructorWithHeaders = new (
+type WebSocketConstructorWithHeaders = (
  url: string,
  options?: { readonly headers?: Headers.Headers },
 ) => globalThis.WebSocket

-export class Service extends Context.Service<Service, Interface>()("@opencode/AI/WebSocketExecutor") {}
-
 const transportError = (
  method: string,
  message: string,
-  input: { readonly url?: string; readonly kind?: string } = {},
+  input: {
+    readonly url?: string
+    readonly kind?: string
+    readonly phase?: TransportReason["phase"]
+    readonly delivery?: TransportReason["delivery"]
+  } = {},
 ) =>
  new AIError({
-    module: "WebSocketExecutor",
+    module: "WebSocketConnector",
    method,
-    reason: new TransportReason({ message, url: input.url, kind: input.kind }),
+    reason: new TransportReason({
+      message,
+      url: input.url,
+      kind: input.kind,
+      phase: input.phase,
+      delivery: input.delivery,
+    }),
  })

+const annotateTransportError = (
+  error: AIError,
+  input: { readonly phase: TransportReason["phase"]; readonly delivery: TransportReason["delivery"] },
+) =>
+  error.reason._tag === "Transport"
+    ? new AIError({
+        module: error.module,
+        method: error.method,
+        reason: new TransportReason({
+          message: error.reason.message,
+          kind: error.reason.kind,
+          url: error.reason.url,
+          http: error.reason.http,
+          phase: input.phase,
+          delivery: input.delivery,
+          recovery: error.reason.recovery,
+        }),
+      })
+    : error
+
 const eventMessage = (event: Event) => {
  if ("message" in event && typeof event.message === "string") return event.message
  return event.type
@@ -56,6 +92,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
      transportError("open", `WebSocket closed before opening (state ${ws.readyState})`, {
        url: input.url,
        kind: "open",
+        phase: "connect",
+        delivery: "not-sent",
      }),
    )
  }
@@ -79,7 +117,12 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
      cleanup()
      resume(
        Effect.fail(
-          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
+          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
+            url: input.url,
+            kind: "open",
+            phase: "connect",
+            delivery: "not-sent",
+          }),
        ),
      )
    }
@@ -90,6 +133,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
          transportError("open", `WebSocket closed before opening with code ${event.code}`, {
            url: input.url,
            kind: "open",
+            phase: "connect",
+            delivery: "not-sent",
          }),
        ),
      )
@@ -119,21 +164,31 @@ const webSocketUrl = (value: string) =>
      transportError("prepare", error instanceof Error ? error.message : "Invalid WebSocket URL", {
        url: value,
        kind: "websocket",
+        phase: "prepare",
+        delivery: "not-sent",
      }),
  })

 export const open = (input: WebSocketRequest) =>
-  Effect.try({
-    try: () =>
-      new (globalThis.WebSocket as unknown as WebSocketConstructorWithHeaders)(input.url, { headers: input.headers }),
-    catch: (error) =>
-      transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
-        url: input.url,
-        kind: "open",
-      }),
-  }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
-
-export const layer: Layer.Layer<Service> = Layer.succeed(Service, Service.of({ open }))
+  Effect.gen(function* () {
+    const constructor = yield* Socket.WebSocketConstructor
+    const ws = yield* Effect.try({
+      try: () =>
+        // Platform implementations may extend Effect's browser-compatible constructor with handshake options.
+        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
+        (constructor as unknown as WebSocketConstructorWithHeaders)(input.url, {
+          headers: input.headers,
+        }),
+      catch: (error) =>
+        transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
+          url: input.url,
+          kind: "open",
+          phase: "connect",
+          delivery: "not-sent",
+        }),
+    })
+    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
@@ -237,27 +356,37 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
        message: input.encodeMessage(yield* input.toMessage(parts.jsonBody)),
      }
    }),
-  frames: (prepared, _request, runtime) => {
-    const webSocket = runtime.webSocket
+  execute: (prepared, request, _runtime, options) => {
+    const webSocket = options?.webSocket
    if (!webSocket) {
-      return Stream.fail(
-        transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
+      return Effect.fail(
+        transportError("json", "WebSocket JSON transport requires StreamOptions.webSocket", {
          url: prepared.url,
          kind: "websocket",
+          phase: "prepare",
+          delivery: "not-sent",
        }),
      )
    }
-    const decoder = new TextDecoder()
-    return Stream.unwrap(
-      Effect.gen(function* () {
-        const connection = yield* Effect.acquireRelease(
-          webSocket.open({ url: prepared.url, headers: prepared.headers }),
-          (connection) => connection.close,
-        )
-        yield* connection.sendText(prepared.message)
-        return connection.messages.pipe(Stream.map((message) => messageText(message, decoder)))
-      }),
-    )
+    const driver: WebSocketChannelDriver = {
+      create: () => Effect.succeed({ message: prepared.message, mode: "full" }),
+      observe: (_create, frame) => Effect.succeed({ type: "frame", frame }),
+    }
+    const exchange: WebSocketChannelExchange = {
+      id: request.id ?? "request",
+      connect: { url: prepared.url, headers: prepared.headers },
+      fallback: () =>
+        Stream.fail(
+          transportError("fallback", "WebSocket JSON transport does not provide HTTP fallback", {
+            url: prepared.url,
+            kind: "websocket",
+            phase: "fallback",
+            delivery: "not-sent",
+          }),
+        ),
+      driver,
+    }
+    return webSocket.execute(exchange)
  },
 })

@@ -266,15 +395,12 @@ export const jsonTransport = {
  with: json,
 } as const

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

 const SSE_HEADERS = { "content-type": "text/event-stream" } as const
@@ -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", () => {
@@ -488,7 +488,7 @@ describe("Anthropic Messages route", () => {
        {
          type: "message_delta",
          delta: { stop_reason: "end_turn", stop_sequence: "\n\nHuman:" },
-          usage: { input_tokens: null, output_tokens: 2 },
+          usage: { output_tokens: 2 },
        },
        { type: "message_stop" },
      )
@@ -388,30 +388,12 @@ describe("Bedrock Converse route", () => {
    Effect.gen(function* () {
      const body = eventStreamBody(
        ["messageStop", { stopReason: "end_turn" }],
-        [
-          "metadata",
-          {
-            usage: {
-              inputTokens: 5,
-              outputTokens: 2,
-              totalTokens: 7,
-              cacheReadInputTokens: null,
-              cacheWriteInputTokens: null,
-            },
-          },
-        ],
-        ["metadata", { usage: null }],
-        ["metadata", null],
+        ["metadata", { usage: { inputTokens: 5, outputTokens: 2, totalTokens: 7 } }],
+        ["metadata", { metrics: { latencyMs: 100 } }],
      )
      const response = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body)))

-      expect(response.usage).toMatchObject({
-        inputTokens: 5,
-        outputTokens: 2,
-        totalTokens: 7,
-        cacheReadInputTokens: undefined,
-        cacheWriteInputTokens: undefined,
-      })
+      expect(response.usage).toMatchObject({ inputTokens: 5, outputTokens: 2, totalTokens: 7 })
    }),
  )

@@ -722,37 +722,14 @@ describe("Gemini route", () => {
    }),
  )

-  it.effect("keeps partial usage only in provider metadata", () =>
+  it.effect("leaves total usage undefined when component counts are missing", () =>
    Effect.gen(function* () {
      const response = yield* LLMClient.generate(request).pipe(
-        Effect.provide(
-          fixedResponse(
-            sseEvents({
-              usageMetadata: {
-                promptTokenCount: null,
-                candidatesTokenCount: null,
-                totalTokenCount: null,
-                thoughtsTokenCount: null,
-                cachedContentTokenCount: 1,
-                promptTokensDetails: [{ modality: "TEXT", tokenCount: 5 }],
-                candidatesTokensDetails: [{ modality: "TEXT", tokenCount: 2 }],
-              },
-            }),
-          ),
-        ),
+        Effect.provide(fixedResponse(sseEvents({ usageMetadata: { thoughtsTokenCount: 1 } }))),
      )

-      expect(response.usage).toMatchObject({
-        inputTokens: undefined,
-        outputTokens: undefined,
-        cacheReadInputTokens: undefined,
-        providerMetadata: {
-          google: {
-            promptTokensDetails: [{ modality: "TEXT", tokenCount: 5 }],
-            candidatesTokensDetails: [{ modality: "TEXT", tokenCount: 2 }],
-          },
-        },
-      })
+      expect(response.usage).toMatchObject({ reasoningTokens: 1 })
+      expect(response.usage?.totalTokens).toBeUndefined()
    }),
  )

@@ -596,37 +596,6 @@ describe("OpenAI Chat route", () => {
    }),
  )

-  it.effect("accepts nullable usage counters", () =>
-    Effect.gen(function* () {
-      const body = sseEvents(
-        deltaChunk({ role: "assistant", content: "Hello" }),
-        deltaChunk({}, "stop"),
-        usageChunk({
-          prompt_tokens: null,
-          completion_tokens: null,
-          total_tokens: null,
-          prompt_tokens_details: { cached_tokens: 1, cache_write_tokens: null },
-          completion_tokens_details: { reasoning_tokens: null },
-        }),
-      )
-      const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body)))
-
-      expect(response.usage).toEqual(
-        new Usage({
-          providerMetadata: {
-            openai: {
-              prompt_tokens: null,
-              completion_tokens: null,
-              total_tokens: null,
-              prompt_tokens_details: { cached_tokens: 1, cache_write_tokens: null },
-              completion_tokens_details: { reasoning_tokens: null },
-            },
-          },
-        }),
-      )
-    }),
-  )
-
  it.effect("parses and replays OpenAI-compatible reasoning fields", () =>
    Effect.gen(function* () {
      const fields = ["reasoning_content", "reasoning", "reasoning_text"] as const
@@ -1079,36 +1048,6 @@ describe("OpenAI Chat route", () => {
    }),
  )

-  it.effect("assembles indexless streamed tool calls", () =>
-    Effect.gen(function* () {
-      const body = sseEvents(
-        deltaChunk({
-          tool_calls: [
-            { id: "call_1", function: { name: "lookup", arguments: '{"query"' } },
-            { index: null, id: "call_2", function: { name: "lookup", arguments: '{"query"' } },
-          ],
-        }),
-        deltaChunk({
-          tool_calls: [
-            { function: { arguments: ':"weather"}' } },
-            { index: null, function: { arguments: ':"time"}' } },
-          ],
-        }),
-        deltaChunk({}, "tool_calls"),
-      )
-      const response = yield* LLMClient.generate(
-        LLMRequest.update(request, {
-          tools: [ToolDefinition.make({ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } })],
-        }),
-      ).pipe(Effect.provide(fixedResponse(body)))
-
-      expect(response.toolCalls).toMatchObject([
-        { id: "call_1", name: "lookup", input: { query: "weather" } },
-        { id: "call_2", name: "lookup", input: { query: "time" } },
-      ])
-    }),
-  )
-
  it.effect("ignores empty identity fields on later tool call deltas", () =>
    Effect.gen(function* () {
      const body = sseEvents(
@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { ConfigProvider, Effect, Layer, Stream } from "effect"
+import { ConfigProvider, Effect, Layer, Ref, Stream } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import {
  LLM,
@@ -11,9 +11,10 @@ import {
  ToolCallPart,
  ToolDefinition,
  ToolResultPart,
+  TransportReason,
  Usage,
 } from "../../src"
-import { Auth, LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route"
+import { Auth, LLMClient, RequestExecutor, WebSocketTransport } from "../../src/route"
 import { compileRequest } from "../../src/route/client"
 import * as Azure from "../../src/providers/azure"
 import * as OpenAI from "../../src/providers/openai"
@@ -238,34 +239,29 @@ describe("OpenAI Responses route", () => {
      const sent: string[] = []
      const opened: Array<{ readonly url: string; readonly authorization: string | undefined }> = []
      let closed = false
-      const deps = Layer.mergeAll(
-        Layer.succeed(
-          RequestExecutor.Service,
-          RequestExecutor.Service.of({
-            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 deps = Layer.succeed(
+        RequestExecutor.Service,
+        RequestExecutor.Service.of({
+          execute: () => Effect.die("unexpected HTTP request"),
+        }),
      )
+      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(
@@ -273,6 +269,7 @@ describe("OpenAI Responses route", () => {
          ),
          prompt: "Say hello.",
        }),
+        { webSocket },
      ).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))))

      expect(response.text).toBe("Hi")
@@ -288,15 +285,158 @@ describe("OpenAI Responses route", () => {
    }),
  )

+  it.effect("closes a direct WebSocket execution after partial consumption", () =>
+    Effect.gen(function* () {
+      const closed = yield* Ref.make(false)
+      const webSocket = WebSocketTransport.makeDirect({
+        open: () =>
+          Effect.succeed({
+            sendText: () => Effect.void,
+            messages: Stream.fromArray([
+              ProviderShared.encodeJson({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }),
+              ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_ws" } }),
+            ]),
+            close: Ref.set(closed, true),
+          }),
+      })
+
+      yield* LLMClient.stream(
+        LLM.request({
+          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+            "gpt-4.1-mini",
+          ),
+          prompt: "Say hello.",
+        }),
+        { webSocket },
+      ).pipe(
+        Stream.take(1),
+        Stream.runDrain,
+        Effect.provide(
+          LLMClient.layer.pipe(
+            Layer.provide(
+              Layer.succeed(
+                RequestExecutor.Service,
+                RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+              ),
+            ),
+          ),
+        ),
+      )
+
+      expect(yield* Ref.get(closed)).toBe(true)
+    }),
+  )
+
+  it.effect("terminates WebSocket control events without waiting for the socket to close", () =>
+    Effect.gen(function* () {
+      const events = [
+        { type: "error", error: { code: "slow_down", message: "Try later" } },
+        {
+          type: "error",
+          status_code: 429,
+          message: "Rate limited",
+          headers: { "retry-after": 1, "x-request-id": "request", cached: false, invalid: [] },
+        },
+        {
+          type: "response.failed",
+          response: { error: { code: "server_error", message: "Unavailable" } },
+        },
+        { type: "error", status: "not-a-status", message: "Malformed status" },
+      ]
+
+      const errors = yield* Effect.forEach(events, (event) =>
+        LLMClient.generate(
+          LLM.request({
+            model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+              "gpt-4.1-mini",
+            ),
+            prompt: "Say hello.",
+          }),
+          {
+            webSocket: WebSocketTransport.makeDirect({
+              open: () =>
+                Effect.succeed({
+                  sendText: () => Effect.void,
+                  messages: Stream.make(ProviderShared.encodeJson(event)).pipe(Stream.concat(Stream.never)),
+                  close: Effect.void,
+                }),
+            }),
+          },
+        ).pipe(
+          Effect.provide(
+            LLMClient.layer.pipe(
+              Layer.provide(
+                Layer.succeed(
+                  RequestExecutor.Service,
+                  RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+                ),
+              ),
+            ),
+          ),
+          Effect.flip,
+        ),
+      )
+
+      expect(errors.map((error) => error.reason._tag)).toEqual([
+        "ProviderInternal",
+        "RateLimit",
+        "ProviderInternal",
+        "UnknownProvider",
+      ])
+    }),
+  )
+
+  it.effect("marks post-send WebSocket failures with delivery state", () =>
+    Effect.gen(function* () {
+      const failure = new AIError({
+        module: "test",
+        method: "receive",
+        reason: new TransportReason({ message: "socket closed", phase: "close" }),
+      })
+      const streams = [
+        Stream.fail(failure),
+        Stream.make(ProviderShared.encodeJson({ type: "response.created" })).pipe(Stream.concat(Stream.fail(failure))),
+      ]
+      const deps = Layer.succeed(
+        RequestExecutor.Service,
+        RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+      )
+      const webSocket = WebSocketTransport.makeDirect({
+        open: () =>
+          Effect.succeed({
+            sendText: () => Effect.void,
+            messages: streams.shift() ?? Stream.die("unexpected WebSocket open"),
+            close: Effect.void,
+          }),
+      })
+      const model = OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+        "gpt-4.1-mini",
+      )
+
+      const errors = yield* Effect.forEach(["first", "second"], (prompt) =>
+        LLMClient.generate(LLM.request({ model, prompt }), { webSocket }).pipe(
+          Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))),
+          Effect.flip,
+        ),
+      )
+
+      expect(errors.map((error) => error.reason)).toEqual([
+        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "ambiguous" }),
+        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "accepted" }),
+      ])
+    }),
+  )
+
  it.effect("fails immediately when WebSocket is already closed", () =>
    Effect.gen(function* () {
-      const error = yield* WebSocketExecutor.fromWebSocket(
+      const error = yield* WebSocketTransport.fromWebSocket(
        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- fromWebSocket reads readyState before touching WebSocket methods on this branch.
        { readyState: globalThis.WebSocket.CLOSED } as globalThis.WebSocket,
        { url: "wss://api.openai.test/v1/responses", headers: Headers.empty },
      ).pipe(Effect.flip)

      expect(error.message).toContain("closed before opening")
+      expect(error.reason).toMatchObject({ _tag: "Transport", phase: "connect", delivery: "not-sent" })
    }),
  )

@@ -885,40 +1025,6 @@ describe("OpenAI Responses route", () => {
    }),
  )

-  it.effect("accepts nullable token details", () =>
-    Effect.gen(function* () {
-      const response = yield* LLMClient.generate(request).pipe(
-        Effect.provide(
-          fixedResponse(
-            sseEvents({
-              type: "response.completed",
-              response: {
-                id: "resp_1",
-                usage: {
-                  input_tokens: 5,
-                  output_tokens: 2,
-                  total_tokens: 7,
-                  input_tokens_details: { cached_tokens: null, cache_write_tokens: null },
-                  output_tokens_details: { reasoning_tokens: null },
-                },
-              },
-            }),
-          ),
-        ),
-      )
-
-      expect(response.usage).toMatchObject({
-        inputTokens: 5,
-        outputTokens: 2,
-        nonCachedInputTokens: 5,
-        totalTokens: 7,
-        cacheReadInputTokens: undefined,
-        cacheWriteInputTokens: undefined,
-        reasoningTokens: undefined,
-      })
-    }),
-  )
-
  it.effect("preserves and replays assistant message phases", () =>
    Effect.gen(function* () {
      const response = yield* LLMClient.generate(request).pipe(
@@ -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,
-        LLMClient.layer.pipe(Layer.provide(deps)),
-        ImageClient.layer.pipe(Layer.provide(deps)),
+        requestExecutor,
+        LLMClient.layer.pipe(Layer.provide(requestExecutor)),
+        ImageClient.layer.pipe(Layer.provide(requestExecutor)),
      )
    },
  })
@@ -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)
+})
@@ -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,
@@ -1,6 +1,7 @@
 export * as Git from "./git"

 import path from "path"
+import { randomUUID } from "crypto"
 import { Context, Effect, Layer, Schema, Stream } from "effect"
 import { ChildProcess } from "effect/unstable/process"
 import { AbsolutePath, RelativePath } from "./schema"
@@ -174,10 +175,17 @@ export interface Interface {
      context?: number
      paths?: readonly RelativePath[]
    }) => Effect.Effect<readonly File.Diff[], OperationError>
+    readonly preview: (input: {
+      repository: Repository
+      current: TreeID
+      files: ReadonlyMap<RelativePath, TreeID>
+      context?: number
+    }) => Effect.Effect<readonly File.Diff[], OperationError>
    readonly restore: (input: {
      repository: Repository
      files: ReadonlyMap<RelativePath, TreeID>
    }) => Effect.Effect<void, OperationError>
+    readonly checkout: (input: { repository: Repository; tree: TreeID }) => Effect.Effect<void, OperationError>
  }
 }

@@ -649,6 +657,58 @@ const layer = Layer.effect(
      return { mode: match[1], object: match[2] }
    })

+    const preview = Effect.fn("Git.tree.preview")(
+      (input: {
+        repository: Repository
+        current: TreeID
+        files: ReadonlyMap<RelativePath, TreeID>
+        context?: number
+      }) =>
+        locked(
+          input.repository,
+          Effect.gen(function* () {
+            const index = path.join(input.repository.gitDirectory, `preview-${randomUUID()}.index`)
+            const env = { GIT_INDEX_FILE: index }
+            return yield* Effect.gen(function* () {
+              yield* repositoryOperation("diff", input.repository, ["read-tree", input.current], { env })
+              yield* Effect.forEach(
+                input.files,
+                ([file, tree]) =>
+                  Effect.gen(function* () {
+                    const source = yield* entry(input.repository, tree, file)
+                    if (!source) {
+                      yield* repositoryOperation(
+                        "diff",
+                        input.repository,
+                        ["update-index", "--force-remove", "--", file],
+                        { env },
+                      )
+                      return
+                    }
+                    yield* repositoryOperation(
+                      "diff",
+                      input.repository,
+                      ["update-index", "--add", "--cacheinfo", source.mode, source.object, file],
+                      { env },
+                    )
+                  }),
+                { discard: true },
+              )
+              const target = TreeID.make(
+                (yield* repositoryOperation("diff", input.repository, ["write-tree"], { env })).text.trim(),
+              )
+              return yield* treeDiff({
+                repository: input.repository,
+                from: input.current,
+                to: target,
+                context: input.context,
+                paths: Array.from(input.files.keys()),
+              })
+            }).pipe(Effect.ensuring(fs.remove(index).pipe(Effect.catch(() => Effect.void))))
+          }),
+        ),
+    )
+
    const restore = Effect.fn("Git.tree.restore")(
      (input: { repository: Repository; files: ReadonlyMap<RelativePath, TreeID> }) =>
        locked(
@@ -678,6 +738,16 @@ const layer = Layer.effect(
        ),
    )

+    const checkoutTree = Effect.fn("Git.tree.checkout")((input: { repository: Repository; tree: TreeID }) =>
+      locked(
+        input.repository,
+        Effect.gen(function* () {
+          yield* repositoryOperation("restore", input.repository, ["read-tree", input.tree])
+          yield* repositoryOperation("restore", input.repository, ["checkout-index", "--all", "--force"])
+        }),
+      ),
+    )
+
    const capture = Effect.fn("Git.change.capture")(function* (input: { repository: Repository; path: AbsolutePath }) {
      const scope = path.relative(input.repository.worktree, input.path).replaceAll("\\", "/") || "."
      const tracked = yield* execute(
@@ -887,7 +957,9 @@ const layer = Layer.effect(
        write: writeTree,
        files: treeFiles,
        diff: treeDiff,
+        preview,
        restore,
+        checkout: checkoutTree,
      },
    })
  }),
@@ -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":
@@ -16,7 +16,7 @@ import { Hash } from "@opencode-ai/util/hash"
 export { ID }

 export class Error extends Schema.TaggedErrorClass<Error>()("Snapshot.Error", {
-  operation: Schema.Literals(["capture", "files", "diff", "restore"]),
+  operation: Schema.Literals(["capture", "files", "diff", "preview", "restore"]),
  message: Schema.String,
  cause: Schema.optional(Schema.Defect()),
 }) {}
@@ -36,6 +36,10 @@ export interface RestoreInput {
  readonly files: ReadonlyMap<RelativePath, ID>
 }

+export interface PreviewInput extends RestoreInput {
+  readonly context?: number
+}
+
 export interface Interface {
  /**
   * Capture the current Location-scoped filesystem state as a content-addressed
@@ -56,11 +60,25 @@ export interface Interface {
   */
  readonly diff: (input: DiffInput) => Effect.Effect<readonly File.Diff[], Error>

+  /**
+   * Preview the filesystem result of a selective restore without modifying the
+   * worktree. Each project-relative path maps to the tree it would be restored
+   * from.
+   */
+  readonly preview: (input: PreviewInput) => Effect.Effect<readonly File.Diff[], Error>
+
  /**
   * Restore selected project-relative paths from their associated trees. A path
   * absent from its selected tree is removed; paths outside the map are untouched.
-  */
+   */
  readonly restore: (input: RestoreInput) => Effect.Effect<void, Error>
+
+  /**
+   * Replace the snapshot index with a captured tree and check out all its entries.
+   * Files absent from the tree remain untouched. Prefer selective `restore` when
+   * only known paths should change.
+   */
+  readonly checkout: (snapshot: ID) => Effect.Effect<void, Error>
 }

 export class Service extends Context.Service<Service, Interface>()("@opencode/Snapshot") {}
@@ -158,26 +176,59 @@ const layer = Layer.effect(
        .pipe(Effect.mapError((cause) => failure("diff", cause)))
    })

-    const plan = Effect.fnUntraced(function* (worktree: AbsolutePath, input: RestoreInput) {
+    const plan = Effect.fnUntraced(function* (
+      operation: "preview" | "restore",
+      worktree: AbsolutePath,
+      input: RestoreInput,
+    ) {
      const files = new Map<RelativePath, Git.TreeID>()
      for (const [file, snapshot] of input.files) {
        const absolute = path.resolve(worktree, file)
        if (!FSUtil.contains(worktree, absolute))
-          return yield* new Error({ operation: "restore", message: `Path escapes the project: ${file}` })
+          return yield* new Error({ operation, message: `Path escapes the project: ${file}` })
        files.set(file, Git.TreeID.make(snapshot))
      }
      return files
    })

+    const preview = Effect.fn("Snapshot.preview")(function* (input: PreviewInput) {
+      if (!(yield* enabled())) return yield* new Error({ operation: "preview", message: "Snapshots are disabled" })
+      const repo = yield* repository.pipe(Effect.mapError((cause) => failure("preview", cause)))
+      const files = yield* plan("preview", repo.worktree, input)
+      const current = yield* git.tree
+        .capture({
+          repository: repo.snapshotRepository,
+          scopes: Array.from(files.keys()),
+          ignores: repo.source,
+          maximumUntrackedFileBytes: 2 * 1024 * 1024,
+        })
+        .pipe(Effect.mapError((cause) => failure("preview", cause)))
+      return yield* git.tree
+        .preview({
+          repository: repo.snapshotRepository,
+          current,
+          files,
+          context: input.context,
+        })
+        .pipe(Effect.mapError((cause) => failure("preview", cause)))
+    })
+
    const restore = Effect.fn("Snapshot.restore")(function* (input: RestoreInput) {
      if (!(yield* enabled())) return yield* new Error({ operation: "restore", message: "Snapshots are disabled" })
      const repo = yield* repository.pipe(Effect.mapError((cause) => failure("restore", cause)))
      yield* git.tree
-        .restore({ repository: repo.snapshotRepository, files: yield* plan(repo.worktree, input) })
+        .restore({ repository: repo.snapshotRepository, files: yield* plan("restore", repo.worktree, input) })
        .pipe(Effect.mapError((cause) => failure("restore", cause)))
    })

-    return Service.of({ capture, files, diff, restore })
+    const checkout = Effect.fn("Snapshot.checkout")(function* (snapshot: ID) {
+      const repo = yield* repository.pipe(Effect.mapError((cause) => failure("restore", cause)))
+      yield* git.tree
+        .checkout({ repository: repo.snapshotRepository, tree: Git.TreeID.make(snapshot) })
+        .pipe(Effect.mapError((cause) => failure("restore", cause)))
+    })
+
+    return Service.of({ capture, files, diff, preview, restore, checkout })
  }).pipe(Effect.withSpan("Snapshot.boot")),
 )

@@ -193,7 +244,9 @@ export const noopLayer = Layer.succeed(
    capture: () => Effect.succeed(undefined),
    files: () => Effect.succeed([]),
    diff: () => Effect.succeed([]),
+    preview: () => Effect.succeed([]),
    restore: () => Effect.void,
+    checkout: () => Effect.void,
  }),
 )

@@ -185,6 +185,9 @@ describe("Git trees", () => {
      ])

      const files = new Map([[RelativePath.make("scope/tracked.txt"), before]])
+      const preview = yield* git.tree.preview({ repository, current: after, files, context: 1 })
+      expect(preview).toHaveLength(1)
+      expect(preview[0]?.file).toBe(RelativePath.make("scope/tracked.txt"))
      yield* git.tree.restore({ repository, files })
      expect(yield* read(path.join(root.path, "scope", "tracked.txt"))).toBe("one\n")
      expect(yield* read(path.join(root.path, "scope", "added.txt"))).toBe("added\n")
@@ -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])
+  })
 })
@@ -117,6 +117,9 @@ describe("Snapshot", () => {
              RelativePath.make("scope/tracked.txt"),
            ])
            const plan = new Map([[RelativePath.make("scope/tracked.txt"), before]])
+            const preview = yield* snapshot.preview({ files: plan, context: 1 })
+            expect(preview).toHaveLength(1)
+            expect(preview[0]?.file).toBe(RelativePath.make("scope/tracked.txt"))
            yield* snapshot.restore({ files: plan })
            expect(yield* read(path.join(location, "tracked.txt"))).toBe("one\n")
            expect(yield* read(path.join(location, "added.txt"))).toBe("added\n")
@@ -182,6 +185,36 @@ describe("Snapshot", () => {
      (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
    ),
  )
+
+  testEffect(Layer.empty).live("checks out a legacy revert snapshot without removing unrelated files", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => tmpdir()),
+      (tmp) =>
+        Effect.gen(function* () {
+          const project = path.join(tmp.path, "project")
+          yield* Effect.promise(async () => {
+            await fs.mkdir(project)
+            await fs.writeFile(path.join(project, "tracked.txt"), "one\n")
+            await initGit(project)
+          })
+
+          yield* Effect.gen(function* () {
+            const snapshot = yield* Snapshot.Service
+            const before = yield* snapshot.capture()
+            expect(before).toBeDefined()
+            if (!before) return
+            yield* Effect.promise(async () => {
+              await fs.writeFile(path.join(project, "tracked.txt"), "two\n")
+              await fs.writeFile(path.join(project, "unrelated.txt"), "keep\n")
+            })
+            yield* snapshot.checkout(before)
+            expect(yield* read(path.join(project, "tracked.txt"))).toBe("one\n")
+            expect(yield* read(path.join(project, "unrelated.txt"))).toBe("keep\n")
+          }).pipe(Effect.provide(snapshotLayer(tmp.path, project)))
+        }),
+      (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+    ),
+  )
 })

 function snapshotLayer(data: string, directory: string) {
Author	SHA1	Message	Date
Shoubhit Dash	0ea8f6d2fe	refactor(core): adapt ai sdk transport execution	2026-08-07 17:18:48 +05:30
Shoubhit Dash	e027ce316b	refactor(ai): add sequential websocket channel seam	2026-08-07 17:17:25 +05:30
Shoubhit Dash	3e7efffcb6	fix(ai): harden websocket error contracts	2026-08-05 23:04:53 +05:30