feat(core): enable the responses websocket by default

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

@@ -153,6 +153,15 @@ export const driver = (input: DriverInput): WebSocketChannelDriver => {
           const rejection = code(event)
           if (rejection === "previous_response_not_found") return rejected(observation, "retry-full")
           if (rejection === "websocket_connection_limit_reached") return rejected(observation, "rotate-and-retry-full")
+          // Only the continuation distinguishes an incremental send from a full one, so an unclassified
+          // invalid request there is retried full; Codex reports a stale previous_response_id that way, with
+          // no code. Classified failures such as context overflow keep their runner-owned recovery.
+          if (
+            create.mode === "incremental" &&
+            observation.error.reason._tag === "InvalidRequest" &&
+            observation.error.reason.classification === undefined
+          )
+            return rejected(observation, "retry-full")
         }
         if (observation.type !== "completed") return observation
         // A trigger installs a different context window. Clear the append baseline, retaining the socket.

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

@@ -115,8 +115,11 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
     }
     const onAbort = () => {
       cleanup()
-      if (ws.readyState !== globalThis.WebSocket.CLOSED && ws.readyState !== globalThis.WebSocket.CLOSING)
-        ws.close(1000)
+      if (ws.readyState === globalThis.WebSocket.CLOSED || ws.readyState === globalThis.WebSocket.CLOSING) return
+      // Node's ws reports an aborted handshake as an error event on the next tick; with no listener left
+      // after cleanup, EventEmitter would throw it as an uncaught exception.
+      ws.addEventListener("error", () => {}, { once: true })
+      ws.close(1000)
     }
     const onOpen = () => {
       cleanup()

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

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { ConfigProvider, Effect, Layer, Ref, Stream } from "effect"
+import { ConfigProvider, Effect, Layer, Ref, Schema, Stream } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import {
   LLM,
@@ -30,6 +30,7 @@ import * as Azure from "../../src/providers/azure.js"
 import * as OpenAI from "../../src/providers/openai.js"
 import * as XAI from "../../src/providers/xai.js"
 import * as OpenAIResponses from "../../src/protocols/openai-responses.js"
+import { OpenResponses } from "../../src/protocols/open-responses.js"
 import { OpenResponsesContinuation } from "../../src/protocols/open-responses-continuation.js"
 import * as ProviderShared from "../../src/protocols/shared.js"
 import { continuationRequest, nativeOpenAIResponsesContinuation } from "../continuation-scenarios.js"
@@ -69,14 +70,34 @@ const baseChannelDriver = (message: string): WebSocketChannelDriver => ({
   },
 })
 
-const continuationDriver = (request: Readonly<Record<string, unknown>>) => {
+/** Classifies error frames the way the production channel does, so recovery can read the canonical reason. */
+const classifyingChannelDriver = (message: string): WebSocketChannelDriver => {
+  const base = baseChannelDriver(message)
+  const decodeEvent = Schema.decodeUnknownSync(OpenResponses.protocol.stream.event)
+  return {
+    ...base,
+    observe: (create, frame) =>
+      base.observe(create, frame).pipe(
+        Effect.map((observation) =>
+          observation.type === "provider-failure"
+            ? {
+                ...observation,
+                error: OpenResponses.providerFailure(decodeEvent(frame), "stream error", frame),
+              }
+            : observation,
+        ),
+      ),
+  }
+}
+
+const continuationDriver = (request: Readonly<Record<string, unknown>>, base = baseChannelDriver) => {
   const message = ProviderShared.encodeJson(request)
   return OpenResponsesContinuation.driver({
     id: "openai-responses",
     name: "OpenAI Responses",
     request,
     message,
-    base: baseChannelDriver(message),
+    base: base(message),
   })
 }
 
@@ -852,6 +873,53 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("retries an incremental send in full when the provider rejects it without a code", () =>
+    Effect.gen(function* () {
+      const firstRequest = {
+        type: "response.create",
+        model: "gpt-5.2",
+        store: false,
+        input: [{ role: "user", content: [{ type: "input_text", text: "First" }] }],
+      }
+      const first = continuationDriver(firstRequest, classifyingChannelDriver)
+      const saved = checkpoint(
+        yield* first.observe(
+          yield* first.create(undefined),
+          ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_1" } }),
+        ),
+      )
+      const second = continuationDriver(
+        {
+          ...firstRequest,
+          input: [...firstRequest.input, { role: "user", content: [{ type: "input_text", text: "Second" }] }],
+        },
+        classifyingChannelDriver,
+      )
+      // Codex reports a stale previous_response_id as a plain invalid_request_error.
+      const stale = ProviderShared.encodeJson({
+        type: "error",
+        error: { type: "invalid_request_error", message: "Invalid `previous_response_id`." },
+      })
+      const incremental = yield* second.create(saved)
+      expect(incremental.mode).toBe("incremental")
+      expect(yield* second.observe(incremental, stale)).toMatchObject({ type: "rejected", recovery: "retry-full" })
+
+      // A full send has no continuation to blame, so the same error stays a provider failure.
+      const full = yield* second.create(undefined)
+      expect(yield* second.observe(full, stale)).toMatchObject({ type: "provider-failure" })
+
+      // A classified failure keeps its runner-owned recovery instead of resending the whole context.
+      const overflow = ProviderShared.encodeJson({
+        type: "error",
+        error: { type: "invalid_request_error", code: "context_length_exceeded", message: "Too long" },
+      })
+      expect(yield* second.observe(yield* second.create(saved), overflow)).toMatchObject({
+        type: "provider-failure",
+        error: { reason: { _tag: "InvalidRequest", classification: "context-overflow" } },
+      })
+    }),
+  )
+
   it.effect("builds WebSocket and HTTP fallback from the same final request", () =>
     Effect.gen(function* () {
       const attempts = yield* Ref.make(0)

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

@@ -1369,6 +1369,7 @@ export type ProviderInfo = {
   activation: "auto" | "enabled" | "disabled"
   package: string
   compaction?: ProviderCompaction
+  websocket?: boolean
   settings?: { [x: string]: any }
   headers?: { [x: string]: string }
   body?: { [x: string]: any }
@@ -1849,6 +1850,7 @@ export type ModelInfo = {
   compatibility?: ModelCompatibility
   package?: string
   compaction?: ProviderCompaction
+  websocket?: boolean
   settings?: { [x: string]: any }
   headers?: { [x: string]: string }
   body?: { [x: string]: any }
@@ -2025,6 +2027,7 @@ export type ConfigEntry =
         providers?: {
           [x: string]: {
             compaction?: ProviderCompaction
+            websocket?: boolean
             canonical?: string
             name?: string
             env?: Array<string>
@@ -2035,6 +2038,7 @@ export type ConfigEntry =
             models?: {
               [x: string]: {
                 compaction?: ProviderCompaction
+                websocket?: boolean
                 modelID?: string
                 family?: string
                 name?: string

packages/core/src/catalog.ts

@@ -77,6 +77,7 @@ const layer = Layer.effect(
         ...(provider.canonical === undefined ? {} : { canonical: provider.canonical }),
         package: model.package ?? provider.package,
         compaction: model.compaction ?? provider.compaction,
+        websocket: model.websocket ?? provider.websocket,
         settings: Provider.mergeOverlay(provider.settings, model.settings),
         headers: Provider.mergeHeaders(provider.headers, model.headers),
         body: Provider.mergeOverlay(provider.body, model.body),

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

@@ -58,6 +58,7 @@ export const Plugin = define({
           if (item.name !== undefined) provider.name = item.name
           if (item.package !== undefined) provider.package = item.package
           if (item.compaction !== undefined) provider.compaction = { ...item.compaction }
+          if (item.websocket !== undefined) provider.websocket = item.websocket
           if (item.settings !== undefined) provider.settings = Provider.mergeOverlay(provider.settings, item.settings)
           if (item.headers !== undefined) provider.headers = Provider.mergeHeaders(provider.headers, item.headers)
           if (item.body !== undefined) provider.body = Provider.mergeOverlay(provider.body, item.body)
@@ -78,6 +79,7 @@ export const Plugin = define({
               model.compatibility = { ...model.compatibility, ...config.compatibility }
             if (config.package !== undefined) model.package = config.package
             if (config.compaction !== undefined) model.compaction = { ...config.compaction }
+            if (config.websocket !== undefined) model.websocket = config.websocket
             if (config.settings !== undefined) model.settings = Provider.mergeOverlay(model.settings, config.settings)
             if (config.headers !== undefined) model.headers = Provider.mergeHeaders(model.headers, config.headers)
             if (config.body !== undefined) model.body = Provider.mergeOverlay(model.body, config.body)

packages/core/src/model-resolver.ts

@@ -85,6 +85,8 @@ export interface Resolved {
   readonly limit: Info["limit"]
   /** Model policy overrides the provider policy; omitted means local compaction. */
   readonly compaction?: Info["compaction"]
+  /** Whether the session WebSocket may carry this model's requests when the route supports it. */
+  readonly websocket: boolean
 }
 
 export interface Interface {
@@ -321,6 +323,7 @@ export const layer = Layer.effect(
         cost: selected.cost,
         limit: selected.limit,
         compaction: selected.compaction,
+        websocket: selected.websocket ?? true,
       }
     })
     return Service.of({

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

@@ -6,7 +6,7 @@ import type { SessionRequestKind } from "@opencode/plugin/effect/session"
 import type { Agent } from "@opencode/schema/agent"
 import type { Model } from "@opencode/schema/model"
 import type { Content } from "@opencode/schema/tool"
-import { Cause, Config, Context, Effect, Layer, Result, Stream } from "effect"
+import { Cause, Context, Effect, Layer, Result, Stream } from "effect"
 import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 import { makeLocationNode } from "@opencode/util/effect/app-node"
 import { App } from "../app.js"
@@ -27,9 +27,6 @@ const IMAGE_BYTES_TARGET = 15 * 1024 * 1024 // 15 MiB
 const IMAGE_REMOVED =
   "[This image was removed to reduce the request size and is no longer visible. Do not make claims about its contents from memory. If needed, retrieve it again with an available tool or ask the user to attach it again.]"
 
-const responsesWebSocketFlag = (providerID: string) =>
-  `OPENCODE_EXPERIMENTAL_${providerID.replace(/[^a-zA-Z0-9]+/g, "_").toUpperCase()}_RESPONSES_WEBSOCKET`
-
 /** Failures a prepared execution can surface: infrastructure errors plus user declines resurfaced from the defect tunnel. */
 export type ExecuteError = Tool.Error | Permission.DeclinedError | QuestionTool.CancelledError
 
@@ -364,13 +361,6 @@ export const layer = Layer.effect(
       const hasHttpHooks =
         (yield* hooks.has("session", "http.request", resolved.ref.providerID)) ||
         (yield* hooks.has("session", "http.response", resolved.ref.providerID))
-      const webSocket =
-        resolved.capabilities.responsesWebsockets === true
-          ? yield* Config.boolean(responsesWebSocketFlag(resolved.ref.providerID)).pipe(
-              Config.withDefault(false),
-              Effect.orDie,
-            )
-          : false
       const http = hasHttpHooks
         ? httpMiddleware(hooks, {
             sessionID: session.id,
@@ -379,9 +369,13 @@ export const layer = Layer.effect(
             kind: input.kind,
           })
         : undefined
+      // HTTP hooks must observe every request, so they keep the provider on HTTP.
       const options: StreamOptions = {
         ...(http ? { http } : {}),
-        ...(input.webSocket === "session" && webSocket && !hasHttpHooks
+        ...(input.webSocket === "session" &&
+        !hasHttpHooks &&
+        resolved.capabilities.responsesWebsockets === true &&
+        resolved.websocket
           ? { webSocket: transport.bind(session.id) }
           : {}),
       }

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

@@ -20,6 +20,7 @@ import { webSocketConstructor } from "../effect/app-node-platform.js"
 
 const ROTATE_AFTER_MS = 55 * 60 * 1000
 const INBOUND_CAPACITY = 128
+const CONNECT_TIMEOUT = "10 seconds"
 const IDLE_TIMEOUT = "5 minutes"
 const events = Metric.counter("opencode_session_websocket_events_total", {
   description: "Session WebSocket lifecycle events",
@@ -167,7 +168,20 @@ export const makeLayer = (connector: WebSocketConnector) =>
         return yield* Effect.uninterruptibleMask((restore) =>
           Effect.gen(function* () {
             const connection = yield* restore(
-              connector.open(exchange.connect).pipe(Effect.withSpan("SessionModelTransport.connect")),
+              connector.open(exchange.connect).pipe(
+                Effect.timeoutOrElse({
+                  duration: CONNECT_TIMEOUT,
+                  orElse: () =>
+                    transportError("Timed out opening the Session WebSocket", {
+                      url: exchange.connect.url,
+                      operation: "request",
+                      code: "connect-timeout",
+                      phase: "connect",
+                      delivery: "not-sent",
+                    }),
+                }),
+                Effect.withSpan("SessionModelTransport.connect"),
+              ),
             )
             if (owner.closed) {
               yield* connection.close
@@ -294,20 +308,22 @@ export const makeLayer = (connector: WebSocketConnector) =>
         const channel = owner.channel
           ? owner.channel
           : yield* open(owner, exchange, key).pipe(
-              Effect.catch((error) =>
-                error.reason._tag === "Transport" && error.reason.code === "owner-closed"
-                  ? Effect.fail(error)
-                  : Effect.logWarning("session websocket connect failed; using http", {
-                      sessionTransport: "websocket",
-                      phase: "connect",
-                      delivery: "not-sent",
-                      code: error.reason._tag === "Transport" ? error.reason.code : error.reason._tag,
-                    }).pipe(
-                      Effect.andThen(metric("connect_failure")),
-                      Effect.andThen(metric("fallback")),
-                      Effect.as(undefined),
-                    ),
-              ),
+              Effect.catch((error) => {
+                if (error.reason._tag === "Transport" && error.reason.code === "owner-closed") return Effect.fail(error)
+                // Any connect failure, transient or not, pins the Session to HTTP until restart or move:
+                // a network that refuses the upgrade would otherwise charge every step for a failed connect.
+                owner.httpFallback = true
+                return Effect.logWarning("session websocket connect failed; using http", {
+                  sessionTransport: "websocket",
+                  phase: "connect",
+                  delivery: "not-sent",
+                  code: error.reason._tag === "Transport" ? error.reason.code : error.reason._tag,
+                }).pipe(
+                  Effect.andThen(metric("connect_failure")),
+                  Effect.andThen(metric("fallback")),
+                  Effect.as(undefined),
+                )
+              }),
             )
         if (!channel) return fallback(exchange)
 

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

@@ -60,6 +60,7 @@ export const resolved = (
     readonly cost: Model.Info["cost"]
     readonly limit: Model.Info["limit"]
     readonly compaction?: Provider.Compaction
+    readonly websocket?: boolean
   },
 ): Resolved => ({
   model,
@@ -72,6 +73,7 @@ export const resolved = (
   cost: options.cost,
   limit: options.limit,
   compaction: options.compaction,
+  websocket: options.websocket ?? true,
 })
 
 const layer = Layer.effect(

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

@@ -35,8 +35,10 @@ export function isRetryable(error: AIError) {
     case "RateLimit":
     case "ProviderInternal":
       return true
+    // HTTP transport errors carry no delivery and always retry. WebSocket marks accepted and rejected
+    // requests as final; not-sent and ambiguous (no frame observed) are still pre-output.
     case "Transport":
-      return error.reason.delivery === undefined || error.reason.delivery === "not-sent"
+      return error.reason.delivery !== "accepted" && error.reason.delivery !== "rejected"
     case "InvalidProviderOutput":
       return error.reason.classification === "incomplete-stream"
     // Unrecognized failures retry: classification records affirmative

packages/core/test/config/provider.test.ts

@@ -80,6 +80,33 @@ describe("ConfigProviderPlugin.Plugin", () => {
     }),
   )
 
+  it.effect("inherits the provider websocket policy with model overrides", () =>
+    Effect.gen(function* () {
+      const catalog = yield* Catalog.Service
+      yield* addPlugin([
+        new Document({
+          type: "document",
+          info: decode({
+            providers: {
+              custom: {
+                package: "@opencode/ai/providers/openai/responses",
+                websocket: false,
+                models: { inherited: {}, override: { websocket: true } },
+              },
+              default: { package: "@opencode/ai/providers/openai/responses", models: { untouched: {} } },
+            },
+          }),
+        }),
+      ])
+      const inherited = required(yield* catalog.model.get(Provider.ID.make("custom"), Model.ID.make("inherited")))
+      const override = required(yield* catalog.model.get(Provider.ID.make("custom"), Model.ID.make("override")))
+      const untouched = required(yield* catalog.model.get(Provider.ID.make("default"), Model.ID.make("untouched")))
+      expect(inherited.websocket).toBe(false)
+      expect(override.websocket).toBe(true)
+      expect(untouched.websocket).toBeUndefined()
+    }),
+  )
+
   it.effect("adds key auth for custom providers without env credentials", () =>
     Effect.gen(function* () {
       const integrations = yield* Integration.Service

packages/core/test/generate.test.ts

@@ -94,6 +94,7 @@ resolverIt.effect("resolves dynamic models with their catalog metadata", () =>
       capabilities: selected.capabilities,
       cost: selected.cost,
       limit: selected.limit,
+      websocket: true,
     })
   }),
 )