fix(cli): sign macOS preview binaries

.github/workflows/publish.yml

@@ -124,12 +124,66 @@ jobs:
 
       - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
         with:
-          name: opencode-preview-cli
+          name: opencode-preview-cli-unsigned
           path: packages/cli/dist/cli-*
 
     outputs:
       version: ${{ needs.version.outputs.version }}
 
+  sign-cli-macos:
+    needs: build-cli
+    runs-on: macos-26
+    if: github.repository == 'anomalyco/opencode'
+    steps:
+      - uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3.6.0
+
+      - uses: apple-actions/import-codesign-certs@8f3fb608891dd2244cdab3d69cd68c0d37a7fe93 # v2.0.0
+        with:
+          keychain: build
+          p12-file-base64: ${{ secrets.APPLE_CERTIFICATE }}
+          p12-password: ${{ secrets.APPLE_CERTIFICATE_PASSWORD }}
+
+      - uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
+        with:
+          name: opencode-preview-cli-unsigned
+          path: packages/cli/dist
+
+      - name: Sign macOS CLI binaries
+        run: |
+          identity=$(security find-identity -v -p codesigning build.keychain | sed -n 's/.*"\(Developer ID Application:.*\)"/\1/p' | head -n 1)
+          if [ -z "$identity" ]; then
+            echo "Developer ID Application identity not found"
+            exit 1
+          fi
+
+          found=0
+          for file in packages/cli/dist/cli-darwin-*/bin/opencode2; do
+            if [ ! -f "$file" ]; then
+              continue
+            fi
+            found=1
+            codesign \
+              --force \
+              --timestamp \
+              --options runtime \
+              --entitlements packages/cli/script/entitlements.plist \
+              --sign "$identity" \
+              "$file"
+            codesign --verify --deep --strict --verbose=4 "$file"
+            codesign --display --requirements - "$file"
+          done
+
+          if [ "$found" -eq 0 ]; then
+            echo "No macOS CLI binaries found"
+            exit 1
+          fi
+
+      - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+        with:
+          name: opencode-preview-cli
+          path: packages/cli/dist/cli-*
+          if-no-files-found: error
+
   build-node-cli:
     needs: version
     if: github.repository == 'anomalyco/opencode'
@@ -471,6 +525,7 @@ jobs:
     needs:
       - version
       - build-cli
+      - sign-cli-macos
       - build-node-cli
       - sign-cli-windows
       - build-electron

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 supports multiple physical transports, selection remains execution policy below its semantic route. OpenAI Responses uses a purpose-built hybrid transport that prepares one final request, executes HTTP by default, and passes a generic channel exchange to a per-call `WebSocketChannelExecutor` when supplied. `Route.streamPrepared` owns decoding and acknowledges channel completion only after successful full consumption.
+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.
 
 ### 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` and `chat`.
+- Use `model` for the default API path and named methods for provider-native alternatives such as OpenAI `responses`, `responsesWebSocket`, 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,10 +124,11 @@ 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`. 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.
+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`.
 
 Do not expose `Route` in provider package settings. Route composition stays an implementation detail behind `model(...)`.
 
@@ -153,10 +154,9 @@ packages/ai/src/
     auth-options.ts         ProviderAuthOption shape, AuthOptions.bearer, AtLeastOne helper
     framing.ts              Framing type + Framing.sse
     transport/              transport implementations
-      index.ts              Transport execution types + HttpTransport / WebSocketTransport namespaces
-      websocket-channel.ts  generic sequential channel executor/driver contract
+      index.ts              Transport type + HttpTransport / WebSocketTransport namespaces
       http.ts               HttpTransport.httpJson — POST + framing
-      websocket.ts          direct one-request channel executor + raw socket adapter
+      websocket.ts          WebSocketTransport.json + WebSocketExecutor service
   protocols/
     shared.ts               ProviderShared toolkit used inside protocol impls
     openai-chat.ts          protocol + route (compose OpenAIChat.protocol)

packages/ai/README.md

@@ -315,6 +315,7 @@ 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 },
 })
@@ -331,7 +332,7 @@ OpenAI Chat and OpenAI Responses are separate semantic entrypoints:
 - `@opencode-ai/ai/providers/google-vertex/responses`
 - `@opencode-ai/ai/providers/google-vertex/messages`
 
-OpenAI Responses has one semantic route and uses HTTP by default. Advanced callers may supply a per-call WebSocket channel executor through `StreamOptions`; transport policy does not change provider settings, model identity, or route identity. Azure follows the same Chat/Responses split at `providers/azure/chat` and `providers/azure/responses`. Generic OpenAI-compatible Chat remains at `providers/openai-compatible`; the Responses adapter at `providers/openai-compatible/responses` uses the provider-neutral Open Responses protocol. OpenAI Responses extends that baseline with OpenAI tools, event variants, metadata, and defaults. Generic Anthropic Messages-compatible providers use `providers/anthropic-compatible`, which the named Anthropic provider composes. Google Gemini and Amazon Bedrock expose their single native API through their existing provider paths.
+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.
 
 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-08-07
+Last reviewed: 2026-07-24
 
 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,7 +16,8 @@ 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                   | `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 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 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.                                                                                                                                                                                                                  |
@@ -47,19 +48,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/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/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-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/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/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/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 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-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/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/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/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/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.                                                                             |
 
 ## Highest-Risk Gaps
 

packages/ai/example/call-sites.md

@@ -67,6 +67,7 @@ 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")
@@ -249,6 +250,11 @@ 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),
@@ -260,11 +266,13 @@ 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,
@@ -334,19 +342,22 @@ const response =
   )
 ```
 
-For direct provider-facade calls, Responses has one semantic model and route:
+For direct provider-facade calls, HTTP versus WebSocket is represented as named
+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 has the same transport-neutral
-`model(...)` contract:
+The package-like OpenAI Responses entrypoint instead keeps transport scoped to
+Responses settings while preserving the same `model(...)` contract:
 
 ```ts
 import { model } from "@opencode-ai/ai/providers/openai/responses"
 
-model("gpt-4o", { apiKey })
+model("gpt-4o", { apiKey, transport: "websocket" })
 ```
 
 Vertex keeps Gemini, Chat, Responses, and Messages as separate package-like entrypoints,
@@ -488,13 +499,16 @@ generic dynamic resolver:
 const model =
   providerID === "azure"
     ? Azure.configure(resolvedAzureConfig).responses(apiModelID)
-    : OpenAI.responses(apiModelID)
+    : endpoint.websocket
+      ? OpenAI.responsesWebSocket(apiModelID)
+      : OpenAI.responses(apiModelID)
 ```
 
 That boundary can branch on durable config/catalog metadata and call typed
-provider APIs directly. Transport selection remains execution policy: a Session
-or other caller may pass a WebSocket channel executor per call without changing
-the model constructed by this boundary.
+provider APIs directly. A direct provider-facade boundary maps metadata like
+`endpoint.websocket` to `OpenAI.responsesWebSocket(apiModelID)`. A package-loading
+boundary passes `transport: "websocket"` to the OpenAI Responses entrypoint.
+The client runtime only executes the route carried by the resulting model.
 
 ## Competitive Shape
 
@@ -530,8 +544,9 @@ 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 setting on a provider or executable model. OpenAI Responses uses
-  HTTP by default and accepts an optional per-call channel executor as execution policy.
+- No transport override on an executable model or request. Direct provider
+  facades use `responses` versus `responsesWebSocket`; the package-like Responses
+  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
@@ -565,10 +580,12 @@ 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` accepts
-      optional per-call channel execution without changing route identity.
+- [x] Collapse the public WebSocket runtime split so one `LLMClient.layer`
+      exposes available transport capabilities and selected routes fail with typed
+      transport config errors when a required capability is missing.
 - [x] Convert OpenAI provider APIs to provider-facade shape:
-      `OpenAI.configure(config).responses(id)` and `.chat(id)`.
+      `OpenAI.configure(config).responses(id)`, `.chat(id)`, and
+      `.responsesWebSocket(id)`.
 - [x] Convert Azure to a configured facade where resource/base URL/api version
       setup happens before selecting deployment ids.
 - [x] Split Cloudflare products into separate facades such as

packages/ai/example/tutorial.ts

@@ -1,6 +1,6 @@
 import { Config, Effect, Formatter, Layer, Schema, Stream } from "effect"
 import { LLM, LLMClient, LLMRequest, Message, ProviderID, Tool, ToolRuntime } from "@opencode-ai/ai"
-import { Route, Auth, Endpoint, Framing, Protocol, RequestExecutor } from "@opencode-ai/ai/route"
+import { Route, Auth, Endpoint, Framing, Protocol, RequestExecutor, WebSocketExecutor } from "@opencode-ai/ai/route"
 import { OpenAI } from "@opencode-ai/ai/providers"
 
 /**
@@ -214,7 +214,8 @@ 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 llmClientLayer = LLMClient.layer.pipe(Layer.provide(requestExecutorLayer))
+const llmDeps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
+const llmClientLayer = LLMClient.layer.pipe(Layer.provide(llmDeps))
 
 const program = Effect.gen(function* () {
   // yield* generateOnce
@@ -222,6 +223,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(requestExecutorLayer, llmClientLayer)))
+}).pipe(Effect.provide(Layer.mergeAll(llmDeps, llmClientLayer)))
 
 Effect.runPromise(program)

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

@@ -211,43 +211,11 @@ 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,
@@ -272,9 +240,6 @@ 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)],
 )
@@ -667,9 +632,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` and `error` are hard failures. All four end
-// the stream, so keep this set aligned with `step` and the protocol's terminal predicate.
-const TERMINAL_TYPES = new Set(["error", "response.completed", "response.incomplete", "response.failed"])
+// `finish` event; `response.failed` is a hard failure. All three end the stream,
+// so keep this set aligned with `step` and the protocol's terminal predicate.
+const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"])
 export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type)
 
 const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => {
@@ -1001,24 +966,16 @@ const providerErrorMessage = (event: Event, fallback: string): string => {
   return message || code || fallback
 }
 
-export const providerFailure = (id: string, event: Event, fallback: string) => {
+const providerError = (state: ParserState, event: Event, fallback: string) => {
   const code = event.code || event.error?.code || event.response?.error?.code || undefined
   const message = providerErrorMessage(event, fallback)
-  const status =
-    typeof event.status === "number"
-      ? event.status
-      : typeof event.status_code === "number"
-        ? event.status_code
-        : undefined
   return new AIError({
-    module: id,
+    module: state.id,
     method: "stream",
-    reason: classifyProviderFailure({ message, code, status }),
+    reason: classifyProviderFailure({ message, code }),
   })
 }
 
-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`)
@@ -1058,11 +1015,7 @@ 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 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`)),
-    )
+  if (event.type === "error") return providerError(state, event, `${state.name} stream error`)
   return Effect.succeed<StepResult>([state, NO_EVENTS])
 }
 

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

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

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

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

packages/ai/src/provider-error.ts

@@ -67,7 +67,6 @@ 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, OpenAIChat.route]
+export const routes = [OpenAIResponses.route, OpenAIResponses.webSocketRoute, 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,6 +63,7 @@ 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
 }
 
@@ -81,12 +82,17 @@ 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) =>
@@ -105,6 +111,7 @@ export const configure = (input: Config = {}) => {
     id,
     model: responses,
     responses,
+    responsesWebSocket,
     chat,
     image,
     configure,
@@ -131,7 +138,10 @@ const config = (settings: Settings): Config => {
 }
 
 export const model: ProviderPackage.Definition<Settings, OpenAIProviderOptionsInput>["model"] = (modelID, settings) => {
-  return configure(config(settings)).responses(modelID)
+  const configured = configure(config(settings))
+  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"] = (
@@ -139,5 +149,6 @@ 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,10 +1,12 @@
 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, WebSocketChannelExecutor } from "./transport"
+import type { HttpMiddleware, Transport, TransportRuntime } from "./transport"
+import { WebSocketExecutor } from "./transport"
 import type { Protocol } from "./protocol"
 import { applyCachePolicy } from "../cache-policy"
 import * as ProviderShared from "../protocols/shared"
@@ -56,7 +58,6 @@ export interface Route<Body, Prepared = unknown> {
     prepared: Prepared,
     request: LLMRequest,
     runtime: TransportRuntime,
-    options?: StreamOptions,
   ) => Stream.Stream<LLMEvent, AIError>
 }
 
@@ -156,7 +157,6 @@ export interface Interface {
 
 export interface StreamOptions {
   readonly http?: HttpMiddleware
-  readonly webSocket?: WebSocketChannelExecutor
 }
 
 export interface StreamMethod {
@@ -255,7 +255,13 @@ const requireTerminalEvent = (route: string) => (events: Stream.Stream<LLMEvent,
         if (LLMEvent.is.finish(event) || LLMEvent.is.providerError(event)) terminal = true
         return Effect.succeed(event)
       }),
-      Stream.onEnd(Effect.suspend(() => (terminal ? Effect.void : Effect.fail(incompleteStreamError(route))))),
+      Stream.onEnd(
+        Effect.suspend(() =>
+          terminal
+            ? Effect.void
+            : Effect.fail(incompleteStreamError(route)),
+        ),
+      ),
     )
   })
 
@@ -314,29 +320,23 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
           encodeBody,
           headers: routeInput.headers,
           middleware: options?.http,
-          webSocket: options?.webSocket,
         }),
-      streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime, options?: StreamOptions) => {
+      streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => {
         const route = `${request.model.provider}/${request.model.route.id}`
-        return Stream.unwrap(
-          routeInput.transport.execute(prepared, request, runtime, options).pipe(
-            Effect.map((execution) => {
-              const events = execution.frames.pipe(
-                Stream.mapEffect(decodeEvent(route)),
-                protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
-              )
-              const stream = events.pipe(
-                Stream.mapAccumEffect(
-                  () => protocol.stream.initial(request),
-                  protocol.stream.step,
-                  protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined,
-                ),
-                Stream.catchCause((cause) => Stream.fail(streamError(route, `Failed to read ${route} stream`, cause))),
-                requireTerminalEvent(route),
-              )
-              return execution.complete ? stream.pipe(Stream.onEnd(execution.complete)) : stream
-            }),
+        const events = routeInput.transport
+          .frames(prepared, request, runtime)
+          .pipe(
+            Stream.mapEffect(decodeEvent(route)),
+            protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,
+          )
+        return events.pipe(
+          Stream.mapAccumEffect(
+            () => protocol.stream.initial(request),
+            protocol.stream.step,
+            protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined,
           ),
+          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, options)
+      return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime)
     }),
   )
 
@@ -457,6 +457,7 @@ 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,28 +16,11 @@ export { AuthOptions } from "./auth-options"
 export { Endpoint } from "./endpoint"
 export { Framing } from "./framing"
 export { Protocol } from "./protocol"
-export { HttpTransport, WebSocketTransport } from "./transport"
+export { HttpTransport, WebSocketExecutor, 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 {
-  ChannelCheckpoint,
-  ChannelCreate,
-  ChannelObservation,
-  HttpHandler,
-  HttpMiddleware,
-  Transport as TransportDef,
-  TransportExecuteOptions,
-  TransportExecution,
-  TransportRuntime,
-  WebSocketConnection,
-  WebSocketChannelDriver,
-  WebSocketChannelExchange,
-  WebSocketChannelExecution,
-  WebSocketChannelExecutor,
-  WebSocketConnector,
-  WebSocketRequest,
-} from "./transport"
+export type { HttpHandler, HttpMiddleware, Transport as TransportDef, TransportRuntime } from "./transport"

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

@@ -86,28 +86,26 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
         middleware: prepareInput.middleware,
       }
     }),
-  execute: (prepared, request, runtime) =>
-    Effect.succeed({
-      frames: Stream.unwrap(
-        runtime.http
-          .execute(prepared.request, prepared.middleware)
-          .pipe(
-            Effect.map((response) =>
-              prepared.framing.frame(
-                response.stream.pipe(
-                  Stream.mapError((error) =>
-                    ProviderShared.eventError(
-                      `${request.model.provider}/${request.model.route.id}`,
-                      `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
-                      ProviderShared.errorText(error),
-                    ),
+  frames: (prepared, request, runtime) =>
+    Stream.unwrap(
+      runtime.http
+        .execute(prepared.request, prepared.middleware)
+        .pipe(
+          Effect.map((response) =>
+            prepared.framing.frame(
+              response.stream.pipe(
+                Stream.mapError((error) =>
+                  ProviderShared.eventError(
+                    `${request.model.provider}/${request.model.route.id}`,
+                    `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
+                    ProviderShared.errorText(error),
                   ),
                 ),
               ),
             ),
           ),
-      ),
-    }),
+        ),
+    ),
 })
 
 export const sseJson = {

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

@@ -1,33 +1,19 @@
-import type { Effect, Scope, Stream } from "effect"
+import type { Effect, Stream } from "effect"
 import { Endpoint } from "../endpoint"
 import { Auth } from "../auth"
 import type { HttpMiddleware, Interface as RequestExecutorInterface } from "../executor"
-import type { WebSocketChannelExecutor } from "./websocket-channel"
+import type { Interface as WebSocketExecutorInterface } from "./websocket"
 import type { AIError, LLMRequest } from "../../schema"
 
 export interface TransportRuntime {
   readonly http: RequestExecutorInterface
-}
-
-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
+  readonly webSocket?: WebSocketExecutorInterface
 }
 
 export interface Transport<Body, Prepared, Frame> {
   readonly id: string
   readonly prepare: (input: TransportPrepareInput<Body>) => Effect.Effect<Prepared, AIError>
-  readonly execute: (
-    prepared: Prepared,
-    request: LLMRequest,
-    runtime: TransportRuntime,
-    options?: TransportExecuteOptions,
-  ) => Effect.Effect<TransportExecution<Frame>, AIError, Scope.Scope>
+  readonly frames: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => Stream.Stream<Frame, AIError>
 }
 
 export interface TransportPrepareInput<Body> {
@@ -38,19 +24,8 @@ 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 type {
-  ChannelCheckpoint,
-  ChannelCreate,
-  ChannelObservation,
-  WebSocketChannelDriver,
-  WebSocketChannelExchange,
-  WebSocketChannelExecution,
-  WebSocketChannelExecutor,
-} from "./websocket-channel"
-export type { WebSocketConnection, WebSocketConnector, WebSocketRequest } from "./websocket"
-export { WebSocketTransport } from "./websocket"
+export { WebSocketExecutor, WebSocketTransport } from "./websocket"

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

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

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

@@ -1,15 +1,8 @@
-import { Cause, Effect, Queue, Stream } from "effect"
+import { Cause, Context, Effect, Layer, 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
@@ -22,57 +15,28 @@ export interface WebSocketConnection {
   readonly close: Effect.Effect<void, never>
 }
 
-export interface WebSocketConnector {
+export interface Interface {
   readonly open: (input: WebSocketRequest) => Effect.Effect<WebSocketConnection, AIError>
 }
 
-type WebSocketConstructorWithHeaders = (
+type WebSocketConstructorWithHeaders = new (
   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
-    readonly phase?: TransportReason["phase"]
-    readonly delivery?: TransportReason["delivery"]
-  } = {},
+  input: { readonly url?: string; readonly kind?: string } = {},
 ) =>
   new AIError({
-    module: "WebSocketConnector",
+    module: "WebSocketExecutor",
     method,
-    reason: new TransportReason({
-      message,
-      url: input.url,
-      kind: input.kind,
-      phase: input.phase,
-      delivery: input.delivery,
-    }),
+    reason: new TransportReason({ message, url: input.url, kind: input.kind }),
   })
 
-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
@@ -92,8 +56,6 @@ 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",
       }),
     )
   }
@@ -117,12 +79,7 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
       cleanup()
       resume(
         Effect.fail(
-          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
-            url: input.url,
-            kind: "open",
-            phase: "connect",
-            delivery: "not-sent",
-          }),
+          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
         ),
       )
     }
@@ -133,8 +90,6 @@ 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",
           }),
         ),
       )
@@ -146,7 +101,7 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
   })
 }
 
-export const toWebSocketUrl = (value: string) =>
+const webSocketUrl = (value: string) =>
   Effect.try({
     try: () => {
       const url = new URL(value)
@@ -164,31 +119,21 @@ export const toWebSocketUrl = (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.gen(function* () {
-    const constructor = yield* Socket.WebSocketConstructor
-    const ws = yield* Effect.try({
-      try: () =>
-        // Platform implementations may extend Effect's browser-compatible constructor with handshake options.
-        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
-        (constructor as unknown as WebSocketConstructorWithHeaders)(input.url, {
-          headers: input.headers,
-        }),
-      catch: (error) =>
-        transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
-          url: input.url,
-          kind: "open",
-          phase: "connect",
-          delivery: "not-sent",
-        }),
-    })
-    return yield* fromWebSocket(ws, input)
-  })
+  Effect.try({
+    try: () =>
+      new (globalThis.WebSocket as unknown as WebSocketConstructorWithHeaders)(input.url, { headers: input.headers }),
+    catch: (error) =>
+      transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
+        url: input.url,
+        kind: "open",
+      }),
+  }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
+
+export const layer: Layer.Layer<Service> = Layer.succeed(Service, Service.of({ open }))
 
 export const fromWebSocket = (
   ws: globalThis.WebSocket,
@@ -205,11 +150,7 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", "Unsupported WebSocket message payload", {
-            url: input.url,
-            kind: "message",
-            phase: "receive",
-          }),
+          transportError("message", "Unsupported WebSocket message payload", { url: input.url, kind: "message" }),
         ),
       )
     }
@@ -217,23 +158,16 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket error: ${eventMessage(event)}`, {
-            url: input.url,
-            kind: "message",
-            phase: "receive",
-          }),
+          transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, kind: "message" }),
         ),
       )
     }
     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",
-            phase: "close",
-          }),
+          transportError("message", `WebSocket closed with code ${event.code}`, { url: input.url, kind: "close" }),
         ),
       )
     }
@@ -255,8 +189,6 @@ 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),
@@ -274,57 +206,6 @@ 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
@@ -351,42 +232,32 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
         ...prepareInput,
       })
       return {
-        url: yield* toWebSocketUrl(parts.url),
+        url: yield* webSocketUrl(parts.url),
         headers: parts.headers,
         message: input.encodeMessage(yield* input.toMessage(parts.jsonBody)),
       }
     }),
-  execute: (prepared, request, _runtime, options) => {
-    const webSocket = options?.webSocket
+  frames: (prepared, _request, runtime) => {
+    const webSocket = runtime.webSocket
     if (!webSocket) {
-      return Effect.fail(
-        transportError("json", "WebSocket JSON transport requires StreamOptions.webSocket", {
+      return Stream.fail(
+        transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
           url: prepared.url,
           kind: "websocket",
-          phase: "prepare",
-          delivery: "not-sent",
         }),
       )
     }
-    const driver: WebSocketChannelDriver = {
-      create: () => Effect.succeed({ message: prepared.message, mode: "full" }),
-      observe: (_create, frame) => Effect.succeed({ type: "frame", frame }),
-    }
-    const exchange: WebSocketChannelExchange = {
-      id: request.id ?? "request",
-      connect: { url: prepared.url, headers: prepared.headers },
-      fallback: () =>
-        Stream.fail(
-          transportError("fallback", "WebSocket JSON transport does not provide HTTP fallback", {
-            url: prepared.url,
-            kind: "websocket",
-            phase: "fallback",
-            delivery: "not-sent",
-          }),
-        ),
-      driver,
-    }
-    return webSocket.execute(exchange)
+    const decoder = new TextDecoder()
+    return Stream.unwrap(
+      Effect.gen(function* () {
+        const connection = yield* Effect.acquireRelease(
+          webSocket.open({ url: prepared.url, headers: prepared.headers }),
+          (connection) => connection.close,
+        )
+        yield* connection.sendText(prepared.message)
+        return connection.messages.pipe(Stream.map((message) => messageText(message, decoder)))
+      }),
+    )
   },
 })
 
@@ -395,13 +266,15 @@ export const jsonTransport = {
   with: json,
 } as const
 
-export const WebSocketTransport = {
-  json,
-  jsonTransport,
-  direct,
-  makeDirect,
+export const WebSocketExecutor = {
+  Service,
+  layer,
   open,
   fromWebSocket,
   messageText,
-  toWebSocketUrl,
+} as const
+
+export const WebSocketTransport = {
+  json,
+  jsonTransport,
 } as const

packages/ai/src/schema/errors.ts

@@ -98,13 +98,6 @@ 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,11 +1,10 @@
 import { describe, expect } from "bun:test"
-import { Deferred, Effect, Fiber, Layer, Ref, Stream } from "effect"
+import { Effect, Layer, Ref } from "effect"
 import { Headers, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 import { LLM, AIError } from "../src"
-import { LLMClient, RequestExecutor, WebSocketTransport, type WebSocketChannelExecutor } from "../src/route"
+import { LLMClient, RequestExecutor } from "../src/route"
 import * as OpenAIChat from "../src/protocols/openai-chat"
-import * as OpenAI from "../src/providers/openai"
-import { dynamicResponse, fixedResponse } from "./lib/http"
+import { dynamicResponse } from "./lib/http"
 import { deltaChunk } from "./lib/openai-chunks"
 import { sseRaw } from "./lib/sse"
 import { it } from "./lib/effect"
@@ -414,125 +413,3 @@ 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, WebSocketTransport } from "@opencode-ai/ai/route"
+import { Route, Protocol } from "@opencode-ai/ai/route"
 import { Provider as ProviderSubpath } from "@opencode-ai/ai/provider"
 import {
   CloudflareAIGateway,
@@ -37,7 +37,6 @@ 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 () => {
@@ -45,6 +44,7 @@ describe("public exports", () => {
 
     expect(OpenAI.model).toBeFunction()
     expect(OpenAI.provider.responses).toBe(OpenAI.responses)
+    expect(OpenAI.provider.responsesWebSocket).toBe(OpenAI.responsesWebSocket)
     expect(OpenAI.configure({ apiKey: "fixture" }).responses).toBeFunction()
     expect(OpenAICompatible.deepseek.model).toBeFunction()
     expect(
@@ -86,6 +86,7 @@ describe("public exports", () => {
     expect(OpenAICompatibleResponses.route.id).toBe("openai-compatible-responses")
     expect(OpenAICompatibleResponses.route.protocol).toBe("open-responses")
     expect(OpenAIResponses.route.id).toBe("openai-responses")
+    expect(OpenAIResponses.webSocketRoute.id).toBe("openai-responses-websocket")
     expect(AnthropicMessages.route.id).toBe("anthropic-messages")
   })
 })

packages/ai/test/lib/http.ts

@@ -1,8 +1,9 @@
 import { Effect, Layer, Ref } from "effect"
 import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
-import { LLMClient, RequestExecutor } from "../../src/route"
+import { LLMClient, RequestExecutor, WebSocketExecutor } 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
@@ -31,12 +32,13 @@ const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
     ),
   )
 
-export type RuntimeEnv = RequestExecutorService | LLMClientService
+export type RuntimeEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService
 
 export const runtimeLayer = (layer: Layer.Layer<HttpClient.HttpClient>): Layer.Layer<RuntimeEnv> => {
   const requestExecutorLayer = RequestExecutor.layer.pipe(Layer.provide(layer))
-  const llmClientLayer = LLMClient.layer.pipe(Layer.provide(requestExecutorLayer))
-  return Layer.mergeAll(requestExecutorLayer, llmClientLayer)
+  const deps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
+  const llmClientLayer = LLMClient.layer.pipe(Layer.provide(deps))
+  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"}', '{"code":"slow_down"}'].map(
+      ['{"code":"resource_exhausted"}', '{"code":"service_unavailable"}'].map(
         (message) => classifyProviderFailure({ message })._tag,
       ),
-    ).toEqual(["ProviderInternal", "ProviderInternal", "ProviderInternal"])
+    ).toEqual(["ProviderInternal", "ProviderInternal"])
   })
 
   test("classifies transient client statuses as provider internal", () => {

packages/ai/test/provider-options/openai.types.ts

@@ -1,18 +1,13 @@
 import { LLM } from "../../src"
 import { OpenAI } from "../../src/providers"
 
-const selected = OpenAI.responses("gpt-5")
+const model = OpenAI.responses("gpt-5")
 
-LLM.request({ model: selected, prompt: "Hello", providerOptions: { openai: { reasoningEffort: "high" } } })
+LLM.request({ model, prompt: "Hello", providerOptions: { openai: { reasoningEffort: "high" } } })
 
 LLM.request({
-  model: selected,
+  model,
   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,6 +80,11 @@ 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,10 +1,9 @@
 import { describe, expect } from "bun:test"
-import { ConfigProvider, Effect, Layer, Ref, Stream } from "effect"
+import { ConfigProvider, Effect, Layer, Stream } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import {
   LLM,
   AIError,
-  HttpOptions,
   LLMEvent,
   LLMRequest,
   Message,
@@ -12,10 +11,9 @@ import {
   ToolCallPart,
   ToolDefinition,
   ToolResultPart,
-  TransportReason,
   Usage,
 } from "../../src"
-import { Auth, LLMClient, RequestExecutor, WebSocketTransport } from "../../src/route"
+import { Auth, LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route"
 import { compileRequest } from "../../src/route/client"
 import * as Azure from "../../src/providers/azure"
 import * as OpenAI from "../../src/providers/openai"
@@ -218,19 +216,19 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
-  it.effect("prepares one OpenAI Responses route for either transport", () =>
+  it.effect("prepares OpenAI Responses WebSocket target", () =>
     Effect.gen(function* () {
       const prepared = yield* compileRequest(
         LLMRequest.update(request, {
-          model: OpenAIResponses.route
+          model: OpenAIResponses.webSocketRoute
             .with({ endpoint: { baseURL: "https://api.openai.test/v1/" }, auth: Auth.bearer("test") })
             .model({ id: "gpt-4.1-mini" }),
         }),
       )
 
-      expect(prepared.route).toBe("openai-responses")
+      expect(prepared.route).toBe("openai-responses-websocket")
       expect(prepared.protocol).toBe("openai-responses")
-      expect(prepared.metadata).toEqual({ transport: "http-json" })
+      expect(prepared.metadata).toEqual({ transport: "websocket-json" })
       expect(prepared.body).toMatchObject({ model: "gpt-4.1-mini", store: false, stream: true })
     }),
   )
@@ -240,35 +238,41 @@ describe("OpenAI Responses route", () => {
       const sent: string[] = []
       const opened: Array<{ readonly url: string; readonly authorization: string | undefined }> = []
       let closed = false
-      const deps = Layer.succeed(
-        RequestExecutor.Service,
-        RequestExecutor.Service.of({
-          execute: () => Effect.die("unexpected HTTP request"),
-        }),
-      )
-      const webSocket = WebSocketTransport.makeDirect({
-        open: (input) =>
-          Effect.succeed({
-            sendText: (message) =>
-              Effect.sync(() => {
-                opened.push({ url: input.url, authorization: input.headers.authorization })
-                sent.push(message)
-              }),
-            messages: Stream.fromArray([
-              ProviderShared.encodeJson({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }),
-              ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_ws" } }),
-            ]),
-            close: Effect.sync(() => {
-              closed = true
-            }),
+      const deps = Layer.mergeAll(
+        Layer.succeed(
+          RequestExecutor.Service,
+          RequestExecutor.Service.of({
+            execute: () => Effect.die("unexpected HTTP request"),
           }),
-      })
+        ),
+        Layer.succeed(
+          WebSocketExecutor.Service,
+          WebSocketExecutor.Service.of({
+            open: (input) =>
+              Effect.succeed({
+                sendText: (message) =>
+                  Effect.sync(() => {
+                    opened.push({ url: input.url, authorization: input.headers.authorization })
+                    sent.push(message)
+                  }),
+                messages: Stream.fromArray([
+                  ProviderShared.encodeJson({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }),
+                  ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_ws" } }),
+                ]),
+                close: Effect.sync(() => {
+                  closed = true
+                }),
+              }),
+          }),
+        ),
+      )
       const response = yield* LLMClient.generate(
         LLM.request({
-          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("gpt-4.1-mini"),
+          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+            "gpt-4.1-mini",
+          ),
           prompt: "Say hello.",
         }),
-        { webSocket },
       ).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))))
 
       expect(response.text).toBe("Hi")
@@ -284,235 +288,15 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
-  it.effect("builds WebSocket and HTTP fallback from the same final request", () =>
-    Effect.gen(function* () {
-      const attempts = yield* Ref.make(0)
-      const message = yield* Ref.make("")
-      const body = yield* Ref.make("")
-      const response = yield* LLMClient.generate(
-        LLM.request({
-          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("gpt-4.1-mini"),
-          prompt: "Say hello.",
-          http: {
-            body: { model: "overlaid-model", metadata: { source: "overlay" } },
-            headers: { "x-request": "request" },
-            query: { mode: "test" },
-          },
-        }),
-        {
-          webSocket: {
-            execute: (exchange) =>
-              Effect.gen(function* () {
-                yield* exchange.driver
-                  .create(undefined)
-                  .pipe(Effect.flatMap((create) => Ref.set(message, create.message)))
-                return { frames: exchange.fallback(), complete: Effect.void }
-              }),
-          },
-        },
-      ).pipe(
-        Effect.provide(
-          dynamicResponse((input) =>
-            Effect.gen(function* () {
-              yield* Ref.update(attempts, (value) => value + 1)
-              yield* Ref.set(body, input.text)
-              expect(input.request.url).toBe("https://api.openai.test/v1/responses?mode=test")
-              expect(input.request.headers.authorization).toBe("Bearer test")
-              expect(input.request.headers["x-request"]).toBe("request")
-              return input.respond(sseEvents({ type: "response.completed", response: {} }), {
-                headers: { "content-type": "text/event-stream" },
-              })
-            }),
-          ),
-        ),
-      )
-
-      const httpBody = JSON.parse(yield* Ref.get(body))
-      const { stream: _stream, ...shared } = httpBody
-      expect(response.finishReason?.normalized).toBe("stop")
-      expect(yield* Ref.get(attempts)).toBe(1)
-      expect(JSON.parse(yield* Ref.get(message))).toEqual({ type: "response.create", ...shared })
-      expect(httpBody).toMatchObject({
-        model: "overlaid-model",
-        metadata: { source: "overlay" },
-        stream: true,
-      })
-    }),
-  )
-
-  it.effect("uses exactly one HTTP request when no WebSocket executor is supplied", () =>
-    Effect.gen(function* () {
-      const attempts = yield* Ref.make(0)
-      yield* LLMClient.generate(
-        LLMRequest.update(request, { http: new HttpOptions({ body: { input: "raw-http-input" } }) }),
-      ).pipe(
-        Effect.provide(
-          dynamicResponse((input) =>
-            Effect.gen(function* () {
-              yield* Ref.update(attempts, (value) => value + 1)
-              expect(JSON.parse(input.text).input).toBe("raw-http-input")
-              return input.respond(sseEvents({ type: "response.completed", response: {} }), {
-                headers: { "content-type": "text/event-stream" },
-              })
-            }),
-          ),
-        ),
-      )
-
-      expect(yield* Ref.get(attempts)).toBe(1)
-    }),
-  )
-
-  it.effect("closes a direct WebSocket execution after partial consumption", () =>
-    Effect.gen(function* () {
-      const closed = yield* Ref.make(false)
-      const webSocket = WebSocketTransport.makeDirect({
-        open: () =>
-          Effect.succeed({
-            sendText: () => Effect.void,
-            messages: Stream.fromArray([
-              ProviderShared.encodeJson({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }),
-              ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_ws" } }),
-            ]),
-            close: Ref.set(closed, true),
-          }),
-      })
-
-      yield* LLMClient.stream(
-        LLM.request({
-          model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("gpt-4.1-mini"),
-          prompt: "Say hello.",
-        }),
-        { webSocket },
-      ).pipe(
-        Stream.take(1),
-        Stream.runDrain,
-        Effect.provide(
-          LLMClient.layer.pipe(
-            Layer.provide(
-              Layer.succeed(
-                RequestExecutor.Service,
-                RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
-              ),
-            ),
-          ),
-        ),
-      )
-
-      expect(yield* Ref.get(closed)).toBe(true)
-    }),
-  )
-
-  it.effect("terminates WebSocket control events without waiting for the socket to close", () =>
-    Effect.gen(function* () {
-      const events = [
-        { type: "error", error: { code: "slow_down", message: "Try later" } },
-        {
-          type: "error",
-          status_code: 429,
-          message: "Rate limited",
-          headers: { "retry-after": 1, "x-request-id": "request", cached: false, invalid: [] },
-        },
-        {
-          type: "response.failed",
-          response: { error: { code: "server_error", message: "Unavailable" } },
-        },
-        { type: "error", status: "not-a-status", message: "Malformed status" },
-      ]
-
-      const errors = yield* Effect.forEach(events, (event) =>
-        LLMClient.generate(
-          LLM.request({
-            model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses(
-              "gpt-4.1-mini",
-            ),
-            prompt: "Say hello.",
-          }),
-          {
-            webSocket: WebSocketTransport.makeDirect({
-              open: () =>
-                Effect.succeed({
-                  sendText: () => Effect.void,
-                  messages: Stream.make(ProviderShared.encodeJson(event)).pipe(Stream.concat(Stream.never)),
-                  close: Effect.void,
-                }),
-            }),
-          },
-        ).pipe(
-          Effect.provide(
-            LLMClient.layer.pipe(
-              Layer.provide(
-                Layer.succeed(
-                  RequestExecutor.Service,
-                  RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
-                ),
-              ),
-            ),
-          ),
-          Effect.flip,
-        ),
-      )
-
-      expect(errors.map((error) => error.reason._tag)).toEqual([
-        "ProviderInternal",
-        "RateLimit",
-        "ProviderInternal",
-        "UnknownProvider",
-      ])
-    }),
-  )
-
-  it.effect("marks post-send WebSocket failures with delivery state", () =>
-    Effect.gen(function* () {
-      const failure = new AIError({
-        module: "test",
-        method: "receive",
-        reason: new TransportReason({ message: "socket closed", phase: "close" }),
-      })
-      const streams = [
-        Stream.fail(failure),
-        Stream.make(ProviderShared.encodeJson({ type: "response.created" })).pipe(Stream.concat(Stream.fail(failure))),
-      ]
-      const deps = Layer.succeed(
-        RequestExecutor.Service,
-        RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
-      )
-      const webSocket = WebSocketTransport.makeDirect({
-        open: () =>
-          Effect.succeed({
-            sendText: () => Effect.void,
-            messages: streams.shift() ?? Stream.die("unexpected WebSocket open"),
-            close: Effect.void,
-          }),
-      })
-      const model = OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses(
-        "gpt-4.1-mini",
-      )
-
-      const errors = yield* Effect.forEach(["first", "second"], (prompt) =>
-        LLMClient.generate(LLM.request({ model, prompt }), { webSocket }).pipe(
-          Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))),
-          Effect.flip,
-        ),
-      )
-
-      expect(errors.map((error) => error.reason)).toEqual([
-        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "ambiguous" }),
-        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "accepted" }),
-      ])
-    }),
-  )
-
   it.effect("fails immediately when WebSocket is already closed", () =>
     Effect.gen(function* () {
-      const error = yield* WebSocketTransport.fromWebSocket(
+      const error = yield* WebSocketExecutor.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,11 +2,12 @@ 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 } from "../src/route"
+import { LLMClient, RequestExecutor, WebSocketExecutor } 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,
@@ -16,7 +17,7 @@ import {
 const __dirname = path.dirname(fileURLToPath(import.meta.url))
 const FIXTURES_DIR = path.resolve(__dirname, "fixtures", "recordings")
 
-type RecordedEnv = RequestExecutorService | LLMClientService | ImageClientService
+type RecordedEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService | ImageClientService
 
 type RecordedTestsOptions = RecordedGroupOptions & {
   readonly options?: HttpRecorder.RecorderOptions
@@ -81,10 +82,11 @@ export const recordedTests = (options: RecordedTestsOptions) =>
           }),
         ),
       )
+      const deps = Layer.mergeAll(requestExecutor, WebSocketExecutor.layer)
       return Layer.mergeAll(
-        requestExecutor,
-        LLMClient.layer.pipe(Layer.provide(requestExecutor)),
-        ImageClient.layer.pipe(Layer.provide(requestExecutor)),
+        deps,
+        LLMClient.layer.pipe(Layer.provide(deps)),
+        ImageClient.layer.pipe(Layer.provide(deps)),
       )
     },
   })

packages/ai/test/schema.test.ts

@@ -11,7 +11,6 @@ import {
   LanguageModel,
   ModelID,
   ProviderID,
-  TransportReason,
   Usage,
 } from "../src/schema"
 import { ProviderShared } from "../src/protocols/shared"
@@ -109,21 +108,3 @@ 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/cli/script/entitlements.plist

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+    <key>com.apple.security.cs.allow-jit</key>
+    <true/>
+    <key>com.apple.security.cs.allow-unsigned-executable-memory</key>
+    <true/>
+    <key>com.apple.security.cs.disable-executable-page-protection</key>
+    <true/>
+    <key>com.apple.security.cs.allow-dyld-environment-variables</key>
+    <true/>
+    <key>com.apple.security.cs.disable-library-validation</key>
+    <true/>
+</dict>
+</plist>

packages/client/src/effect/api/api.ts

@@ -420,7 +420,6 @@ export type Endpoint5_26Output =
           readonly data: {
             readonly sessionID: Session.ID
             readonly delta: { readonly [x: string]: (string & Brand.Brand<"Instruction.Hash">) | "removed" }
-            readonly text?: string | undefined
           }
         }
       | {

packages/client/src/promise/generated/types.ts

@@ -676,7 +676,7 @@ export type SessionInstructionsUpdated = {
   type: "session.instructions.updated"
   durable: { aggregateID: string; seq: number; version: 2 }
   location?: LocationRef
-  data: { sessionID: string; delta: { [x: string]: string | "removed" }; text?: string }
+  data: { sessionID: string; delta: { [x: string]: string | "removed" } }
 }
 
 export type SessionSynthetic = {

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),
-      execute: () => Effect.succeed({ frames: Stream.empty }),
+      frames: () => Stream.empty,
     },
     defaults: {
       headers: info.headers,

packages/core/src/file-mutation.ts

@@ -1,7 +1,8 @@
 export * as FileMutation from "./file-mutation"
 
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
-import { Context, Effect, Layer } from "effect"
+import { Context, Effect, Layer, Schema } from "effect"
+import { dirname } from "path"
 import { KeyedMutex } from "./effect/keyed-mutex"
 import { FSUtil } from "@opencode-ai/util/fs-util"
 import { Bom } from "@opencode-ai/util/bom"
@@ -21,6 +22,22 @@ export interface TextWriteInput {
   readonly content: string
 }
 
+export interface ConditionalWriteInput extends WriteInput {
+  readonly expected: Uint8Array
+}
+
+export interface RemoveInput {
+  readonly target: Target
+}
+
+export class StaleContentError extends Schema.TaggedErrorClass<StaleContentError>()("FileMutation.StaleContentError", {
+  path: Schema.String,
+}) {}
+
+export class TargetExistsError extends Schema.TaggedErrorClass<TargetExistsError>()("FileMutation.TargetExistsError", {
+  path: Schema.String,
+}) {}
+
 export interface WriteResult {
   readonly operation: "write"
   readonly target: string
@@ -28,10 +45,24 @@ export interface WriteResult {
   readonly existed: boolean
 }
 
+export interface RemoveResult {
+  readonly operation: "remove"
+  readonly target: string
+  readonly resource: string
+  readonly existed: boolean
+}
+
 export interface Interface {
+  /** Create without replacing an existing target. */
+  readonly create: (input: WriteInput) => Effect.Effect<WriteResult, TargetExistsError | FSUtil.Error>
   readonly write: (input: WriteInput) => Effect.Effect<WriteResult, FSUtil.Error>
   /** Write text while retaining an existing UTF-8 BOM and emitting at most one BOM. */
   readonly writeTextPreservingBom: (input: TextWriteInput) => Effect.Effect<WriteResult, FSUtil.Error>
+  /** Commit only if an existing target still has the expected bytes. */
+  readonly writeIfUnchanged: (
+    input: ConditionalWriteInput,
+  ) => Effect.Effect<WriteResult, StaleContentError | FSUtil.Error>
+  readonly remove: (input: RemoveInput) => Effect.Effect<RemoveResult, FSUtil.Error>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/FileMutation") {}
@@ -58,6 +89,13 @@ const layer = Layer.effect(
       existed,
     })
 
+    const removeResult = (target: Target, existed: boolean): RemoveResult => ({
+      operation: "remove",
+      target: target.canonical,
+      resource: target.resource,
+      existed,
+    })
+
     const write = Effect.fn("FileMutation.write")((input: WriteInput) =>
       withTargetLock(input.target)(
         Effect.gen(function* () {
@@ -84,10 +122,62 @@ const layer = Layer.effect(
       ),
     )
 
-    return Service.of({ write, writeTextPreservingBom })
+    const create = Effect.fn("FileMutation.create")((input: WriteInput) =>
+      withTargetLock(input.target)(
+        Effect.gen(function* () {
+          const write =
+            typeof input.content === "string"
+              ? fs.writeFileString(input.target.canonical, input.content, { flag: "wx" })
+              : fs.writeFile(input.target.canonical, input.content, { flag: "wx" })
+          yield* write.pipe(
+            Effect.catchReason("PlatformError", "NotFound", () =>
+              fs.ensureDir(dirname(input.target.canonical)).pipe(Effect.andThen(write)),
+            ),
+            Effect.catchReason("PlatformError", "AlreadyExists", () =>
+              Effect.fail(new TargetExistsError({ path: input.target.canonical })),
+            ),
+          )
+          return writeResult(input.target, false)
+        }),
+      ),
+    )
+
+    const writeIfUnchanged = Effect.fn("FileMutation.writeIfUnchanged")((input: ConditionalWriteInput) =>
+      withTargetLock(input.target)(
+        Effect.gen(function* () {
+          const current = yield* fs.readFile(input.target.canonical)
+          if (!sameBytes(current, input.expected)) {
+            return yield* new StaleContentError({ path: input.target.canonical })
+          }
+          yield* typeof input.content === "string"
+            ? fs.writeFileString(input.target.canonical, input.content)
+            : fs.writeFile(input.target.canonical, input.content)
+          return writeResult(input.target, true)
+        }),
+      ),
+    )
+
+    const remove = Effect.fn("FileMutation.remove")((input: RemoveInput) =>
+      withTargetLock(input.target)(
+        Effect.gen(function* () {
+          const existed = yield* fs.remove(input.target.canonical).pipe(
+            Effect.as(true),
+            Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(false)),
+          )
+          return removeResult(input.target, existed)
+        }),
+      ),
+    )
+
+    return Service.of({ create, write, writeTextPreservingBom, writeIfUnchanged, remove })
   }),
 )
 
+function sameBytes(left: Uint8Array, right: Uint8Array) {
+  if (left.length !== right.length) return false
+  return left.every((byte, index) => byte === right[index])
+}
+
 export const node = makeLocationNode({ service: Service, layer, deps: [FSUtil.node] })
 
 /**

packages/core/src/filesystem/search.ts

@@ -31,7 +31,10 @@ export const ripgrepLayer = Layer.effect(
     const location = yield* Location.Service
     const ripgrep = yield* Ripgrep.Service
     const scope = yield* Scope.Scope
-    const files: string[] = []
+    const state = {
+      files: [] as string[],
+      directories: [] as string[],
+    }
     const directories = new Set<string>()
     yield* ripgrep
       .find({
@@ -40,9 +43,10 @@ export const ripgrepLayer = Layer.effect(
         limit: location.vcs ? Number.MAX_SAFE_INTEGER : 100_000,
         onEntry: (entry) =>
           Effect.sync(() => {
-            files.push(entry.path)
+            state.files.push(entry.path)
             const parts = entry.path.split("/")
             parts.slice(0, -1).forEach((_, index) => directories.add(parts.slice(0, index + 1).join("/") + path.sep))
+            state.directories = Array.from(directories)
           }),
       })
       .pipe(Effect.orDie, Effect.asVoid, Effect.forkIn(scope))
@@ -102,10 +106,10 @@ export const ripgrepLayer = Layer.effect(
         Effect.gen(function* () {
           const items =
             input.type === "file"
-              ? files
+              ? state.files
               : input.type === "directory"
-                ? Array.from(directories)
-                : [...files, ...directories]
+                ? state.directories
+                : [...state.files, ...state.directories]
           return fuzzysort.go(input.query, items, { limit: input.limit ?? 50 }).map((item) => {
             const relative = item.target
             const type = relative.endsWith(path.sep) ? ("directory" as const) : ("file" as const)

packages/core/src/session.ts

@@ -410,7 +410,7 @@ const layer = Layer.effect(
       fork: Effect.fn("Session.fork")(function* (input) {
         const parent = yield* result.get(input.sessionID)
         const boundary = yield* db
-          .select({ id: SessionMessageTable.id })
+          .select({ id: SessionMessageTable.id, seq: SessionMessageTable.seq })
           .from(SessionMessageTable)
           .where(
             and(
@@ -429,14 +429,13 @@ const layer = Layer.effect(
           })
         if (!boundary) return yield* new ForkEmptyError({ sessionID: input.sessionID })
         const sessionID = SessionSchema.ID.create()
-        // The fork adopts the parent's newest instruction values rather than the
-        // values in effect at the boundary; copied history may contain frozen
-        // instruction-update text the initial baseline already reflects.
+        const instructionThrough =
+          input.boundary.type === "before" ? boundary.seq - 1 : yield* Bus.latestSequence(db, parent.id)
         yield* bus.publish(SessionEvent.Forked, {
           sessionID,
           parentID: parent.id,
           boundary: { ...input.boundary, messageID: boundary.id },
-          instructions: yield* InstructionState.current(db, parent.id),
+          instructions: yield* InstructionState.valuesAt(db, parent.id, instructionThrough),
         })
         return yield* result.get(sessionID).pipe(Effect.orDie)
       }),

packages/core/src/session/history.ts

@@ -80,9 +80,10 @@ export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(fun
     .transaction(() =>
       Effect.gen(function* () {
         const messages = yield* messageEntries(db, sessionID)
+        const assembled = yield* InstructionState.assemble(db, sessionID, instructions)
         return {
-          initial: yield* InstructionState.initial(db, sessionID, instructions),
-          entries: messages,
+          initial: assembled.initial,
+          entries: [...messages, ...assembled.updates].toSorted((a, b) => a.seq - b.seq),
         }
       }),
     )
@@ -105,9 +106,10 @@ export const preview = Effect.fn("SessionHistory.preview")(function* (
         )
         const settled = unsettled === -1 ? messages : messages.slice(0, unsettled)
         const assembled = yield* InstructionState.preview(db, sessionID, instructions, observed)
+        const entries = [...settled, ...assembled.updates].toSorted((a, b) => a.seq - b.seq)
         return {
           initial: assembled.initial,
-          messages: settled.map((entry) => entry.message),
+          messages: entries.map((entry) => entry.message),
           instructionUpdate: assembled.update,
         }
       }),

packages/core/src/session/instruction-state.ts

@@ -1,20 +1,25 @@
 export * as InstructionState from "./instruction-state"
 
-import { eq, inArray, sql } from "drizzle-orm"
-import { Effect, Option, Schema } from "effect"
+import { and, asc, desc, eq, gt, inArray, lte, sql } from "drizzle-orm"
+import { DateTime, Effect, Option, Schema } from "effect"
 import type { Database } from "../database/database"
-import type { Bus } from "../bus"
+import { Bus } from "../bus"
+import { EventTable } from "../event/sql"
 import { Instructions } from "../instructions/index"
 import { SessionEvent } from "./event"
+import { SessionMessage } from "./message"
+import { Event } from "@opencode-ai/schema/event"
 import { SessionSchema } from "./schema"
 import { InstructionBlobTable, InstructionStateTable } from "./sql"
 
 type DatabaseService = Database.Interface["db"]
 
+const decodeInstructionsUpdated = Schema.decodeUnknownSync(SessionEvent.InstructionsUpdated.data)
+const decodeForked = Schema.decodeUnknownSync(SessionEvent.Forked.data)
+
 export interface Observation extends Instructions.Admission {
   readonly sessionID: SessionSchema.ID
   readonly initial: boolean
-  readonly previous: Instructions.Values
   readonly current: Instructions.Values
 }
 
@@ -23,14 +28,13 @@ export const observe = Effect.fn("InstructionState.observe")(function* (
   instructions: Instructions.Instructions,
   sessionID: SessionSchema.ID,
 ): Effect.fn.Return<Observation, Instructions.InitializationBlocked> {
-  const [observed, stored] = yield* Effect.all([Instructions.read(instructions), find(db, sessionID)], {
+  const [observed, stored] = yield* Effect.all([Instructions.read(instructions), ensure(db, sessionID)], {
     concurrency: "unbounded",
   })
   const result = yield* observeAgainst(observed, stored?.current_values)
   return {
     sessionID,
     initial: !stored,
-    previous: stored?.current_values ?? {},
     ...result,
   }
 })
@@ -38,20 +42,12 @@ export const observe = Effect.fn("InstructionState.observe")(function* (
 export const commit = Effect.fn("InstructionState.commit")(function* (
   db: DatabaseService,
   bus: Bus.Interface,
-  instructions: Instructions.Instructions,
   observation: Observation,
 ) {
   if (!observation.initial && Object.keys(observation.delta).length === 0) return
-  // The rendered text is frozen into the durable event: replaying it later would
-  // require the Location-scoped registry that produced it.
-  const text = observation.initial ? "" : yield* renderUpdateText(db, instructions, observation)
   yield* bus.publish(
     SessionEvent.InstructionsUpdated,
-    {
-      sessionID: observation.sessionID,
-      delta: observation.delta,
-      ...(text.length > 0 ? { text } : {}),
-    },
+    { sessionID: observation.sessionID, delta: observation.delta },
     {
       // Initial sync establishes the baseline; unlike later deltas it is not chronological history.
       ...(observation.initial ? { metadata: { instructions: { initial: true } } } : {}),
@@ -60,27 +56,13 @@ export const commit = Effect.fn("InstructionState.commit")(function* (
   )
 })
 
-const renderUpdateText = Effect.fnUntraced(function* (
-  db: DatabaseService,
-  instructions: Instructions.Instructions,
-  observation: Observation,
-) {
-  const replaced = Object.entries(observation.previous).filter(([key]) => Object.hasOwn(observation.delta, key))
-  const blobs = yield* loadBlobs(db, replaced.map(([, hash]) => hash))
-  const previous = Object.fromEntries(replaced.map(([key, hash]) => [key, requireBlob(blobs, hash)]))
-  const admitted = new Map(
-    Object.entries(observation.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]),
-  )
-  return Instructions.renderUpdate(instructions, previous, dereferenceDelta(observation.delta, admitted))
-})
-
 export const prepare = Effect.fn("InstructionState.prepare")(function* (
   db: DatabaseService,
   bus: Bus.Interface,
   instructions: Instructions.Instructions,
   sessionID: SessionSchema.ID,
 ) {
-  yield* commit(db, bus, instructions, yield* observe(db, instructions, sessionID))
+  yield* commit(db, bus, yield* observe(db, instructions, sessionID))
 })
 
 export const apply = Effect.fn("InstructionState.apply")(function* (
@@ -158,24 +140,79 @@ export const reset = Effect.fn("InstructionState.reset")(function* (db: Database
     .pipe(Effect.orDie)
 })
 
-/** Renders the epoch baseline shown at the start of every model request. */
-export const initial = Effect.fn("InstructionState.initial")(function* (
+export const rebuild = Effect.fn("InstructionState.rebuild")(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+) {
+  const state = yield* stateFromEvents(db, sessionID)
+  if (!state) {
+    yield* reset(db, sessionID)
+    return undefined
+  }
+  yield* db
+    .insert(InstructionStateTable)
+    .values(state)
+    .onConflictDoUpdate({
+      target: InstructionStateTable.session_id,
+      set: {
+        epoch_start: state.epoch_start,
+        through_seq: state.through_seq,
+        initial_values: state.initial_values,
+        current_values: state.current_values,
+      },
+    })
+    .run()
+    .pipe(Effect.orDie)
+  return state
+})
+
+const assembleState = Effect.fnUntraced(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  instructions: Instructions.Instructions,
+  state: typeof InstructionStateTable.$inferSelect,
+) {
+  const rows = yield* instructionUpdatesAfter(db, sessionID, state.epoch_start)
+  const updates = rows.map((row) => ({
+    row,
+    delta: decodeInstructionsUpdated(row.data).delta,
+  }))
+  const blobs = yield* loadBlobs(db, [
+    ...Object.values(state.initial_values),
+    ...updates.flatMap((update) =>
+      Object.values(update.delta).filter((hash): hash is Instructions.Hash => hash !== "removed"),
+    ),
+  ])
+  const valuesAtStart = dereference(state.initial_values, blobs)
+  let values = valuesAtStart
+  const result: Array<{ readonly seq: number; readonly message: SessionMessage.System }> = []
+  for (const update of updates) {
+    const delta = dereferenceDelta(update.delta, blobs)
+    const text = Instructions.renderUpdate(instructions, values, delta)
+    if (text.length > 0)
+      result.push({
+        seq: update.row.seq,
+        message: SessionMessage.System.make({
+          id: SessionMessage.ID.fromEvent(Event.ID.make(update.row.id)),
+          type: "system",
+          text,
+          time: { created: DateTime.makeUnsafe(update.row.created) },
+        }),
+      })
+    values = Instructions.applyDelta(values, delta)
+  }
+  return { initial: Instructions.renderInitial(instructions, valuesAtStart), updates: result, current: values }
+})
+
+export const assemble = Effect.fn("InstructionState.assemble")(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
   instructions: Instructions.Instructions,
 ) {
   const state = yield* find(db, sessionID)
   if (!state) return yield* Effect.die(new Error(`Instruction state not found during assembly: ${sessionID}`))
-  const blobs = yield* loadBlobs(db, Object.values(state.initial_values))
-  return Instructions.renderInitial(instructions, dereference(state.initial_values, blobs))
-})
-
-/** The current instruction values, used to seed a fork's baseline. */
-export const current = Effect.fn("InstructionState.current")(function* (
-  db: DatabaseService,
-  sessionID: SessionSchema.ID,
-) {
-  return (yield* find(db, sessionID))?.current_values
+  const assembled = yield* assembleState(db, sessionID, instructions, state)
+  return { initial: assembled.initial, updates: assembled.updates }
 })
 
 export const preview = Effect.fn("InstructionState.preview")(function* (
@@ -184,26 +221,20 @@ export const preview = Effect.fn("InstructionState.preview")(function* (
   instructions: Instructions.Instructions,
   observed: Instructions.ReadResult,
 ) {
-  const state = yield* find(db, sessionID)
+  const state = yield* readState(db, sessionID)
   const result = yield* observeAgainst(observed, state?.current_values)
-  const observedBlobs = new Map<Instructions.Hash, Schema.Json>(
+  const blobs = new Map<Instructions.Hash, Schema.Json>(
     Object.entries(result.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]),
   )
   if (!state) {
-    const values = dereference(result.current, observedBlobs)
-    return { initial: Instructions.renderInitial(instructions, values), update: "" }
+    const values = dereference(result.current, blobs)
+    return { initial: Instructions.renderInitial(instructions, values), updates: [], update: "" }
   }
-  const stored = yield* loadBlobs(db, [
-    ...Object.values(state.initial_values),
-    ...Object.values(state.current_values),
-  ])
+  const assembled = yield* assembleState(db, sessionID, instructions, state)
   return {
-    initial: Instructions.renderInitial(instructions, dereference(state.initial_values, stored)),
-    update: Instructions.renderUpdate(
-      instructions,
-      dereference(state.current_values, stored),
-      dereferenceDelta(result.delta, new Map([...stored, ...observedBlobs])),
-    ),
+    initial: assembled.initial,
+    updates: assembled.updates,
+    update: Instructions.renderUpdate(instructions, assembled.current, dereferenceDelta(result.delta, blobs)),
   }
 })
 
@@ -224,6 +255,46 @@ const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: Sessio
     .pipe(Effect.orDie)
 })
 
+const ensure = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
+  const stored = yield* find(db, sessionID)
+  if (!stored) return yield* rebuild(db, sessionID)
+  const latest = yield* latestRelevantSequence(db, sessionID)
+  if (!latest || latest.seq <= stored.through_seq) return stored
+  return yield* rebuild(db, sessionID)
+})
+
+const readState = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
+  const stored = yield* find(db, sessionID)
+  if (!stored) return yield* stateFromEvents(db, sessionID)
+  const latest = yield* latestRelevantSequence(db, sessionID)
+  if (!latest || latest.seq <= stored.through_seq) return stored
+  return yield* stateFromEvents(db, sessionID)
+})
+
+const stateFromEvents = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
+  const folded = fold(yield* instructionEvents(db, sessionID))
+  return folded ? foldedState(sessionID, folded) : undefined
+})
+
+export const valuesAt = Effect.fn("InstructionState.valuesAt")(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  through: number,
+) {
+  return fold(yield* instructionEvents(db, sessionID, through))?.current
+})
+
+const latestRelevantSequence = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
+  return yield* db
+    .select({ seq: EventTable.seq })
+    .from(EventTable)
+    .where(and(eq(EventTable.aggregate_id, sessionID), inArray(EventTable.type, relevantEventTypes)))
+    .orderBy(desc(EventTable.seq))
+    .limit(1)
+    .get()
+    .pipe(Effect.orDie)
+})
+
 const insertBlobs = Effect.fnUntraced(function* (db: DatabaseService, blobs: Readonly<Record<string, Schema.Json>>) {
   const rows = Object.entries(blobs).map(([hash, value]) => ({ hash: Instructions.Hash.make(hash), value }))
   if (rows.length === 0) return
@@ -268,3 +339,106 @@ function requireBlob(blobs: ReadonlyMap<Instructions.Hash, Schema.Json>, hash: I
   if (value === undefined) throw new Error(`Instruction blob not found: ${hash}`)
   return value
 }
+
+const instructionEventType = Bus.versionedType(
+  SessionEvent.InstructionsUpdated.type,
+  SessionEvent.InstructionsUpdated.durable.version,
+)
+const compactionEventType = Bus.versionedType(
+  SessionEvent.Compaction.Ended.type,
+  SessionEvent.Compaction.Ended.durable.version,
+)
+const movedEventType = Bus.versionedType(SessionEvent.Moved.type, SessionEvent.Moved.durable.version)
+const revertedEventType = Bus.versionedType(
+  SessionEvent.RevertEvent.Committed.type,
+  SessionEvent.RevertEvent.Committed.durable.version,
+)
+const forkedEventType = Bus.versionedType(SessionEvent.Forked.type, SessionEvent.Forked.durable.version)
+const relevantEventTypes = [
+  forkedEventType,
+  instructionEventType,
+  compactionEventType,
+  movedEventType,
+  revertedEventType,
+]
+
+type InstructionEventRow = typeof EventTable.$inferSelect
+
+const instructionEvents = Effect.fnUntraced(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  through?: number,
+): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
+  return yield* eventRows(db, sessionID, relevantEventTypes, undefined, through)
+})
+
+const instructionUpdatesAfter = Effect.fnUntraced(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  after: number,
+) {
+  return yield* eventRows(db, sessionID, [instructionEventType], after)
+})
+
+const eventRows = Effect.fnUntraced(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  types: ReadonlyArray<string>,
+  after?: number,
+  through?: number,
+): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
+  return yield* db
+    .select()
+    .from(EventTable)
+    .where(
+      and(
+        eq(EventTable.aggregate_id, sessionID),
+        inArray(EventTable.type, types),
+        after === undefined ? undefined : gt(EventTable.seq, after),
+        through === undefined ? undefined : lte(EventTable.seq, through),
+      ),
+    )
+    .orderBy(asc(EventTable.seq))
+    .all()
+    .pipe(Effect.orDie)
+})
+
+function fold(rows: ReadonlyArray<InstructionEventRow>) {
+  return rows.reduce<
+    | {
+        readonly epochStart: number
+        readonly throughSeq: number
+        readonly initial: Instructions.Values
+        readonly current: Instructions.Values
+      }
+    | undefined
+  >((state, row) => {
+    if (row.type === forkedEventType) {
+      const instructions = decodeForked(row.data).instructions
+      return instructions
+        ? { epochStart: row.seq, throughSeq: row.seq, initial: instructions, current: instructions }
+        : undefined
+    }
+    if (row.type === movedEventType || row.type === revertedEventType) return undefined
+    if (row.type === compactionEventType)
+      return state
+        ? { epochStart: row.seq, throughSeq: row.seq, initial: state.current, current: state.current }
+        : undefined
+    if (row.type !== instructionEventType) return state
+    const delta = decodeInstructionsUpdated(row.data).delta
+    const current = Instructions.applyHashDelta(state?.current ?? {}, delta)
+    return state
+      ? { ...state, throughSeq: row.seq, current }
+      : { epochStart: row.seq, throughSeq: row.seq, initial: current, current }
+  }, undefined)
+}
+
+function foldedState(sessionID: SessionSchema.ID, folded: NonNullable<ReturnType<typeof fold>>) {
+  return {
+    session_id: sessionID,
+    epoch_start: folded.epochStart,
+    through_seq: folded.throughSeq,
+    initial_values: folded.initial,
+    current_values: folded.current,
+  }
+}

packages/core/src/session/message-updater.ts

@@ -179,18 +179,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       "session.execution.succeeded": () => clearCurrentRetry,
       "session.execution.failed": () => clearCurrentRetry,
       "session.execution.interrupted": () => clearCurrentRetry,
-      "session.instructions.updated": (event) => {
-        if (event.data.text === undefined) return Effect.void
-        return adapter.appendMessage(
-          SessionMessage.System.make({
-            id: SessionMessage.ID.fromEvent(event.id),
-            type: "system",
-            text: event.data.text,
-            metadata: event.metadata,
-            time: { created: event.created },
-          }),
-        )
-      },
+      "session.instructions.updated": () => Effect.void,
       "session.synthetic": (event) => {
         return adapter.appendMessage(
           SessionMessage.Synthetic.make({

packages/core/src/session/pending.ts

@@ -12,8 +12,10 @@ import {
   User,
   UserData,
 } from "@opencode-ai/schema/session-pending"
+import { Event } from "@opencode-ai/schema/event"
 import type { Database } from "../database/database"
 import { Bus } from "../bus"
+import { EventTable } from "../event/sql"
 import { KeyedMutex } from "../effect/keyed-mutex"
 import { SessionEvent } from "./event"
 import { SessionMessage } from "./message"
@@ -35,7 +37,11 @@ const decodeUser = Schema.decodeUnknownSync(UserData)
 const encodeUser = Schema.encodeSync(UserData)
 const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData)
 const encodeSynthetic = Schema.encodeSync(SyntheticData)
-const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
+const decodeAdmittedEvent = Schema.decodeUnknownOption(SessionEvent.InputAdmitted.data)
+const admittedEventType = Bus.versionedType(
+  SessionEvent.InputAdmitted.type,
+  SessionEvent.InputAdmitted.durable.version,
+)
 const inboxLocks = KeyedMutex.makeUnsafe<SessionSchema.ID>()
 
 export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()(
@@ -97,35 +103,46 @@ export const compaction = Effect.fn("SessionPending.compaction")(function* (
   return entry.type === "compaction" ? entry : undefined
 })
 
-const promotedFromMessage = Effect.fn("SessionPending.promotedFromMessage")(function* (
+/**
+ * Reconstruct the admitted record for a pending row that was already consumed
+ * by promotion. The projected `session_message` row proves promotion happened;
+ * the durable `session.input.admitted` event retains the exact admitted
+ * message, including delivery.
+ */
+const promotedFromHistory = Effect.fn("SessionPending.promotedFromHistory")(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
   id: SessionMessage.ID,
-  delivery: Delivery,
 ) {
-  const row = yield* db
+  const message = yield* db
     .select()
     .from(SessionMessageTable)
     .where(eq(SessionMessageTable.id, id))
     .get()
     .pipe(Effect.orDie)
-  if (row === undefined) return undefined
-  if (row.session_id !== sessionID || (row.type !== "user" && row.type !== "synthetic"))
+  if (message === undefined) return undefined
+  if (message.session_id !== sessionID || (message.type !== "user" && message.type !== "synthetic"))
     return yield* Effect.die(new LifecycleConflict({ id }))
-  const message = decodeMessage({ ...row.data, id: row.id, type: row.type })
-  const base = { id, sessionID, timeCreated: message.time.created, delivery }
-  if (message.type === "user")
-    return User.make({
-      ...base,
-      type: "user",
-      data: decodeUser(message),
-    })
-  if (message.type === "synthetic")
-    return Synthetic.make({
-      ...base,
-      type: "synthetic",
-      data: decodeSynthetic(message),
-    })
+  const rows = yield* db
+    .select()
+    .from(EventTable)
+    .where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, admittedEventType)))
+    .all()
+    .pipe(Effect.orDie)
+  for (const row of rows) {
+    const decoded = decodeAdmittedEvent(row.data)
+    if (decoded._tag !== "Some" || decoded.value.inputID !== id) continue
+    const base = {
+      id,
+      sessionID,
+      timeCreated: DateTime.makeUnsafe(row.created),
+    }
+    return decoded.value.input.type === "user"
+      ? User.make({ ...base, ...decoded.value.input })
+      : Synthetic.make({ ...base, ...decoded.value.input })
+  }
+  // A projected message without an admitted event in this aggregate (for
+  // example fork-copied history) is not a retryable admission.
   return yield* Effect.die(new LifecycleConflict({ id }))
 })
 
@@ -143,7 +160,7 @@ export const admit = Effect.fn("SessionPending.admit")(function* (
     if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id }))
     return existing
   }
-  const promoted = yield* promotedFromMessage(db, request.sessionID, request.id, request.input.delivery)
+  const promoted = yield* promotedFromHistory(db, request.sessionID, request.id)
   if (promoted !== undefined) return promoted
   return yield* bus
     .publish(SessionEvent.InputAdmitted, {
@@ -409,7 +426,7 @@ const publish = Effect.fn("SessionPending.publish")(function* (
         .pipe(
           Effect.catchDefect((defect) =>
             defect instanceof LifecycleConflict
-              ? promotedFromMessage(db, sessionID, entry.id, entry.delivery).pipe(
+              ? promotedFromHistory(db, sessionID, entry.id).pipe(
                   Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
                 )
               : Effect.die(defect),

packages/core/src/session/projector.ts

@@ -1,6 +1,6 @@
 export * as SessionProjector from "./projector"
 
-import { and, asc, desc, eq, gt, gte, lt, lte, sql } from "drizzle-orm"
+import { and, asc, desc, eq, gt, gte, inArray, lt, lte, sql } from "drizzle-orm"
 import { DateTime, Effect, Layer, Schema, Stream } from "effect"
 import { Database } from "../database/database"
 import { Bus } from "../bus"
@@ -21,7 +21,10 @@ import { Money } from "@opencode-ai/schema/money"
 
 type DatabaseService = Database.Interface["db"]
 type CurrentDurableEvent = Extract<SessionEvent.Event, { readonly durable: object }>
-type MessageEvent = Exclude<CurrentDurableEvent, typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type>
+type MessageEvent = Exclude<
+  CurrentDurableEvent,
+  typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type | typeof SessionEvent.InstructionsUpdated.Type
+>
 
 const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
 const encodeMessage = Schema.encodeSync(SessionMessage.Info)
@@ -252,22 +255,66 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
       .pipe(Effect.orDie)
     if (rows.length === 0) break
 
+    const idMap = new Map(rows.map((row) => [row.id, SessionMessage.ID.create()]))
     yield* db
       .insert(SessionMessageTable)
       .values(
-        rows.map((row) => ({
-          id: SessionMessage.ID.create(),
-          session_id: event.data.sessionID,
-          type: row.type,
-          seq: row.seq,
-          time_created: row.time_created,
-          time_updated: row.time_updated,
-          data: row.data,
-        })),
+        rows.map((row) => {
+          const id = idMap.get(row.id)
+          if (!id) throw new Error(`Fork message ID mapping missing: ${row.id}`)
+          return {
+            id,
+            session_id: event.data.sessionID,
+            type: row.type,
+            seq: row.seq,
+            time_created: row.time_created,
+            time_updated: row.time_updated,
+            data: row.data,
+          }
+        }),
       )
       .run()
       .pipe(Effect.orDie)
 
+    const pendingRows = yield* db
+      .select()
+      .from(SessionPendingTable)
+      .where(
+        and(
+          eq(SessionPendingTable.session_id, event.data.parentID),
+          inArray(
+            SessionPendingTable.id,
+            rows.map((row) => row.id),
+          ),
+        ),
+      )
+      .all()
+      .pipe(Effect.orDie)
+    if (pendingRows.length > 0) {
+      yield* db
+        .insert(SessionPendingTable)
+        .values(
+          pendingRows.flatMap((row) => {
+            const id = idMap.get(row.id)
+            return id && row.type !== "compaction"
+              ? [
+                  {
+                    id,
+                    session_id: event.data.sessionID,
+                    type: row.type,
+                    data: row.data,
+                    delivery: row.delivery,
+                    admitted_seq: row.admitted_seq,
+                    time_created: row.time_created,
+                  },
+                ]
+              : []
+          }),
+        )
+        .run()
+        .pipe(Effect.orDie)
+    }
+
     cursor = rows.at(-1)!.seq
   }
   if (copiedSeq !== undefined) yield* Bus.reserveSequence(db, event.data.sessionID, copiedSeq)
@@ -635,10 +682,7 @@ const layer = Layer.effectDiscard(
     yield* bus.project(SessionEvent.Execution.Failed, (event) => run(db, event))
     yield* bus.project(SessionEvent.Execution.Interrupted, (event) => run(db, event))
     yield* bus.project(SessionEvent.InstructionsUpdated, (event) =>
-      Effect.gen(function* () {
-        yield* run(db, event)
-        yield* InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta)
-      }),
+      InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta),
     )
     yield* bus.project(SessionEvent.Synthetic, (event) => run(db, event))
     yield* bus.project(SessionEvent.Skill.Activated, (event) => run(db, event))

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

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

packages/core/test/file-mutation.test.ts

@@ -89,6 +89,68 @@ describe("FileMutation", () => {
     ),
   )
 
+  it.live("rejects create when a prospective target appears after resolution", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const targetPath = path.join(directory, "appeared.txt")
+        const target = yield* (yield* LocationMutation.Service).resolve({ path: "appeared.txt" })
+        yield* Effect.promise(() => fs.writeFile(targetPath, "winner"))
+
+        expect(
+          yield* (yield* FileMutation.Service).create({ target, content: "replacement" }).pipe(Effect.flip),
+        ).toMatchObject({
+          _tag: "FileMutation.TargetExistsError",
+        })
+        expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("winner")
+      }).pipe(provide(directory)),
+    ),
+  )
+
+  it.live("creates when an existing target disappears after resolution", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const targetPath = path.join(directory, "removed.txt")
+        yield* Effect.promise(() => fs.writeFile(targetPath, "before"))
+        const target = yield* (yield* LocationMutation.Service).resolve({ path: "removed.txt" })
+        yield* Effect.promise(() => fs.rm(targetPath))
+
+        expect(yield* (yield* FileMutation.Service).create({ target, content: "after" })).toEqual({
+          operation: "write",
+          target: target.canonical,
+          resource: "removed.txt",
+          existed: false,
+        })
+        expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("after")
+      }).pipe(provide(directory)),
+    ),
+  )
+
+  it.live("removes an existing internal file", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const targetPath = path.join(directory, "remove.txt")
+        yield* Effect.promise(() => fs.writeFile(targetPath, "remove"))
+        const target = yield* (yield* LocationMutation.Service).resolve({ path: "remove.txt" })
+        const result = yield* (yield* FileMutation.Service).remove({ target })
+
+        expect(result).toEqual({
+          operation: "remove",
+          target: target.canonical,
+          resource: "remove.txt",
+          existed: true,
+        })
+        expect(
+          yield* Effect.promise(() =>
+            fs.stat(targetPath).then(
+              () => true,
+              () => false,
+            ),
+          ),
+        ).toBe(false)
+      }).pipe(provide(directory)),
+    ),
+  )
+
   it.live("writes an explicitly resolved external target", () =>
     withTmp((directory) =>
       withTmp((outside) =>
@@ -109,6 +171,49 @@ describe("FileMutation", () => {
     ),
   )
 
+  it.live("removes an explicitly resolved external target", () =>
+    withTmp((directory) =>
+      withTmp((outside) =>
+        Effect.gen(function* () {
+          const targetPath = path.join(outside, "external.txt")
+          yield* Effect.promise(() => fs.writeFile(targetPath, "external"))
+          const target = yield* (yield* LocationMutation.Service).resolve({ path: targetPath })
+          const result = yield* (yield* FileMutation.Service).remove({ target })
+
+          expect(result).toEqual({
+            operation: "remove",
+            target: target.canonical,
+            resource: target.resource,
+            existed: true,
+          })
+          expect(
+            yield* Effect.promise(() =>
+              fs.stat(targetPath).then(
+                () => true,
+                () => false,
+              ),
+            ),
+          ).toBe(false)
+        }).pipe(provide(directory)),
+      ),
+    ),
+  )
+
+  it.live("reports a missing target as not removed without checking existence first", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const target = yield* (yield* LocationMutation.Service).resolve({ path: "missing.txt" })
+
+        expect(yield* (yield* FileMutation.Service).remove({ target })).toEqual({
+          operation: "remove",
+          target: target.canonical,
+          resource: "missing.txt",
+          existed: false,
+        })
+      }).pipe(provide(directory)),
+    ),
+  )
+
   it.live("serializes concurrent writes to the same canonical target", () =>
     withTmp((directory) =>
       Effect.gen(function* () {
@@ -152,6 +257,63 @@ describe("FileMutation", () => {
     ),
   )
 
+  it.live("allows only one concurrent conditional write based on the same bytes", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const targetPath = path.join(directory, "shared.txt")
+        yield* Effect.promise(() => fs.writeFile(targetPath, "initial"))
+        const firstStarted = yield* Deferred.make<void>()
+        const releaseFirst = yield* Deferred.make<void>()
+        let writes = 0
+        const filesystem = instrumentWrites((write) =>
+          Effect.gen(function* () {
+            writes++
+            if (writes === 1) {
+              yield* Deferred.succeed(firstStarted, undefined)
+              yield* Deferred.await(releaseFirst)
+            }
+            yield* write
+          }),
+        )
+
+        yield* Effect.gen(function* () {
+          const mutation = yield* LocationMutation.Service
+          const files = yield* FileMutation.Service
+          const target = yield* mutation.resolve({ path: "shared.txt" })
+          const expected = new TextEncoder().encode("initial")
+          const first = yield* files.writeIfUnchanged({ target, expected, content: "first" }).pipe(Effect.forkChild)
+          yield* Deferred.await(firstStarted)
+          const second = yield* files
+            .writeIfUnchanged({ target, expected, content: "second" })
+            .pipe(Effect.flip, Effect.forkChild)
+
+          yield* Deferred.succeed(releaseFirst, undefined)
+          yield* Fiber.join(first)
+          expect(yield* Fiber.join(second)).toMatchObject({ _tag: "FileMutation.StaleContentError" })
+          expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("first")
+          expect(writes).toBe(1)
+        }).pipe(provide(directory, filesystem))
+      }),
+    ),
+  )
+
+  it.live("rejects a conditional write when target content is already stale", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const targetPath = path.join(directory, "stale.txt")
+        yield* Effect.promise(() => fs.writeFile(targetPath, "current"))
+        const target = yield* (yield* LocationMutation.Service).resolve({ path: "stale.txt" })
+
+        expect(
+          yield* (yield* FileMutation.Service)
+            .writeIfUnchanged({ target, expected: new TextEncoder().encode("older"), content: "replacement" })
+            .pipe(Effect.flip),
+        ).toMatchObject({ _tag: "FileMutation.StaleContentError", path: target.canonical })
+        expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("current")
+      }).pipe(provide(directory)),
+    ),
+  )
+
   it.live("allows distinct canonical targets to proceed independently", () =>
     withTmp((directory) =>
       Effect.gen(function* () {

packages/core/test/instruction-state.test.ts

@@ -14,7 +14,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
 import { InstructionState } from "@opencode-ai/core/session/instruction-state"
 import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { SessionSchema } from "@opencode-ai/core/session/schema"
-import { InstructionBlobTable, InstructionStateTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
+import { InstructionBlobTable, InstructionStateTable, SessionTable } from "@opencode-ai/core/session/sql"
 import { testEffect } from "./lib/effect"
 
 const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node])))
@@ -105,7 +105,6 @@ describe("InstructionState", () => {
       expect(observation).toEqual({
         sessionID,
         initial: true,
-        previous: {},
         current: {
           "test/first": Instructions.hash("first"),
           "test/second": Instructions.hash("second"),
@@ -157,7 +156,7 @@ describe("InstructionState", () => {
 
       const initial = yield* InstructionState.observe(db, instructions, sessionID)
       expect(reads).toBe(2)
-      yield* InstructionState.commit(db, events, instructions, initial)
+      yield* InstructionState.commit(db, events, initial)
       expect(reads).toBe(2)
 
       current = "changed"
@@ -167,10 +166,6 @@ describe("InstructionState", () => {
       expect(changed).toMatchObject({
         sessionID,
         initial: false,
-        previous: {
-          "test/current": Instructions.hash("initial"),
-          "test/retired": Instructions.hash("retired"),
-        },
         current: { "test/current": Instructions.hash("changed") },
         delta: {
           "test/current": Instructions.hash("changed"),
@@ -178,7 +173,7 @@ describe("InstructionState", () => {
         },
         blobs: { [Instructions.hash("changed")]: "changed" },
       })
-      yield* InstructionState.commit(db, events, instructions, changed)
+      yield* InstructionState.commit(db, events, changed)
       expect(reads).toBe(4)
       yield* unsubscribe
 
@@ -195,11 +190,6 @@ describe("InstructionState", () => {
           "test/retired": "removed",
         },
       ])
-      // The chronological update text is frozen into the event; the baseline has none.
-      expect((yield* instructionEvents(db, sessionID)).map((event) => event.data.text)).toEqual([
-        undefined,
-        "changed\n\nRemoved retired",
-      ])
       expect(yield* db.select().from(InstructionStateTable).get().pipe(Effect.orDie)).toMatchObject({
         initial_values: {
           "test/current": Instructions.hash("initial"),
@@ -232,19 +222,18 @@ describe("InstructionState", () => {
       expect(observation).toEqual({
         sessionID,
         initial: false,
-        previous: { "test/context": Instructions.hash("unchanged") },
         current: { "test/context": Instructions.hash("unchanged") },
         delta: {},
         blobs: {},
       })
-      yield* InstructionState.commit(db, events, instructions, observation)
+      yield* InstructionState.commit(db, events, observation)
 
       expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
       expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs)
     }),
   )
 
-  it.effect("treats a missing state row as a fresh baseline without repairing it", () =>
+  it.effect("assembles a fresh private update without repairing a missing cache", () =>
     Effect.gen(function* () {
       const sessionID = SessionSchema.ID.make("ses_instruction_generate")
       const { db, events } = yield* setup(sessionID)
@@ -265,7 +254,7 @@ describe("InstructionState", () => {
 
       const assembled = yield* preview(db, sessionID, instructions)
 
-      expect(assembled).toEqual({ initial: "Changed context", update: "" })
+      expect(assembled).toEqual({ initial: "Initial context", updates: [], update: "Changed context" })
       expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
       expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs)
       expect(
@@ -279,7 +268,7 @@ describe("InstructionState", () => {
     }),
   )
 
-  it.effect("trusts the projected state without consulting durable events", () =>
+  it.effect("reads through a stale cache without repairing it", () =>
     Effect.gen(function* () {
       const sessionID = SessionSchema.ID.make("ses_instruction_generate_stale")
       const { db, events } = yield* setup(sessionID)
@@ -291,7 +280,6 @@ describe("InstructionState", () => {
       yield* InstructionState.prepare(db, events, instructions, sessionID)
       value = "Committed update"
       yield* InstructionState.prepare(db, events, instructions, sessionID)
-      // Tamper with the projected state; the authoritative row wins over event history.
       yield* db
         .update(InstructionStateTable)
         .set({ through_seq: 0, current_values: { "test/context": Instructions.hash("Initial context") } })
@@ -306,6 +294,7 @@ describe("InstructionState", () => {
       const assembled = yield* preview(db, sessionID, instructions)
 
       expect(assembled.initial).toBe("Initial context")
+      expect(assembled.updates.map((entry) => entry.message.text)).toEqual(["Committed update"])
       expect(assembled.update).toBe("Private update")
       expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
       expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs)
@@ -313,41 +302,6 @@ describe("InstructionState", () => {
     }),
   )
 
-  it.effect("persists chronological updates as system messages", () =>
-    Effect.gen(function* () {
-      const sessionID = SessionSchema.ID.make("ses_instruction_messages")
-      const { db, events } = yield* setup(sessionID)
-      let value = "Initial context"
-      const instructions = source(
-        "test/context",
-        Effect.sync(() => value),
-      )
-      const messages = () =>
-        db
-          .select()
-          .from(SessionMessageTable)
-          .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "system")))
-          .orderBy(asc(SessionMessageTable.seq))
-          .all()
-          .pipe(Effect.orDie)
-
-      // The initial baseline is not chronological history and produces no message.
-      yield* InstructionState.prepare(db, events, instructions, sessionID)
-      expect(yield* messages()).toEqual([])
-
-      value = "Changed context"
-      yield* InstructionState.prepare(db, events, instructions, sessionID)
-      const rows = yield* messages()
-      expect(rows).toHaveLength(1)
-      expect(rows[0]?.data).toMatchObject({ text: "Changed context" })
-      expect(rows.map((row) => row.seq)).toEqual([(yield* instructionEvents(db, sessionID)).at(-1)!.seq])
-
-      // A no-op observation adds nothing.
-      yield* InstructionState.prepare(db, events, instructions, sessionID)
-      expect(yield* messages()).toHaveLength(1)
-    }),
-  )
-
   it.effect("assembles initial instructions without persisting a baseline", () =>
     Effect.gen(function* () {
       const sessionID = SessionSchema.ID.make("ses_instruction_generate_initial")
@@ -356,6 +310,7 @@ describe("InstructionState", () => {
 
       expect(yield* preview(db, sessionID, instructions)).toEqual({
         initial: "Initial context",
+        updates: [],
         update: "",
       })
       expect(yield* instructionEvents(db, sessionID)).toEqual([])
@@ -381,6 +336,7 @@ describe("InstructionState", () => {
 
       expect(yield* preview(db, sessionID, instructions)).toEqual({
         initial: "Committed context",
+        updates: [],
         update: "",
       })
       expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
@@ -432,7 +388,7 @@ describe("InstructionState", () => {
       for (const next of ["initial", "changed", "changed", Instructions.removed] as const) {
         value = next
         yield* InstructionState.observe(db, observedInstructions, observedSessionID).pipe(
-          Effect.flatMap((observation) => InstructionState.commit(db, events, observedInstructions, observation)),
+          Effect.flatMap((observation) => InstructionState.commit(db, events, observation)),
         )
         yield* InstructionState.prepare(db, events, preparedInstructions, preparedSessionID)
       }

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

@@ -284,9 +284,13 @@ describe("Session.create", () => {
       })
       expect(yield* SessionPending.find(db, forkContext[0].id)).toBeUndefined()
       expect(yield* SessionPending.find(db, forkContext[1].id)).toBeUndefined()
+      // Fork-copied messages have no admitted event in the fork aggregate, so
+      // reusing their IDs as prompt IDs is conflicting reuse, not a retry.
       expect(
-        yield* session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
-      ).toMatchObject({ id: forkContext[0].id, type: "user", data: { text: "First" } })
+        yield* session
+          .prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false })
+          .pipe(Effect.flip),
+      ).toMatchObject({ _tag: "Session.PromptConflictError", messageID: forkContext[0].id })
 
       yield* session.prompt({
         sessionID: parent.id,

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

@@ -110,26 +110,4 @@ 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-prompt.test.ts

@@ -553,47 +553,6 @@ describe("Session.prompt", () => {
     }),
   )
 
-  it.effect("reconciles an exact retry from the promoted message without admission history", () =>
-    Effect.gen(function* () {
-      yield* setup
-      const session = yield* Session.Service
-      const bus = yield* Bus.Service
-      const { db } = yield* Database.Service
-      const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
-      const first = yield* session.prompt(input)
-      yield* SessionPending.promote(db, bus, sessionID, "steer")
-      yield* db
-        .delete(EventTable)
-        .where(eq(EventTable.aggregate_id, sessionID))
-        .run()
-        .pipe(Effect.orDie)
-
-      const retried = yield* session.prompt(input)
-
-      expect(retried).toMatchObject({ id: first.id, type: "user", data: { text: first.data.text } })
-      expect(yield* session.messages({ sessionID })).toMatchObject([
-        { id: messageID, type: "user", text: "Fix the failing tests" },
-      ])
-    }),
-  )
-
-  it.effect("ignores delivery when retrying a promoted message", () =>
-    Effect.gen(function* () {
-      yield* setup
-      const session = yield* Session.Service
-      const bus = yield* Bus.Service
-      const { db } = yield* Database.Service
-      const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
-      yield* session.prompt(input)
-      yield* SessionPending.promote(db, bus, sessionID, "steer")
-
-      const retried = yield* session.prompt({ ...input, delivery: "queue" })
-
-      expect(retried).toMatchObject({ id: messageID, type: "user", data: { text: input.text } })
-      expect(yield* admitted(messageID)).toBeUndefined()
-    }),
-  )
-
   it.effect("wakes execution when an exact prompt retry recovers a committed message", () =>
     Effect.gen(function* () {
       yield* setup

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

@@ -1180,7 +1180,7 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
-  it.effect("seeds a fork with the parent's newest instruction values", () =>
+  it.effect("forks instruction values at the selected message instead of the parent's latest state", () =>
     Effect.gen(function* () {
       const session = yield* setup
       yield* runPrompt(session, "First")
@@ -1197,16 +1197,14 @@ describe("SessionRunnerLLM", () => {
           .where(eq(InstructionStateTable.session_id, forked.id))
           .get(),
       ).toMatchObject({
-        initial_values: { "test/context": Instructions.hash("Latest context") },
-        current_values: { "test/context": Instructions.hash("Latest context") },
+        initial_values: { "test/context": Instructions.hash("Changed context") },
+        current_values: { "test/context": Instructions.hash("Changed context") },
       })
       yield* session.prompt({ sessionID: forked.id, text: "Forked", resume: false })
       yield* session.resume(forked.id)
 
-      expect(requests.at(-1)?.system.map((part) => part.text)).toEqual([defaultSystem, "Latest context"])
-      // Copied history keeps the frozen chronological update; no new update is emitted.
-      expect(systemTexts(requests.at(-1)!)).toContain("Changed context")
-      expect(systemTexts(requests.at(-1)!)).not.toContain("Latest context")
+      expect(requests.at(-1)?.system.map((part) => part.text)).toEqual([defaultSystem, "Changed context"])
+      expect(systemTexts(requests.at(-1)!)).toContain("Latest context")
 
       const { db } = yield* Database.Service
       const bus = yield* Bus.Service
@@ -1265,7 +1263,7 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
-  it.effect("re-establishes a fresh baseline when instruction state is missing", () =>
+  it.effect("rebuilds a missing instruction cache without admitting another delta", () =>
     Effect.gen(function* () {
       const session = yield* setup
       const { db } = yield* Database.Service
@@ -1279,15 +1277,13 @@ describe("SessionRunnerLLM", () => {
       expect(requests).toHaveLength(1)
       expect(requests[0]?.system.map((part) => part.text)).toEqual([defaultSystem, "Initial context"])
       expect(messageRoles(requests[0])).toEqual(["user", "user"])
-      // The projected row is authoritative: a missing row admits a fresh baseline
-      // instead of rebuilding from durable events.
       expect(
         yield* db
-          .select({ data: EventTable.data })
+          .select({ id: EventTable.id })
           .from(EventTable)
           .where(eq(EventTable.type, "session.instructions.updated.2"))
           .all(),
-      ).toHaveLength(2)
+      ).toHaveLength(1)
       expect(yield* db.select().from(InstructionStateTable).get()).toMatchObject({
         initial_values: { "test/context": Instructions.hash("Initial context") },
         current_values: { "test/context": Instructions.hash("Initial context") },
@@ -1314,10 +1310,7 @@ describe("SessionRunnerLLM", () => {
       ])
       expect(messageRoles(requests[1])).toEqual(["user", "system", "user"])
       expect(requests[1]?.messages.at(1)?.content).toEqual([{ type: "text", text: "Changed context" }])
-      // The chronological update is a durable client-visible system message.
-      const messages = yield* session.messages({ sessionID })
-      expect(messages).toHaveLength(3)
-      expect(messages[1]).toMatchObject({ type: "system", text: "Changed context" })
+      expect(yield* session.messages({ sessionID })).toHaveLength(2)
       const { db } = yield* Database.Service
       const updates = yield* db
         .select({ data: EventTable.data })
@@ -1334,10 +1327,9 @@ describe("SessionRunnerLLM", () => {
       expect(updates[1]?.data).toEqual({
         sessionID,
         delta: { "test/context": Instructions.hash("Changed context") },
-        text: "Changed context",
       })
       yield* replaySessionProjection(sessionID)
-      expect(yield* session.messages({ sessionID })).toHaveLength(3)
+      expect(yield* session.messages({ sessionID })).toHaveLength(2)
     }),
   )
 
@@ -1604,7 +1596,7 @@ describe("SessionRunnerLLM", () => {
       expect(requests[1]?.messages.at(1)?.content).toEqual([
         { type: "text", text: "System context source removed: test/context" },
       ])
-      expect(yield* session.messages({ sessionID })).toHaveLength(3)
+      expect(yield* session.messages({ sessionID })).toHaveLength(2)
     }),
   )
 
@@ -1716,14 +1708,12 @@ describe("SessionRunnerLLM", () => {
       expect(requests[2]?.messages.filter((message) => message.role === "system")).toHaveLength(2)
       expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([
         "user",
-        "system",
         "user",
         "model-switched",
-        "system",
         "user",
       ])
       yield* replaySessionProjection(sessionID)
-      expect(yield* session.messages({ sessionID })).toHaveLength(6)
+      expect(yield* session.messages({ sessionID })).toHaveLength(4)
       yield* runPrompt(session, "Fourth")
     }),
   )

packages/schema/src/session-event.ts

@@ -186,11 +186,6 @@ export const InstructionsUpdated = Event.durable({
   schema: {
     ...Base,
     delta: Instruction.Delta,
-    /**
-     * The rendered chronological update shown to the model, frozen at emit time.
-     * Absent for the initial baseline observation and for deltas that render empty.
-     */
-    text: Schema.String.pipe(optional),
   },
 })
 export type InstructionsUpdated = typeof InstructionsUpdated.Type

packages/tui/test/cli/tui/diff-viewer.test.tsx

@@ -147,12 +147,12 @@ async function renderDiffViewer(vcsDiff: unknown[], height = 20, initialRoute?:
   const config = createTuiResolvedConfig()
   const transport = createFetch((url) => {
     if (url.pathname !== "/api/vcs/diff") return
+    if (fail) return json({ message: "boom" }, { status: 500 })
     vcsDiffInput = {
       location: { directory: url.searchParams.get("location[directory]") },
       mode: url.searchParams.get("mode"),
       context: url.searchParams.get("context"),
     }
-    if (fail) return json({ message: "boom" }, { status: 500 })
     return json({
       location: { directory: "/repo/session", project: { id: "project-1", directory: "/repo/session" } },
       data: vcsDiff,
@@ -238,7 +238,6 @@ async function renderDiffViewer(vcsDiff: unknown[], height = 20, initialRoute?:
 
   const app = await testRender(() => <Harness />, { width: 80, height })
   await waitForCommand(app, commands, "diff.close")
-  await app.waitFor(() => vcsDiffInput !== undefined)
   return {
     app,
     commands,