refactor(core): expose session transport eligibility

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. `OpenResponsesChannel.transport(...)` owns the provider-neutral Responses WebSocket concept: it prepares one final request, executes HTTP by default, strips WebSocket-disallowed fields, and passes a generic channel exchange to a per-call `WebSocketChannelExecutor` when supplied. Provider-specific Responses routes opt in with handshake and connection-age policy. `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,14 +153,16 @@ 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)
     open-responses.ts         provider-neutral Responses protocol baseline
-    openai-responses.ts       OpenAI tools/events/transports composed over OpenResponses
+    open-responses-channel.ts provider-neutral Responses WebSocket transport factory
+    openai-responses.ts       OpenAI tools/events and channel policy composed over OpenResponses
     anthropic-messages.ts
     gemini.ts
     bedrock-converse.ts

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. The provider-neutral Open Responses implementation owns the reusable WebSocket request and event contract, while each provider opts in with its own handshake and connection policy. 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
 
@@ -78,24 +77,24 @@ Other `aisdk:` packages, including Google Vertex, Azure, and Bedrock, currently
 
 These are implementation/API slices, not separate npm packages.
 
-| API slice                     | Package-like entrypoint                                     | Purpose                                                                      |
+| API slice                     | Package-like entrypoint                                     | Purpose                                                                                    |
-| ----------------------------- | ----------------------------------------------------------- | ---------------------------------------------------------------------------- |
+| ----------------------------- | ----------------------------------------------------------- | ------------------------------------------------------------------------------------------ |
-| OpenAI Chat                   | `@opencode-ai/ai/providers/openai/chat`                     | OpenAI `/chat/completions` semantics.                                        |
+| OpenAI Chat                   | `@opencode-ai/ai/providers/openai/chat`                     | OpenAI `/chat/completions` semantics.                                                      |
-| OpenAI Responses              | `@opencode-ai/ai/providers/openai/responses`                | OpenAI `/responses` semantics with HTTP/WebSocket selected through settings. |
+| OpenAI Responses              | `@opencode-ai/ai/providers/openai/responses`                | OpenAI `/responses` semantics with HTTP default and optional per-call WebSocket execution. |
-| OpenAI-compatible Chat        | `@opencode-ai/ai/providers/openai-compatible`               | Generic OpenAI-compatible `/chat/completions`.                               |
+| OpenAI-compatible Chat        | `@opencode-ai/ai/providers/openai-compatible`               | Generic OpenAI-compatible `/chat/completions`.                                             |
-| Open Responses-compatible     | `@opencode-ai/ai/providers/openai-compatible/responses`     | Generic provider-neutral `/responses`.                                       |
+| Open Responses-compatible     | `@opencode-ai/ai/providers/openai-compatible/responses`     | Generic provider-neutral `/responses`.                                                     |
-| Anthropic-compatible Messages | `@opencode-ai/ai/providers/anthropic-compatible`            | Generic Anthropic-compatible `/messages`.                                    |
+| Anthropic-compatible Messages | `@opencode-ai/ai/providers/anthropic-compatible`            | Generic Anthropic-compatible `/messages`.                                                  |
-| Anthropic Messages            | `@opencode-ai/ai/providers/anthropic`                       | Anthropic Messages API.                                                      |
+| Anthropic Messages            | `@opencode-ai/ai/providers/anthropic`                       | Anthropic Messages API.                                                                    |
-| Gemini Developer API          | `@opencode-ai/ai/providers/google`                          | Google AI Studio Gemini API.                                                 |
+| Gemini Developer API          | `@opencode-ai/ai/providers/google`                          | Google AI Studio Gemini API.                                                               |
-| Vertex Gemini                 | `@opencode-ai/ai/providers/google-vertex/gemini`            | Vertex Gemini API; `providers/google-vertex` is the default alias.           |
+| Vertex Gemini                 | `@opencode-ai/ai/providers/google-vertex/gemini`            | Vertex Gemini API; `providers/google-vertex` is the default alias.                         |
-| Vertex Chat                   | `@opencode-ai/ai/providers/google-vertex/chat`              | Vertex OpenAI-compatible Chat Completions for MaaS models.                   |
+| Vertex Chat                   | `@opencode-ai/ai/providers/google-vertex/chat`              | Vertex OpenAI-compatible Chat Completions for MaaS models.                                 |
-| Vertex Responses              | `@opencode-ai/ai/providers/google-vertex/responses`         | Vertex Open Responses for Grok models.                                       |
+| Vertex Responses              | `@opencode-ai/ai/providers/google-vertex/responses`         | Vertex Open Responses for Grok models.                                                     |
-| Vertex Messages               | `@opencode-ai/ai/providers/google-vertex/messages`          | Vertex-hosted Anthropic Messages API.                                        |
+| Vertex Messages               | `@opencode-ai/ai/providers/google-vertex/messages`          | Vertex-hosted Anthropic Messages API.                                                      |
-| Bedrock Converse              | `@opencode-ai/ai/providers/amazon-bedrock`                  | AWS Bedrock Converse API.                                                    |
+| Bedrock Converse              | `@opencode-ai/ai/providers/amazon-bedrock`                  | AWS Bedrock Converse API.                                                                  |
-| Bedrock Mantle Chat           | `@opencode-ai/ai/providers/amazon-bedrock/mantle/chat`      | AWS Bedrock Mantle OpenAI-compatible Chat API.                               |
+| Bedrock Mantle Chat           | `@opencode-ai/ai/providers/amazon-bedrock/mantle/chat`      | AWS Bedrock Mantle OpenAI-compatible Chat API.                                             |
-| Bedrock Mantle Responses      | `@opencode-ai/ai/providers/amazon-bedrock/mantle/responses` | AWS Bedrock Mantle OpenAI-compatible Responses API.                          |
+| Bedrock Mantle Responses      | `@opencode-ai/ai/providers/amazon-bedrock/mantle/responses` | AWS Bedrock Mantle OpenAI-compatible Responses API.                                        |
-| Azure OpenAI Chat             | `@opencode-ai/ai/providers/azure/chat`                      | Azure specialization of OpenAI Chat.                                         |
+| Azure OpenAI Chat             | `@opencode-ai/ai/providers/azure/chat`                      | Azure specialization of OpenAI Chat.                                                       |
-| Azure OpenAI Responses        | `@opencode-ai/ai/providers/azure/responses`                 | Azure specialization of OpenAI Responses.                                    |
+| Azure OpenAI Responses        | `@opencode-ai/ai/providers/azure/responses`                 | Azure specialization of OpenAI Responses.                                                  |
 
 ## Suggested Next Work Slices
 

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,
@@ -387,11 +376,9 @@ import { model } from "@opencode-ai/ai/providers/google-vertex/messages"
 model("claude-sonnet-4-6", { project, location: "global" })
 ```
 
-The client should not require a different public layer just because a selected
+The client does not require a different public layer for WebSocket execution.
-route uses WebSocket. Use one `LLMClient.layer` with HTTP and WebSocket runtime
+Responses routes use HTTP by default, and callers may pass a channel executor per
-capabilities available; routes that do not need WebSocket simply never touch it.
+call. Routes without channel support simply ignore that execution capability.
-If a WebSocket route is selected in an environment without WebSocket support,
-fail with a typed transport configuration error.
 
 Azure is a route specialization with auth/path/default changes plus input
 mapping. The public API configures the Azure resource once, then selects
@@ -499,16 +486,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 +528,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 +563,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
@@ -599,10 +580,8 @@ App boundary = explicit durable-config -> typed-provider call
 - [ ] Decide whether a tiny `Provider.define(...)` helper is warranted after two
       or three provider conversions; start with plain objects if duplication is not
       yet painful.
-- [x] Update `packages/opencode/src/session/llm/native-request.ts` to construct
+- [x] Keep executable model construction transport-neutral at the Session boundary;
-      executable models at the session boundary with explicit provider facade
+      Session-scoped execution policy supplies channel capability separately.
-      calls, mapping catalog metadata such as `endpoint.websocket` to the correct
-      named route selector.
 - [ ] Update tests so direct route/provider tests assert route values are carried
       by executable models, and opencode/native tests assert boundary-based route
       selection.

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/index.ts

@@ -7,3 +7,4 @@ export * as OpenAICompatibleChat from "./openai-compatible-chat"
 export * as OpenAICompatibleResponses from "./openai-compatible-responses"
 export * as OpenAIResponses from "./openai-responses"
 export * as OpenResponses from "./open-responses"
+export * as OpenResponsesChannel from "./open-responses-channel"

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

@@ -0,0 +1,119 @@
+import { Effect, Schema, Stream } from "effect"
+import { Headers } from "effect/unstable/http"
+import { Framing } from "../route/framing"
+import {
+  HttpTransport,
+  WebSocketTransport,
+  type Transport,
+  type WebSocketChannelDriver,
+  type WebSocketChannelExchange,
+} from "../route/transport"
+import * as ProviderShared from "./shared"
+import { OpenResponses } from "./open-responses"
+
+const WebSocketResponseCreate = Schema.StructWithRest(Schema.Struct({ type: Schema.tag("response.create") }), [
+  Schema.Record(Schema.String, Schema.Unknown),
+])
+const decodeMessage = ProviderShared.validateWith(Schema.decodeUnknownEffect(WebSocketResponseCreate))
+const encodeMessage = Schema.encodeSync(Schema.fromJsonString(WebSocketResponseCreate))
+const decodeEvent = Schema.decodeUnknownEffect(OpenResponses.protocol.stream.event)
+
+export interface Options {
+  readonly id: string
+  readonly name: string
+  readonly rotateAfterMs?: number
+  readonly headers?: (headers: Headers.Headers) => Headers.Headers
+}
+
+export interface Prepared {
+  readonly http: HttpTransport.HttpPrepared<string>
+  readonly channel?: {
+    readonly url: string
+    readonly headers: Headers.Headers
+    readonly rotateAfterMs?: number
+    readonly driver: WebSocketChannelDriver
+  }
+}
+
+const message = (body: unknown) =>
+  Effect.gen(function* () {
+    if (!ProviderShared.isRecord(body))
+      return yield* ProviderShared.invalidRequest("Open Responses WebSocket body must be a JSON object")
+    const { stream: _stream, stream_options: _streamOptions, background: _background, ...request } = body
+    return encodeMessage(yield* decodeMessage({ ...request, type: "response.create" }))
+  })
+
+const driver = (options: Options, body: string): WebSocketChannelDriver => ({
+  create: () => Effect.succeed({ message: body, mode: "full" }),
+  observe: (_create, frame) =>
+    Effect.gen(function* () {
+      const event = yield* decodeEvent(frame).pipe(
+        Effect.mapError(() => ProviderShared.eventError(options.id, `Invalid ${options.name} 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(options.id, event, `${options.name} response failed`),
+        }
+      if (event.type === "error") {
+        yield* OpenResponses.decodeKnownErrorEvent(event).pipe(
+          Effect.mapError(() =>
+            ProviderShared.eventError(options.id, `${options.name} returned a malformed error event`, frame),
+          ),
+        )
+        return {
+          type: "provider-failure",
+          error: OpenResponses.providerFailure(options.id, event, `${options.name} stream error`),
+        }
+      }
+      return { type: "frame", frame }
+    }),
+})
+
+export const transport = <Body>(options: Options): Transport<Body, Prepared, string> => {
+  const http = HttpTransport.sseJson.with<Body>()
+  return {
+    id: http.id,
+    prepare: (input) =>
+      Effect.gen(function* () {
+        const parts = yield* HttpTransport.jsonRequestParts(input)
+        const headers = Headers.remove(options.headers?.(parts.headers) ?? parts.headers, "content-length")
+        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,
+                rotateAfterMs: options.rotateAfterMs,
+                driver: driver(options, yield* message(parts.jsonBody)),
+              }
+            : undefined,
+        }
+      }),
+    execute: (prepared, request, runtime, executeOptions) => {
+      if (!executeOptions?.webSocket || !prepared.channel) return http.execute(prepared.http, request, runtime)
+      const exchange: WebSocketChannelExchange = {
+        id: request.id ?? "request",
+        connect: {
+          url: prepared.channel.url,
+          headers: prepared.channel.headers,
+          rotateAfterMs: prepared.channel.rotateAfterMs,
+        },
+        fallback: () =>
+          Stream.unwrap(
+            http.execute(prepared.http, request, runtime).pipe(Effect.map((execution) => execution.frames)),
+          ),
+        driver: prepared.channel.driver,
+      }
+      return executeOptions.webSocket.execute(exchange)
+    },
+  }
+}
+
+export const OpenResponsesChannel = { transport } as const

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.ts

@@ -1,18 +1,22 @@
 import { Effect, Encoding, Schema } from "effect"
+import { Headers } from "effect/unstable/http"
 import { Route } from "../route/client"
 import { Auth } from "../route/auth"
 import { Endpoint } from "../route/endpoint"
 import { Protocol } from "../route/protocol"
-import { HttpTransport, WebSocketTransport } from "../route/transport"
+import { HttpTransport } 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 { OpenResponsesChannel } from "./open-responses-channel"
 
 const ADAPTER = "openai-responses"
 const NAME = "OpenAI Responses"
+const WEBSOCKET_PROTOCOL_HEADER = "responses_websockets=2026-02-06"
+const WEBSOCKET_ROTATE_AFTER_MS = 55 * 60 * 1000
 export const DEFAULT_BASE_URL = "https://api.openai.com/v1"
 export const PATH = OpenResponses.PATH
 
@@ -57,16 +61,6 @@ const OpenAIResponsesBody = Schema.Struct({
 })
 export type OpenAIResponsesBody = Schema.Schema.Type<typeof OpenAIResponsesBody>
 
-const OpenAIResponsesWebSocketMessage = Schema.StructWithRest(
-  Schema.Struct({
-    type: Schema.tag("response.create"),
-    ...OpenAIResponsesCoreFields,
-  }),
-  [Schema.Record(Schema.String, Schema.Unknown)],
-)
-type OpenAIResponsesWebSocketMessage = Schema.Schema.Type<typeof OpenAIResponsesWebSocketMessage>
-const encodeWebSocketMessage = Schema.encodeSync(Schema.fromJsonString(OpenAIResponsesWebSocketMessage))
-
 const extension = {
   id: ADAPTER,
   name: NAME,
@@ -249,6 +243,12 @@ const endpoint = Endpoint.path<OpenAIResponsesBody>(PATH, { baseURL: DEFAULT_BAS
 const auth = Auth.none
 
 export const httpTransport = HttpTransport.sseJson.with<OpenAIResponsesBody>()
+export const transport = OpenResponsesChannel.transport<OpenAIResponsesBody>({
+  id: ADAPTER,
+  name: NAME,
+  rotateAfterMs: WEBSOCKET_ROTATE_AFTER_MS,
+  headers: (headers) => Headers.set(headers, "openai-beta", headers["openai-beta"] ?? WEBSOCKET_PROTOCOL_HEADER),
+})
 
 export const route = Route.make({
   id: ADAPTER,
@@ -257,36 +257,7 @@ export const route = Route.make({
   protocol,
   endpoint,
   auth,
-  transport: httpTransport,
+  transport,
-  defaults: { providerOptions: { openai: { store: false } } },
-})
-
-const decodeWebSocketMessage = ProviderShared.validateWith(Schema.decodeUnknownEffect(OpenAIResponsesWebSocketMessage))
-
-const webSocketMessage = (body: OpenAIResponsesBody | Record<string, unknown>) =>
-  Effect.gen(function* () {
-    if (!ProviderShared.isRecord(body))
-      return yield* ProviderShared.invalidRequest("OpenAI Responses WebSocket body must be a JSON object")
-    const { stream: _stream, ...message } = body
-    return yield* decodeWebSocketMessage({ ...message, type: "response.create" })
-  })
-
-export const webSocketTransport = WebSocketTransport.jsonTransport.with<
-  OpenAIResponsesBody,
-  OpenAIResponsesWebSocketMessage
->({
-  toMessage: webSocketMessage,
-  encodeMessage: encodeWebSocketMessage,
-})
-
-export const webSocketRoute = Route.make({
-  id: `${ADAPTER}-websocket`,
-  provider: "openai",
-  providerMetadataKey: "openai",
-  protocol,
-  endpoint,
-  auth,
-  transport: webSocketTransport,
   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,50 @@
+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
+    /** Provider-safe connection age after which Core should rotate before sending. */
+    readonly rotateAfterMs?: number
+  }
+  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,
@@ -17,6 +17,7 @@ import {
   OpenAICompatibleResponses,
   OpenAIResponses,
   OpenResponses,
+  OpenResponsesChannel,
 } from "@opencode-ai/ai/protocols"
 import * as AnthropicMessages from "@opencode-ai/ai/protocols/anthropic-messages"
 import { TestLLM } from "@opencode-ai/ai/testing"
@@ -37,6 +38,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 +46,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(
@@ -83,10 +84,10 @@ describe("public exports", () => {
     expect(OpenAIChat.route.id).toBe("openai-chat")
     expect(OpenAICompatibleChat.route.id).toBe("openai-compatible-chat")
     expect(OpenResponses.protocol.id).toBe("open-responses")
+    expect(OpenResponsesChannel.transport).toBeFunction()
     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 })
     }),
   )
@@ -236,47 +238,59 @@ describe("OpenAI Responses route", () => {
   it.effect("streams OpenAI Responses over WebSocket", () =>
     Effect.gen(function* () {
       const sent: string[] = []
-      const opened: Array<{ readonly url: string; readonly authorization: string | undefined }> = []
+      const opened: Array<{
+        readonly url: string
+        readonly authorization: string | undefined
+        readonly protocol: 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,
+                  protocol: input.headers["openai-beta"],
+                })
+                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({
-            "gpt-4.1-mini",
+            baseURL: "https://api.openai.test/v1/",
-          ),
+            apiKey: "test",
+            headers: { "openai-beta": "custom-protocol" },
+          }).responses("gpt-4.1-mini"),
           prompt: "Say hello.",
         }),
+        { webSocket },
       ).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))))
 
       expect(response.text).toBe("Hi")
-      expect(opened).toEqual([{ url: "wss://api.openai.test/v1/responses", authorization: "Bearer test" }])
+      expect(opened).toEqual([
+        {
+          url: "wss://api.openai.test/v1/responses",
+          authorization: "Bearer test",
+          protocol: "custom-protocol",
+        },
+      ])
       expect(closed).toBe(true)
       expect(sent).toHaveLength(1)
       expect(JSON.parse(sent[0])).toEqual({
@@ -288,15 +302,245 @@ 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" },
+              stream_options: { include_usage: true },
+              background: true,
+            },
+            headers: { "x-request": "request" },
+            query: { mode: "test" },
+          },
+        }),
+        {
+          webSocket: {
+            execute: (exchange) =>
+              Effect.gen(function* () {
+                expect(exchange.connect.rotateAfterMs).toBe(55 * 60 * 1000)
+                expect(exchange.connect.headers["openai-beta"]).toBe("responses_websockets=2026-02-06")
+                expect(exchange.connect.headers["content-length"]).toBeUndefined()
+                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, stream_options: _streamOptions, background: _background, ...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,
+        stream_options: { include_usage: true },
+        background: 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/bus.ts

@@ -3,7 +3,7 @@ export * as Bus from "./bus"
 import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { EventLog } from "@opencode-ai/schema/event-log"
-import { and, asc, eq, gt, lte, sql } from "drizzle-orm"
+import { and, asc, eq, gt, inArray, lte, sql } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
@@ -134,6 +134,8 @@ export interface Interface {
     readonly after?: number
     readonly follow?: boolean
   }) => Stream.Stream<LogItem>
+  /** Latest committed seq per aggregate. Aggregates without events are absent. */
+  readonly sequences: (aggregateIDs: ReadonlyArray<string>) => Effect.Effect<ReadonlyMap<string, Event.Seq>>
   /** @deprecated Use `subscribe()` and consume the returned stream. */
   readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
   readonly project: <D extends Event.Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
@@ -655,6 +657,19 @@ export const layerWith = (options?: LayerOptions) =>
           }),
         )
 
+      const sequences = (aggregateIDs: ReadonlyArray<string>): Effect.Effect<ReadonlyMap<string, Event.Seq>> => {
+        if (aggregateIDs.length === 0) return Effect.succeed(new Map())
+        return db
+          .select({ aggregateID: EventSequenceTable.aggregate_id, seq: EventSequenceTable.seq })
+          .from(EventSequenceTable)
+          .where(inArray(EventSequenceTable.aggregate_id, Array.from(aggregateIDs)))
+          .all()
+          .pipe(
+            Effect.orDie,
+            Effect.map((rows) => new Map(rows.map((row) => [row.aggregateID, Event.Seq.make(row.seq)]))),
+          )
+      }
+
       const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
         Effect.sync(() => {
           listeners.push(listener)
@@ -676,6 +691,7 @@ export const layerWith = (options?: LayerOptions) =>
         publish,
         subscribe,
         log,
+        sequences,
         listen,
         project,
         replay,

packages/core/src/formatter.ts

@@ -1,6 +1,6 @@
 export * as Formatter from "./formatter"
 
-import { Context, Effect, Layer } from "effect"
+import { Context, Effect, Layer, Schema } from "effect"
 import { ChildProcess } from "effect/unstable/process"
 import path from "path"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
@@ -11,7 +11,16 @@ import { Config } from "./config"
 import { Location } from "./location"
 import { make, type Info } from "./formatter/builtins"
 
+export const Status = Schema.Struct({
+  name: Schema.String,
+  extensions: Schema.Array(Schema.String),
+  enabled: Schema.Boolean,
+}).annotate({ identifier: "FormatterStatus" })
+export type Status = typeof Status.Type
+
 export interface Interface {
+  readonly init: () => Effect.Effect<void>
+  readonly status: () => Effect.Effect<Status[]>
   readonly file: (filepath: string) => Effect.Effect<boolean>
 }
 
@@ -75,6 +84,23 @@ const layer = Layer.effect(
       return result
     })
 
+    const init = Effect.fn("Formatter.init")(function* () {
+      yield* load
+    })
+
+    const status = Effect.fn("Formatter.status")(function* () {
+      yield* load
+      return yield* Effect.forEach(formatters, (formatter) =>
+        command(formatter).pipe(
+          Effect.map((enabled) => ({
+            name: formatter.name,
+            extensions: [...formatter.extensions],
+            enabled: enabled !== false,
+          })),
+        ),
+      )
+    })
+
     const file = Effect.fn("Formatter.file")(function* (filepath: string) {
       yield* load
       const matching = formatters.filter((formatter) =>
@@ -117,7 +143,7 @@ const layer = Layer.effect(
       return false
     })
 
-    return Service.of({ file })
+    return Service.of({ init, status, file })
   }),
 )
 

packages/core/src/git.ts

@@ -1,7 +1,8 @@
 export * as Git from "./git"
 
 import path from "path"
-import { Context, Effect, Layer, Schema } from "effect"
+import { randomUUID } from "crypto"
+import { Context, Effect, Layer, Schema, Stream } from "effect"
 import { ChildProcess } from "effect/unstable/process"
 import { AbsolutePath, RelativePath } from "./schema"
 import { FSUtil } from "@opencode-ai/util/fs-util"
@@ -35,6 +36,9 @@ const snapshotConfig = `[core]
 	threads = true
 `
 
+export const ChangeSet = Schema.String.pipe(Schema.brand("Git.ChangeSet"))
+export type ChangeSet = typeof ChangeSet.Type
+
 export const TreeID = Schema.String.pipe(Schema.brand("Git.TreeID"))
 export type TreeID = typeof TreeID.Type
 
@@ -69,6 +73,13 @@ export class WorktreeError extends Schema.TaggedErrorClass<WorktreeError>()("Git
   cause: Schema.optional(Schema.Defect()),
 }) {}
 
+export class PatchError extends Schema.TaggedErrorClass<PatchError>()("Git.PatchError", {
+  operation: Schema.Literals(["capture", "apply", "reset"]),
+  directory: AbsolutePath,
+  message: Schema.String,
+  cause: Schema.optional(Schema.Defect()),
+}) {}
+
 export interface Interface {
   readonly repo: {
     readonly discover: (input: AbsolutePath) => Effect.Effect<Repository | undefined>
@@ -105,6 +116,20 @@ export interface Interface {
     ) => Effect.Effect<void, OperationError>
     readonly resetHard: (repository: Repository, revision: string) => Effect.Effect<void, OperationError>
   }
+  readonly change: {
+    readonly capture: (input: { repository: Repository; path: AbsolutePath }) => Effect.Effect<ChangeSet, PatchError>
+    readonly apply: (input: {
+      repository: Repository
+      path: AbsolutePath
+      changes: ChangeSet
+    }) => Effect.Effect<void, PatchError>
+    readonly discard: (input: {
+      repository: Repository
+      path: AbsolutePath
+      index: "preserve" | "reset"
+      untracked: "preserve" | "remove"
+    }) => Effect.Effect<void, PatchError>
+  }
   readonly worktree: {
     readonly create: (input: {
       repository: Repository
@@ -150,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>
   }
 }
 
@@ -625,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(
@@ -654,6 +738,142 @@ 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(
+        input.repository.worktree,
+        proc,
+      )(["diff", "--binary", "HEAD", "--", scope]).pipe(
+        Effect.mapError(
+          (cause) => new PatchError({ operation: "capture", directory: input.path, message: cause.message, cause }),
+        ),
+      )
+      if (tracked.exitCode !== 0) {
+        return yield* new PatchError({
+          operation: "capture",
+          directory: input.path,
+          message: tracked.stderr.trim() || tracked.text.trim() || "Failed to capture tracked changes",
+        })
+      }
+
+      const untracked = yield* execute(
+        input.repository.worktree,
+        proc,
+      )(["ls-files", "--others", "--exclude-standard", "-z", "--", scope]).pipe(
+        Effect.mapError(
+          (cause) => new PatchError({ operation: "capture", directory: input.path, message: cause.message, cause }),
+        ),
+      )
+      if (untracked.exitCode !== 0) {
+        return yield* new PatchError({
+          operation: "capture",
+          directory: input.path,
+          message: untracked.stderr.trim() || untracked.text.trim() || "Failed to list untracked changes",
+        })
+      }
+
+      const created = yield* Effect.forEach(untracked.text.split("\0").filter(Boolean), (file) =>
+        execute(
+          input.repository.worktree,
+          proc,
+        )(["diff", "--binary", "--no-index", "--", "/dev/null", file]).pipe(
+          Effect.mapError(
+            (cause) => new PatchError({ operation: "capture", directory: input.path, message: cause.message, cause }),
+          ),
+          Effect.flatMap((result) =>
+            // git diff --no-index returns 1 when differences were found.
+            result.exitCode === 0 || result.exitCode === 1
+              ? Effect.succeed(result.text)
+              : Effect.fail(
+                  new PatchError({
+                    operation: "capture",
+                    directory: input.path,
+                    message:
+                      result.stderr.trim() || result.text.trim() || `Failed to capture untracked change: ${file}`,
+                  }),
+                ),
+          ),
+        ),
+      )
+      return ChangeSet.make([tracked.text, ...created].filter(Boolean).join("\n"))
+    })
+
+    const apply = Effect.fn("Git.change.apply")(function* (input: {
+      repository: Repository
+      path: AbsolutePath
+      changes: ChangeSet
+    }) {
+      const result = yield* proc
+        .run(
+          ChildProcess.make("git", ["apply", "-"], {
+            cwd: input.path,
+            extendEnv: true,
+            stdin: Stream.make(new TextEncoder().encode(input.changes)),
+          }),
+        )
+        .pipe(
+          Effect.mapError(
+            (cause) => new PatchError({ operation: "apply", directory: input.path, message: cause.message, cause }),
+          ),
+        )
+      if (result.exitCode === 0) return
+      return yield* new PatchError({
+        operation: "apply",
+        directory: input.path,
+        message:
+          result.stderr.toString("utf8").trim() || result.stdout.toString("utf8").trim() || "Failed to apply changes",
+      })
+    })
+
+    const discard = Effect.fn("Git.change.discard")(function* (input: {
+      repository: Repository
+      path: AbsolutePath
+      index: "preserve" | "reset"
+      untracked: "preserve" | "remove"
+    }) {
+      const scope = path.relative(input.repository.worktree, input.path).replaceAll("\\", "/") || "."
+      const restore = yield* execute(
+        input.repository.worktree,
+        proc,
+      )(input.index === "reset" ? ["checkout", "HEAD", "--", scope] : ["checkout", "--", scope]).pipe(
+        Effect.mapError(
+          (cause) => new PatchError({ operation: "reset", directory: input.path, message: cause.message, cause }),
+        ),
+      )
+      if (restore.exitCode !== 0) {
+        return yield* new PatchError({
+          operation: "reset",
+          directory: input.path,
+          message: restore.stderr.trim() || restore.text.trim() || "Failed to restore tracked changes",
+        })
+      }
+      if (input.untracked === "preserve") return
+      const clean = yield* execute(
+        input.repository.worktree,
+        proc,
+      )(["clean", "-fd", "--", scope]).pipe(
+        Effect.mapError(
+          (cause) => new PatchError({ operation: "reset", directory: input.path, message: cause.message, cause }),
+        ),
+      )
+      if (clean.exitCode === 0) return
+      return yield* new PatchError({
+        operation: "reset",
+        directory: input.path,
+        message: clean.stderr.trim() || clean.text.trim() || "Failed to clean untracked changes",
+      })
+    })
+
     const worktreeRun = Effect.fnUntraced(function* (
       operation: "create" | "remove" | "list",
       repository: Repository,
@@ -729,6 +949,7 @@ const layer = Layer.effect(
       remote: { get: remote },
       history: { head, branch, defaultRemoteBranch: remoteHead, rootCommits: roots },
       sync: { fetchRemotes: fetch, fetchBranch, checkoutRemoteBranch: checkout, resetHard: reset },
+      change: { capture, apply, discard },
       worktree: { create: worktreeCreate, remove: worktreeRemove, list: worktreeList },
       index: { refresh, ignored },
       tree: {
@@ -736,7 +957,9 @@ const layer = Layer.effect(
         write: writeTree,
         files: treeFiles,
         diff: treeDiff,
+        preview,
         restore,
+        checkout: checkoutTree,
       },
     })
   }),

packages/core/src/plugin/provider/openai.ts

@@ -6,7 +6,6 @@ import { App } from "../../app"
 import { Credential } from "../../credential"
 import { Bus } from "../../bus"
 import { Integration } from "../../integration"
-import { Model } from "../../model"
 import { OauthCallbackPage } from "../../oauth/page"
 import { Provider } from "../../provider"
 import type { PluginInternal } from "../internal"
@@ -198,10 +197,10 @@ export const OpenAIPlugin = define({
       if (!item) return
       item.provider.settings = Provider.mergeOverlay(item.provider.settings, { baseURL: codexBaseURL })
       const account = chatgpt.metadata?.accountID
-      item.provider.headers = Provider.mergeHeaders(
+      item.provider.headers = Provider.mergeHeaders(item.provider.headers, {
-        item.provider.headers,
+        originator: "opencode",
-        typeof account === "string" ? { "chatgpt-account-id": account } : undefined,
+        ...(typeof account === "string" ? { "chatgpt-account-id": account } : {}),
-      )
+      })
       for (const model of item.models.values()) {
         // ChatGPT-plan tokens only authorize codex-eligible models, and the
         // subscription covers usage, so hide the rest and zero the cost.
@@ -225,17 +224,6 @@ export const OpenAIPlugin = define({
         })
       }
     })
-    yield* ctx.session.hook("http", (evt) =>
-      evt.use((request, next) => {
-        if (!chatgpt || evt.model.providerID !== Provider.ID.openai) return next(request)
-        const url = new URL(request.url)
-        request.headers.set("originator", "opencode")
-        request.headers.set("session-id", evt.sessionID)
-        if (url.origin !== "https://api.openai.com") return next(request)
-        return next(new Request(`${codexBaseURL}${url.pathname.replace(/^\/v1/, "")}${url.search}`, request))
-      }),
-    )
-
     const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.catalog.reload())))
     yield* bus.subscribe(Integration.Event.ConnectionUpdated).pipe(
       Stream.filter((event) => event.data.integrationID === Integration.ID.make("openai")),

packages/core/src/session/model-request.ts

@@ -46,6 +46,8 @@ const declineDefect = (cause: Cause.Cause<Tool.Error>) => {
 interface Prepared {
   readonly request: LLMRequest
   readonly options: StreamOptions
+  /** False when Session HTTP middleware requires the request to remain on HTTP. */
+  readonly webSocketEligible: boolean
   /**
    * One request-scoped execution operation. Unknown, hook-removed, and
    * step-limit-violating calls fail individually through the same seam.
@@ -76,6 +78,40 @@ const unsupportedMedia = (mime: string, name: string | undefined, capabilities:
   }
 }
 
+export const composeHttpMiddleware = (middlewares: ReadonlyArray<SessionHttpMiddleware>): StreamOptions["http"] => {
+  if (middlewares.length === 0) return undefined
+  return (request, handler) =>
+    Effect.gen(function* () {
+      let latest = request
+      const origins = new WeakMap<Response, HttpClientRequest.HttpClientRequest>()
+      const web = yield* HttpClientRequest.toWeb(request)
+      const send = (input: Request) =>
+        Effect.gen(function* () {
+          let sent = HttpClientRequest.fromWeb(input)
+          if (input.body)
+            sent = HttpClientRequest.bodyUint8Array(
+              sent,
+              new Uint8Array(yield* Effect.promise(() => input.clone().arrayBuffer())),
+              input.headers.get("content-type") ?? undefined,
+            )
+          latest = sent
+          const response = yield* handler(sent)
+          const body = [204, 205, 304].includes(response.status)
+            ? null
+            : yield* Stream.toReadableStreamEffect(response.stream)
+          const output = new Response(body, { status: response.status, headers: response.headers })
+          origins.set(output, sent)
+          return output
+        })
+      const dispatch = middlewares.reduce<SessionHttpHandler>(
+        (next, item) => (input: Request) => item(input, next),
+        send,
+      )
+      const response = yield* dispatch(web)
+      return HttpClientResponse.fromWeb(origins.get(response) ?? latest, response)
+    }).pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause)))))
+}
+
 export const unsupportedParts = (messages: LLMRequest["messages"], capabilities: Model.Capabilities) =>
   messages.map((message) =>
     Message.make({
@@ -227,49 +263,18 @@ export const layer = Layer.effect(
         tools: Array.from(hooked, ([name, tool]) => ({ ...tool, name })),
         toolChoice: stepLimitReached ? "none" : undefined,
       })
-      const options: StreamOptions = {
+      const middlewares: SessionHttpMiddleware[] = []
-        http: (request, handler) =>
+      yield* hooks.trigger("session", "http", {
-          Effect.gen(function* () {
+        sessionID: session.id,
-            let latest = request
+        agent: agent.id,
-            const origins = new WeakMap<Response, HttpClientRequest.HttpClientRequest>()
+        model: resolved.ref,
-            const middlewares: SessionHttpMiddleware[] = []
+        use: (item) =>
-            const web = yield* HttpClientRequest.toWeb(request)
+          Effect.sync(() => {
-            yield* hooks.trigger("session", "http", {
+            middlewares.push(item)
-              sessionID: session.id,
+          }),
-              agent: agent.id,
+      })
-              model: resolved.ref,
+      const http = composeHttpMiddleware(middlewares)
-              use: (item) =>
+      const options: StreamOptions = http ? { http } : {}
-                Effect.sync(() => {
-                  middlewares.push(item)
-                }),
-            })
-            const send = (input: Request) =>
-              Effect.gen(function* () {
-                let sent = HttpClientRequest.fromWeb(input)
-                if (input.body)
-                  sent = HttpClientRequest.bodyUint8Array(
-                    sent,
-                    new Uint8Array(yield* Effect.promise(() => input.clone().arrayBuffer())),
-                    input.headers.get("content-type") ?? undefined,
-                  )
-                latest = sent
-                const response = yield* handler(sent)
-                const body = [204, 205, 304].includes(response.status)
-                  ? null
-                  : yield* Stream.toReadableStreamEffect(response.stream)
-                const output = new Response(body, { status: response.status, headers: response.headers })
-                origins.set(output, sent)
-                return output
-              })
-            const dispatch = middlewares.reduce<SessionHttpHandler>(
-              (next, item) => (input: Request) => item(input, next),
-              send,
-            )
-            const response = yield* dispatch(web)
-            const origin = origins.get(response) ?? latest
-            return HttpClientResponse.fromWeb(origin, response)
-          }).pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause))))),
-      }
       if (promptCacheSnapshots) {
         const current = PromptCacheDiagnostics.snapshot(request)
         const comparison = PromptCacheDiagnostics.compare(promptCacheSnapshots.get(session.id), current)
@@ -301,6 +306,7 @@ export const layer = Layer.effect(
       return {
         request,
         options,
+        webSocketEligible: middlewares.length === 0,
         executeTool,
         stepLimitReached,
       }

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/src/shell/select.ts

@@ -1,12 +1,15 @@
 export * as ShellSelect from "./select"
 
 import path from "path"
+import { spawn, type ChildProcess } from "child_process"
 import { readFile } from "fs/promises"
 import { statSync } from "fs"
+import { setTimeout } from "node:timers/promises"
 import { Schema } from "effect"
 import { FSUtil } from "@opencode-ai/util/fs-util"
 import { which } from "../util/which"
 
+const SIGKILL_TIMEOUT_MS = 200
 const META: Record<string, { deny?: boolean; login?: boolean; posix?: boolean; ps?: boolean }> = {
   bash: { login: true, posix: true },
   dash: { login: true, posix: true },
@@ -30,6 +33,37 @@ export const Options = Schema.Struct({
 })
 export type Options = typeof Options.Type
 
+export async function killTree(proc: ChildProcess, opts?: { exited?: () => boolean }): Promise<void> {
+  const pid = proc.pid
+  if (!pid || opts?.exited?.()) return
+
+  if (process.platform === "win32") {
+    await new Promise<void>((resolve) => {
+      const killer = spawn("taskkill", ["/pid", String(pid), "/f", "/t"], {
+        stdio: "ignore",
+        windowsHide: true,
+      })
+      killer.once("exit", () => resolve())
+      killer.once("error", () => resolve())
+    })
+    return
+  }
+
+  try {
+    process.kill(-pid, "SIGTERM")
+    await setTimeout(SIGKILL_TIMEOUT_MS)
+    if (!opts?.exited?.()) {
+      process.kill(-pid, "SIGKILL")
+    }
+  } catch {
+    proc.kill("SIGTERM")
+    await setTimeout(SIGKILL_TIMEOUT_MS)
+    if (!opts?.exited?.()) {
+      proc.kill("SIGKILL")
+    }
+  }
+}
+
 function stat(file: string) {
   return statSync(file, { throwIfNoEntry: false }) ?? undefined
 }

packages/core/src/snapshot.ts

@@ -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,
   }),
 )
 

packages/core/test/bus.test.ts

@@ -1298,4 +1298,24 @@ describe("Bus", () => {
     }),
   )
 
+  it.effect("sequences returns the latest committed seq per aggregate and omits unknown aggregates", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const first = Session.ID.create()
+      const second = Session.ID.create()
+      yield* bus.publish(DurableMessage, durableData(first, "zero"))
+      yield* bus.publish(DurableMessage, durableData(first, "one"))
+      yield* bus.publish(DurableMessage, durableData(second, "zero"))
+
+      const sequences = yield* bus.sequences([first, second, Session.ID.create()])
+
+      expect(sequences).toEqual(
+        new Map([
+          [first, Event.Seq.make(1)],
+          [second, Event.Seq.make(0)],
+        ]),
+      )
+      expect(yield* bus.sequences([])).toEqual(new Map())
+    }),
+  )
 })

packages/core/test/formatter.test.ts

@@ -56,22 +56,52 @@ function withTemp<A, E, R>(body: (directory: string) => Effect.Effect<A, E, R>)
 }
 
 describe("Formatter", () => {
-  it.live("does not run formatters marked as disabled in config", () =>
+  it.live("status() returns empty list when no formatters are configured", () =>
     withTemp((directory) =>
-      Effect.gen(function* () {
+      Formatter.Service.use((formatter) => formatter.status()).pipe(Effect.provide(formatterLayer(directory))),
-        const file = path.join(directory, "test.disabled")
+    ),
-        expect(yield* Formatter.Service.use((formatter) => formatter.file(file))).toBe(false)
+  )
-      }).pipe(
+
-        Effect.provide(
+  it.live("status() returns built-in formatters when formatter is true", () =>
-          formatterLayer(directory, {
+    withTemp((directory) =>
-            disabled: {
+      Formatter.Service.use((formatter) =>
-              disabled: true,
+        Effect.gen(function* () {
-              command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
+          const statuses = yield* formatter.status()
-              extensions: [".disabled"],
+          const gofmt = statuses.find((item) => item.name === "gofmt")
-            },
+          expect(gofmt).toBeDefined()
-          }),
+          expect(gofmt?.extensions).toContain(".go")
-        ),
+        }),
-      ),
+      ).pipe(Effect.provide(formatterLayer(directory, true))),
+    ),
+  )
+
+  it.live("status() keeps built-in formatters when config object is provided", () =>
+    withTemp((directory) =>
+      Formatter.Service.use((formatter) =>
+        Effect.gen(function* () {
+          const statuses = yield* formatter.status()
+          expect(statuses.find((item) => item.name === "gofmt")?.extensions).toContain(".go")
+          expect(statuses.find((item) => item.name === "mix")).toBeDefined()
+        }),
+      ).pipe(Effect.provide(formatterLayer(directory, { gofmt: {} }))),
+    ),
+  )
+
+  it.live("status() excludes formatters marked as disabled in config", () =>
+    withTemp((directory) =>
+      Formatter.Service.use((formatter) =>
+        Effect.gen(function* () {
+          const statuses = yield* formatter.status()
+          expect(statuses.find((item) => item.name === "gofmt")).toBeUndefined()
+          expect(statuses.find((item) => item.name === "mix")).toBeDefined()
+        }),
+      ).pipe(Effect.provide(formatterLayer(directory, { gofmt: { disabled: true } }))),
+    ),
+  )
+
+  it.live("service initializes without error", () =>
+    withTemp((directory) =>
+      Formatter.Service.use((formatter) => formatter.init()).pipe(Effect.provide(formatterLayer(directory))),
     ),
   )
 
@@ -85,29 +115,22 @@ describe("Formatter", () => {
     ),
   )
 
-  it.live("loads formatter state per directory", () =>
+  it.live("status() initializes formatter state per directory", () =>
-    withTemp((off) =>
+    Effect.acquireUseRelease(
-      withTemp((on) =>
+      Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
+      ([off, on]) =>
         Effect.gen(function* () {
-          const offFile = path.join(off, "test.isolated")
+          const disabled = yield* Formatter.Service.use((formatter) => formatter.status()).pipe(
-          const onFile = path.join(on, "test.isolated")
+            Effect.provide(formatterLayer(off.path, false)),
-          const disabled = yield* Formatter.Service.use((formatter) => formatter.file(offFile)).pipe(
-            Effect.provide(formatterLayer(off, false)),
           )
-          const enabled = yield* Formatter.Service.use((formatter) => formatter.file(onFile)).pipe(
+          const enabled = yield* Formatter.Service.use((formatter) => formatter.status()).pipe(
-            Effect.provide(
+            Effect.provide(formatterLayer(on.path, true)),
-              formatterLayer(on, {
-                isolated: {
-                  command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
-                  extensions: [".isolated"],
-                },
-              }),
-            ),
           )
-          expect(disabled).toBe(false)
+          expect(disabled).toEqual([])
-          expect(enabled).toBe(true)
+          expect(enabled.find((item) => item.name === "gofmt")).toBeDefined()
         }),
-      ),
+      (directories) =>
+        Effect.promise(() => Promise.all(directories.map((tmp) => tmp[Symbol.asyncDispose]())).then(() => undefined)),
     ),
   )
 

packages/core/test/git.test.ts

@@ -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")

packages/core/test/plugin/provider-openai.test.ts

@@ -12,7 +12,6 @@ import { PluginHost } from "@opencode-ai/core/plugin/host"
 import { PluginHooks } from "@opencode-ai/core/plugin/hooks"
 import { OpenAIPlugin } from "@opencode-ai/core/plugin/provider/openai"
 import { Provider } from "@opencode-ai/core/provider"
-import type { SessionHttpHandler } from "@opencode-ai/plugin/effect/session"
 import { testEffect } from "../lib/effect"
 import { PluginTestLayer } from "./fixture"
 
@@ -30,27 +29,15 @@ function required<T>(value: T | undefined): T {
   return value
 }
 
-const http = Effect.fn(function* (providerID: Provider.ID, url: string) {
+const httpMiddlewareCount = Effect.fn(function* () {
   const middlewares: Parameters<PluginHooks.Domains["session"]["http"]["use"]>[0][] = []
   yield* (yield* PluginHooks.Service).trigger("session", "http", {
     sessionID: Session.ID.make("ses_test"),
     agent: Agent.ID.make("build"),
-    model: Model.Ref.make({ providerID, id: Model.ID.make("gpt-5.5") }),
+    model: Model.Ref.make({ providerID: Provider.ID.openai, id: Model.ID.make("gpt-5.5") }),
-    use: (item) =>
+    use: (middleware) => Effect.sync(() => middlewares.push(middleware)),
-      Effect.sync(() => {
-        middlewares.push(item)
-      }),
   })
-  const request = middlewares.reduce<SessionHttpHandler>(
+  return middlewares.length
-    (next, item) => (input: Request) => item(input, next),
-    (input: Request) => {
-      const headers = new Headers(input.headers)
-      headers.set("x-seen-url", input.url)
-      return Effect.succeed(new Response(null, { headers }))
-    },
-  )
-  const response = yield* request(new Request(url, { method: "POST", body: "{}" }))
-  return { url: response.headers.get("x-seen-url"), headers: Object.fromEntries(response.headers.entries()) }
 })
 
 describe("OpenAIPlugin", () => {
@@ -124,30 +111,18 @@ describe("OpenAIPlugin", () => {
       })
       yield* addPlugin()
 
-      const request = yield* http(Provider.ID.openai, "https://api.openai.com/v1/responses")
-      const custom = yield* http(Provider.ID.make("custom-openai"), "https://custom.example/v1/responses")
-      const proxy = yield* http(Provider.ID.openai, "https://proxy.example/v1/responses?region=us")
-
       const provider = required(yield* catalog.provider.get(Provider.ID.openai))
       expect(provider.package).toBe("@opencode-ai/ai/providers/openai")
       expect(provider.settings).toMatchObject({ baseURL: "https://chatgpt.com/backend-api/codex" })
-      expect(provider.headers).toMatchObject({ "chatgpt-account-id": "acct_123" })
+      expect(provider.headers).toMatchObject({ originator: "opencode", "chatgpt-account-id": "acct_123" })
-      expect(request.url).toBe("https://chatgpt.com/backend-api/codex/responses")
+      expect(yield* httpMiddlewareCount()).toBe(0)
-      expect(request.headers).toMatchObject({ originator: "opencode", "session-id": "ses_test" })
-      expect(custom.headers).not.toHaveProperty("originator")
-      expect(proxy.url).toBe("https://proxy.example/v1/responses?region=us")
-      expect(proxy.headers).toMatchObject({ originator: "opencode", "session-id": "ses_test" })
       const eligible = required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.5")))
       expect(eligible.package).toBe("@opencode-ai/ai/providers/openai")
       expect(eligible.cost).toEqual([])
       expect(eligible.limit).toEqual({ context: 272_000, input: 272_000, output: 128_000 })
       expect(eligible.enabled).toBe(true)
-      expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.5-pro"))).enabled).toBe(
+      expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.5-pro"))).enabled).toBe(false)
-        false,
+      expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.4-pro"))).enabled).toBe(false)
-      )
-      expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.4-pro"))).enabled).toBe(
-        false,
-      )
       expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.4"))).limit).toEqual({
         context: 272_000,
         input: 272_000,
@@ -184,15 +159,14 @@ describe("OpenAIPlugin", () => {
       })
       yield* addPlugin()
 
-      const request = yield* http(Provider.ID.openai, "https://api.openai.com/v1/responses")
+      const provider = required(yield* catalog.provider.get(Provider.ID.openai))
-
       const model = required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.5")))
       expect(model.package).toBe("@opencode-ai/ai/providers/openai")
       expect(model.enabled).toBe(true)
       expect(model.limit).toEqual({ context: 1_050_000, input: 922_000, output: 128_000 })
-      expect(request.headers).not.toHaveProperty("originator")
+      expect(provider.headers).not.toHaveProperty("originator")
+      expect(yield* httpMiddlewareCount()).toBe(0)
       expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-4.1"))).enabled).toBe(true)
     }),
   )
-
 })

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])
+  })
 })

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

@@ -41,15 +41,17 @@ describe("Session.log", () => {
   it.effect("replays public session events and marks synced at the aggregate watermark", () =>
     Effect.gen(function* () {
       const session = yield* Session.Service
+      const bus = yield* Bus.Service
       const created = yield* session.create({ location })
       yield* session.rename({ sessionID: created.id, title: "session.renamed" })
 
       const items = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id })))
+      const watermark = (yield* bus.sequences([created.id])).get(created.id)
 
       // Session creation commits a non-public durable event, so the marker's
       // seq covers more of the aggregate than the public events emitted.
       expect(items.map((item) => item.type)).toEqual(["session.renamed", "log.synced"])
-      expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: Event.Seq.make(1) })
+      expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: watermark })
     }),
   )
 

packages/core/test/session-model-request.test.ts

@@ -1,6 +1,9 @@
 import { describe, expect, test } from "bun:test"
 import { Message, ToolResultPart } from "@opencode-ai/ai"
-import { boundImages, unsupportedParts } from "@opencode-ai/core/session/model-request"
+import { boundImages, composeHttpMiddleware, unsupportedParts } from "@opencode-ai/core/session/model-request"
+import type { SessionHttpMiddleware } from "@opencode-ai/plugin/effect/session"
+import { Effect } from "effect"
+import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 
 const capabilities = (input: string[]) => ({ tools: true, input, output: ["text"] })
 
@@ -110,3 +113,60 @@ describe("SessionModelRequest.boundImages", () => {
     })
   })
 })
+
+describe("SessionModelRequest.composeHttpMiddleware", () => {
+  test("keeps WebSocket eligibility when no middleware is registered", () => {
+    expect(composeHttpMiddleware([])).toBeUndefined()
+  })
+
+  test("forces HTTP when middleware is registered", () => {
+    expect(composeHttpMiddleware([(request, next) => next(request)])).toBeFunction()
+  })
+
+  test("preserves middleware nesting order", async () => {
+    const order: string[] = []
+    const middleware =
+      (name: string): SessionHttpMiddleware =>
+      (request, next) =>
+        Effect.sync(() => order.push(`${name}:before`)).pipe(
+          Effect.andThen(next(request)),
+          Effect.tap(() => Effect.sync(() => order.push(`${name}:after`))),
+        )
+    const composed = composeHttpMiddleware([middleware("first"), middleware("second")])
+    if (!composed) throw new Error("Expected HTTP middleware")
+    const request = HttpClientRequest.post("https://provider.test/responses").pipe(
+      HttpClientRequest.bodyText("payload", "text/plain"),
+    )
+    const response = await Effect.runPromise(
+      composed(request, (sent) =>
+        Effect.sync(() => {
+          order.push("send")
+          return HttpClientResponse.fromWeb(sent, new Response("response"))
+        }),
+      ),
+    )
+
+    expect(order).toEqual(["second:before", "first:before", "send", "first:after", "second:after"])
+    expect(await Effect.runPromise(response.text)).toBe("response")
+  })
+
+  test("preserves a synthetic replacement response", async () => {
+    let sent = false
+    const composed = composeHttpMiddleware([() => Effect.succeed(new Response("synthetic", { status: 202 }))])
+    if (!composed) throw new Error("Expected HTTP middleware")
+    const request = HttpClientRequest.post("https://provider.test/responses")
+    const response = await Effect.runPromise(
+      composed(request, (input) =>
+        Effect.sync(() => {
+          sent = true
+          return HttpClientResponse.fromWeb(input, new Response("network"))
+        }),
+      ),
+    )
+
+    expect(sent).toBe(false)
+    expect(response.status).toBe(202)
+    expect(response.request.url).toBe(request.url)
+    expect(await Effect.runPromise(response.text)).toBe("synthetic")
+  })
+})

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

@@ -897,6 +897,26 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("collects session HTTP middleware once per prepared request", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const hooks = yield* PluginHooks.Service
+      let triggers = 0
+      yield* hooks.register("session", "http", (event) =>
+        Effect.gen(function* () {
+          triggers++
+          yield* event.use((request, next) => next(request))
+        }),
+      )
+      yield* admit(session, "Use HTTP middleware")
+      yield* TestLLM.push(TestLLM.text("Done", "text-http-middleware"))
+
+      yield* session.resume(sessionID)
+
+      expect(triggers).toBe(1)
+    }),
+  )
+
   it.effect("executes a tool renamed by a session context hook", () =>
     Effect.gen(function* () {
       const session = yield* setup

packages/core/test/snapshot.test.ts

@@ -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) {

packages/server/src/auth.ts

@@ -2,6 +2,10 @@ export * as ServerAuth from "./auth"
 
 import { Context, Layer, Option, Redacted } from "effect"
 
+export type Credentials = {
+  password?: string
+}
+
 export type DecodedCredentials = {
   readonly username: string
   readonly password: Redacted.Redacted
@@ -33,3 +37,16 @@ export function authorized(credentials: DecodedCredentials, config: Info) {
     Redacted.value(credentials.password) === config.password.value
   )
 }
+
+export function header(credentials?: Credentials) {
+  const password = credentials?.password
+  if (!password) return undefined
+
+  return `Basic ${Buffer.from(`opencode:${password}`).toString("base64")}`
+}
+
+export function headers(credentials?: Credentials) {
+  const authorization = header(credentials)
+  if (!authorization) return undefined
+  return { Authorization: authorization }
+}

packages/server/test/auth.test.ts

@@ -7,3 +7,7 @@ test("accepts only the fixed opencode username", () => {
   expect(ServerAuth.authorized({ username: "opencode", password: Redacted.make("secret") }, config)).toBe(true)
   expect(ServerAuth.authorized({ username: "custom", password: Redacted.make("secret") }, config)).toBe(false)
 })
+
+test("encodes the fixed opencode username", () => {
+  expect(ServerAuth.header({ password: "secret" })).toBe(`Basic ${Buffer.from("opencode:secret").toString("base64")}`)
+})

packages/tui/src/routes/session/index.tsx

@@ -96,7 +96,6 @@ import { findMessageBoundary, messageNavigationSlack } from "./message-navigatio
 import { stringWidth } from "../../util/string-width"
 import { useArgs } from "../../context/args"
 import { withTimestampedFallback } from "@opencode-ai/util/session-title-fallback"
-import { installSyntaxHighlightCache } from "../../util/syntax-highlight-cache"
 
 addDefaultParsers(parsers.parsers)
 
@@ -129,7 +128,6 @@ function use() {
 }
 
 export function Session() {
-  installSyntaxHighlightCache()
   const setEpilogue = useEpilogue()
   const clipboard = useClipboard()
   const writeExport = async (file: string, content: string) => {

packages/tui/src/util/syntax-highlight-cache.ts

@@ -1,38 +0,0 @@
-import { getTreeSitterClient, type TreeSitterClient } from "@opentui/core"
-
-const CACHE_SIZE = 500
-const installed = new WeakSet<TreeSitterClient>()
-
-export function installSyntaxHighlightCache() {
-  const client = getTreeSitterClient()
-  if (installed.has(client)) return
-  installed.add(client)
-  client.highlightOnce = cacheHighlights(client.highlightOnce.bind(client))
-}
-
-export function cacheHighlights(highlight: TreeSitterClient["highlightOnce"], capacity = CACHE_SIZE) {
-  const cache = new Map<string, ReturnType<TreeSitterClient["highlightOnce"]>>()
-
-  return (content: string, filetype: string) => {
-    const key = `${filetype}\0${content}`
-    const cached = cache.get(key)
-    if (cached) {
-      cache.delete(key)
-      cache.set(key, cached)
-      return cached
-    }
-
-    const result = highlight(content, filetype)
-    cache.set(key, result)
-    if (cache.size > capacity) cache.delete(cache.keys().next().value!)
-
-    void result
-      .then((value) => {
-        if (value.error && cache.get(key) === result) cache.delete(key)
-      })
-      .catch(() => {
-        if (cache.get(key) === result) cache.delete(key)
-      })
-    return result
-  }
-}

packages/tui/test/util/syntax-highlight-cache.test.ts

@@ -1,69 +0,0 @@
-import { describe, expect, test } from "bun:test"
-import { cacheHighlights } from "../../src/util/syntax-highlight-cache"
-
-describe("syntax highlight cache", () => {
-  test("reuses completed and in-flight highlights", async () => {
-    let calls = 0
-    const highlight = cacheHighlights(async () => {
-      calls++
-      return { highlights: [[0, 5, "keyword"]] }
-    })
-
-    const first = highlight("const", "typescript")
-    const second = highlight("const", "typescript")
-
-    expect(second).toBe(first)
-    expect(await second).toEqual({ highlights: [[0, 5, "keyword"]] })
-    expect(await highlight("const", "typescript")).toEqual({ highlights: [[0, 5, "keyword"]] })
-    expect(calls).toBe(1)
-  })
-
-  test("evicts least recently used highlights", async () => {
-    let calls = 0
-    const highlight = cacheHighlights(async () => {
-      calls++
-      return { highlights: [] }
-    }, 2)
-
-    await highlight("one", "text")
-    await highlight("two", "text")
-    await highlight("one", "text")
-    await highlight("three", "text")
-    await highlight("two", "text")
-
-    expect(calls).toBe(4)
-  })
-
-  test("retries failed highlights", async () => {
-    let calls = 0
-    const highlight = cacheHighlights(async () => {
-      calls++
-      if (calls === 1) return { error: "parser unavailable" }
-      return { highlights: [] }
-    })
-
-    await highlight("const", "typescript")
-    await highlight("const", "typescript")
-
-    expect(calls).toBe(2)
-  })
-
-  test("an evicted failure does not delete its replacement", async () => {
-    const pending = Promise.withResolvers<{ highlights: [] }>()
-    let calls = 0
-    const highlight = cacheHighlights(() => {
-      calls++
-      if (calls === 1) return pending.promise
-      return Promise.resolve({ highlights: [] })
-    }, 1)
-
-    const stale = highlight("one", "text")
-    await highlight("two", "text")
-    const current = highlight("one", "text")
-    pending.reject(new Error("parser unavailable"))
-
-    await expect(stale).rejects.toThrow("parser unavailable")
-    expect(highlight("one", "text")).toBe(current)
-    expect(calls).toBe(3)
-  })
-})