fix(ai): harden websocket error contracts

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

@@ -211,11 +211,43 @@ export type StreamItem = Schema.Schema.Type<typeof StreamItem>
 // event-level `error` envelope, so accept all three shapes here.
 // https://www.openresponses.org/specification
 const OpenResponsesErrorPayload = Schema.Struct({
+  type: optionalNull(Schema.String),
   code: optionalNull(Schema.String),
   message: optionalNull(Schema.String),
   param: optionalNull(Schema.String),
 })
 
+const WebSocketErrorHeader = Schema.Union([Schema.String, Schema.Number, Schema.Boolean])
+export const WebSocketErrorEvent = Schema.StructWithRest(
+  Schema.Struct({
+    type: Schema.tag("error"),
+    status: Schema.optional(Schema.Number),
+    status_code: Schema.optional(Schema.Number),
+    code: optionalNull(Schema.String),
+    message: Schema.optional(Schema.String),
+    param: optionalNull(Schema.String),
+    error: optionalNull(OpenResponsesErrorPayload),
+    headers: Schema.optional(Schema.Record(Schema.String, WebSocketErrorHeader)),
+  }),
+  [Schema.Record(Schema.String, Schema.Unknown)],
+)
+const decodeWebSocketErrorEvent = Schema.decodeUnknownEffect(WebSocketErrorEvent)
+
+const decodeKnownErrorEvent = (event: Event) =>
+  decodeWebSocketErrorEvent({
+    ...event,
+    status: typeof event.status === "number" ? event.status : undefined,
+    status_code: typeof event.status_code === "number" ? event.status_code : undefined,
+    headers: ProviderShared.isRecord(event.headers)
+      ? Object.fromEntries(
+          Object.entries(event.headers).filter(
+            (entry): entry is [string, string | number | boolean] =>
+              typeof entry[1] === "string" || typeof entry[1] === "number" || typeof entry[1] === "boolean",
+          ),
+        )
+      : undefined,
+  })
+
 export const Event = Schema.StructWithRest(
   Schema.Struct({
     type: Schema.String,
@@ -240,6 +272,9 @@ export const Event = Schema.StructWithRest(
     message: Schema.optional(Schema.String),
     param: optionalNull(Schema.String),
     error: optionalNull(OpenResponsesErrorPayload),
+    status: Schema.optional(Schema.Unknown),
+    status_code: Schema.optional(Schema.Unknown),
+    headers: Schema.optional(Schema.Unknown),
   }),
   [Schema.Record(Schema.String, Schema.Unknown)],
 )
@@ -632,9 +667,9 @@ export type StepResult = readonly [ParserState, ReadonlyArray<LLMEvent>]
 const NO_EVENTS: StepResult["1"] = []
 
 // `response.completed` / `response.incomplete` are clean finishes that emit a
-// `finish` event; `response.failed` is a hard failure. All three end the stream,
+// `finish` event; `response.failed` and `error` are hard failures. All four end
-// so keep this set aligned with `step` and the protocol's terminal predicate.
+// the stream, so keep this set aligned with `step` and the protocol's terminal predicate.
-const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"])
+const TERMINAL_TYPES = new Set(["error", "response.completed", "response.incomplete", "response.failed"])
 export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type)
 
 const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => {
@@ -969,10 +1004,16 @@ const providerErrorMessage = (event: Event, fallback: string): string => {
 const providerError = (state: ParserState, event: Event, fallback: string) => {
   const code = event.code || event.error?.code || event.response?.error?.code || undefined
   const message = providerErrorMessage(event, fallback)
+  const status =
+    typeof event.status === "number"
+      ? event.status
+      : typeof event.status_code === "number"
+        ? event.status_code
+        : undefined
   return new AIError({
     module: state.id,
     method: "stream",
-    reason: classifyProviderFailure({ message, code }),
+    reason: classifyProviderFailure({ message, code, status }),
   })
 }
 
@@ -1015,7 +1056,11 @@ export const step = (state: ParserState, event: Event) => {
   if (event.type === "response.completed" || event.type === "response.incomplete")
     return Effect.succeed(onResponseFinish(state, event))
   if (event.type === "response.failed") return providerError(state, event, `${state.name} response failed`)
-  if (event.type === "error") return providerError(state, event, `${state.name} stream error`)
+  if (event.type === "error")
+    return decodeKnownErrorEvent(event).pipe(
+      Effect.mapError(() => ProviderShared.eventError(state.id, `${state.name} returned a malformed error event`)),
+      Effect.flatMap(() => providerError(state, event, `${state.name} stream error`)),
+    )
   return Effect.succeed<StepResult>([state, NO_EVENTS])
 }
 

packages/ai/src/provider-error.ts

@@ -67,6 +67,7 @@ const SERVER_CODES = new Set([
   "overloaded_error",
   "server_error",
   "server_is_overloaded",
+  "slow_down",
   "serviceunavailableexception",
 ])
 const INVALID_REQUEST_CODES = new Set(["invalid_prompt", "invalid_request_error", "validationexception"])

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

@@ -29,14 +29,45 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/AI
 const transportError = (
   method: string,
   message: string,
-  input: { readonly url?: string; readonly kind?: string } = {},
+  input: {
+    readonly url?: string
+    readonly kind?: string
+    readonly phase?: TransportReason["phase"]
+    readonly delivery?: TransportReason["delivery"]
+  } = {},
 ) =>
   new AIError({
     module: "WebSocketExecutor",
     method,
-    reason: new TransportReason({ message, url: input.url, kind: input.kind }),
+    reason: new TransportReason({
+      message,
+      url: input.url,
+      kind: input.kind,
+      phase: input.phase,
+      delivery: input.delivery,
+    }),
   })
 
+const annotateTransportError = (
+  error: AIError,
+  input: { readonly phase: TransportReason["phase"]; readonly delivery: TransportReason["delivery"] },
+) =>
+  error.reason._tag === "Transport"
+    ? new AIError({
+        module: error.module,
+        method: error.method,
+        reason: new TransportReason({
+          message: error.reason.message,
+          kind: error.reason.kind,
+          url: error.reason.url,
+          http: error.reason.http,
+          phase: input.phase,
+          delivery: input.delivery,
+          recovery: error.reason.recovery,
+        }),
+      })
+    : error
+
 const eventMessage = (event: Event) => {
   if ("message" in event && typeof event.message === "string") return event.message
   return event.type
@@ -56,6 +87,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
       transportError("open", `WebSocket closed before opening (state ${ws.readyState})`, {
         url: input.url,
         kind: "open",
+        phase: "connect",
+        delivery: "not-sent",
       }),
     )
   }
@@ -79,7 +112,12 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
       cleanup()
       resume(
         Effect.fail(
-          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
+          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
+            url: input.url,
+            kind: "open",
+            phase: "connect",
+            delivery: "not-sent",
+          }),
         ),
       )
     }
@@ -90,6 +128,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
           transportError("open", `WebSocket closed before opening with code ${event.code}`, {
             url: input.url,
             kind: "open",
+            phase: "connect",
+            delivery: "not-sent",
           }),
         ),
       )
@@ -119,6 +159,8 @@ const webSocketUrl = (value: string) =>
       transportError("prepare", error instanceof Error ? error.message : "Invalid WebSocket URL", {
         url: value,
         kind: "websocket",
+        phase: "prepare",
+        delivery: "not-sent",
       }),
   })
 
@@ -130,6 +172,8 @@ export const open = (input: WebSocketRequest) =>
       transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
         url: input.url,
         kind: "open",
+        phase: "connect",
+        delivery: "not-sent",
       }),
   }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
 
@@ -150,7 +194,11 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", "Unsupported WebSocket message payload", { url: input.url, kind: "message" }),
+          transportError("message", "Unsupported WebSocket message payload", {
+            url: input.url,
+            kind: "message",
+            phase: "receive",
+          }),
         ),
       )
     }
@@ -158,16 +206,23 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, kind: "message" }),
+          transportError("message", `WebSocket error: ${eventMessage(event)}`, {
+            url: input.url,
+            kind: "message",
+            phase: "receive",
+          }),
         ),
       )
     }
     const onClose = (event: CloseEvent) => {
-      if (event.code === 1000 || event.code === 1005) return Queue.endUnsafe(messages)
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket closed with code ${event.code}`, { url: input.url, kind: "close" }),
+          transportError("message", `WebSocket closed with code ${event.code}`, {
+            url: input.url,
+            kind: "close",
+            phase: "close",
+          }),
         ),
       )
     }
@@ -189,6 +244,8 @@ export const fromWebSocket = (
             transportError("sendText", error instanceof Error ? error.message : "Failed to send WebSocket message", {
               url: input.url,
               kind: "write",
+              phase: "send",
+              delivery: "not-sent",
             }),
         }),
       messages: Stream.fromQueue(messages),
@@ -244,6 +301,8 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
         transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
           url: prepared.url,
           kind: "websocket",
+          phase: "prepare",
+          delivery: "not-sent",
         }),
       )
     }
@@ -251,11 +310,27 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
     return Stream.unwrap(
       Effect.gen(function* () {
         const connection = yield* Effect.acquireRelease(
-          webSocket.open({ url: prepared.url, headers: prepared.headers }),
+          webSocket
+            .open({ url: prepared.url, headers: prepared.headers })
+            .pipe(
+              Effect.mapError((error) => annotateTransportError(error, { phase: "connect", delivery: "not-sent" })),
+            ),
           (connection) => connection.close,
         )
         yield* connection.sendText(prepared.message)
-        return connection.messages.pipe(Stream.map((message) => messageText(message, decoder)))
+        let observed = false
+        return 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",
+            }),
+          ),
+        )
       }),
     )
   },

packages/ai/src/schema/errors.ts

@@ -98,6 +98,13 @@ export class TransportReason extends Schema.Class<TransportReason>("AI.Error.Tra
   kind: Schema.optional(Schema.String),
   url: Schema.optional(Schema.String),
   http: Schema.optional(HttpContext),
+  phase: Schema.optional(
+    Schema.Literals(["prepare", "queue", "connect", "send", "receive", "decode", "complete", "fallback", "close"]),
+  ),
+  delivery: Schema.optional(Schema.Literals(["not-sent", "rejected", "ambiguous", "accepted"])),
+  recovery: Schema.optional(
+    Schema.Literals(["retry-connect", "retry-full", "rotate-and-retry-full", "fallback-http", "fail"]),
+  ),
 }) {}
 
 export class InvalidProviderOutputReason extends Schema.Class<InvalidProviderOutputReason>(

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

@@ -69,10 +69,10 @@ describe("provider error classification", () => {
 
   test("classifies V1 overloaded provider codes", () => {
     expect(
-      ['{"code":"resource_exhausted"}', '{"code":"service_unavailable"}'].map(
+      ['{"code":"resource_exhausted"}', '{"code":"service_unavailable"}', '{"code":"slow_down"}'].map(
         (message) => classifyProviderFailure({ message })._tag,
       ),
-    ).toEqual(["ProviderInternal", "ProviderInternal"])
+    ).toEqual(["ProviderInternal", "ProviderInternal", "ProviderInternal"])
   })
 
   test("classifies transient client statuses as provider internal", () => {

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

@@ -11,6 +11,7 @@ import {
   ToolCallPart,
   ToolDefinition,
   ToolResultPart,
+  TransportReason,
   Usage,
 } from "../../src"
 import { Auth, LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route"
@@ -288,6 +289,114 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("terminates WebSocket control events without waiting for the socket to close", () =>
+    Effect.gen(function* () {
+      const events = [
+        { type: "error", error: { code: "slow_down", message: "Try later" } },
+        {
+          type: "error",
+          status_code: 429,
+          message: "Rate limited",
+          headers: { "retry-after": 1, "x-request-id": "request", cached: false, invalid: [] },
+        },
+        {
+          type: "response.failed",
+          response: { error: { code: "server_error", message: "Unavailable" } },
+        },
+        { type: "error", status: "not-a-status", message: "Malformed status" },
+      ]
+
+      const errors = yield* Effect.forEach(events, (event) =>
+        LLMClient.generate(
+          LLM.request({
+            model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+              "gpt-4.1-mini",
+            ),
+            prompt: "Say hello.",
+          }),
+        ).pipe(
+          Effect.provide(
+            LLMClient.layer.pipe(
+              Layer.provide(
+                Layer.mergeAll(
+                  Layer.succeed(
+                    RequestExecutor.Service,
+                    RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+                  ),
+                  Layer.succeed(
+                    WebSocketExecutor.Service,
+                    WebSocketExecutor.Service.of({
+                      open: () =>
+                        Effect.succeed({
+                          sendText: () => Effect.void,
+                          messages: Stream.make(ProviderShared.encodeJson(event)).pipe(Stream.concat(Stream.never)),
+                          close: Effect.void,
+                        }),
+                    }),
+                  ),
+                ),
+              ),
+            ),
+          ),
+          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.mergeAll(
+        Layer.succeed(
+          RequestExecutor.Service,
+          RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
+        ),
+        Layer.succeed(
+          WebSocketExecutor.Service,
+          WebSocketExecutor.Service.of({
+            open: () =>
+              Effect.succeed({
+                sendText: () => Effect.void,
+                messages: streams.shift() ?? Stream.die("unexpected WebSocket open"),
+                close: Effect.void,
+              }),
+          }),
+        ),
+      )
+      const model = OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
+        "gpt-4.1-mini",
+      )
+
+      const errors = yield* Effect.forEach(["first", "second"], (prompt) =>
+        LLMClient.generate(LLM.request({ model, prompt })).pipe(
+          Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))),
+          Effect.flip,
+        ),
+      )
+
+      expect(errors.map((error) => error.reason)).toEqual([
+        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "ambiguous" }),
+        expect.objectContaining({ _tag: "Transport", phase: "close", delivery: "accepted" }),
+      ])
+    }),
+  )
+
   it.effect("fails immediately when WebSocket is already closed", () =>
     Effect.gen(function* () {
       const error = yield* WebSocketExecutor.fromWebSocket(
@@ -297,6 +406,7 @@ describe("OpenAI Responses route", () => {
       ).pipe(Effect.flip)
 
       expect(error.message).toContain("closed before opening")
+      expect(error.reason).toMatchObject({ _tag: "Transport", phase: "connect", delivery: "not-sent" })
     }),
   )
 

packages/ai/test/schema.test.ts

@@ -11,6 +11,7 @@ import {
   LanguageModel,
   ModelID,
   ProviderID,
+  TransportReason,
   Usage,
 } from "../src/schema"
 import { ProviderShared } from "../src/protocols/shared"
@@ -108,3 +109,21 @@ test("AI errors expose the shared runtime tag", async () => {
     await Effect.runPromise(Effect.fail(error).pipe(Effect.catchTag("AI.Error", () => Effect.succeed("caught")))),
   ).toBe("caught")
 })
+
+test("transport errors serialize execution facts", () => {
+  const reason = new TransportReason({
+    message: "connection closed",
+    phase: "receive",
+    delivery: "ambiguous",
+    recovery: "fail",
+  })
+
+  expect(Schema.encodeSync(TransportReason)(reason)).toEqual({
+    _tag: "Transport",
+    message: "connection closed",
+    phase: "receive",
+    delivery: "ambiguous",
+    recovery: "fail",
+  })
+  expect(Schema.decodeUnknownSync(TransportReason)(Schema.encodeSync(TransportReason)(reason))).toEqual(reason)
+})

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

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

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

@@ -110,4 +110,26 @@ describe("toSessionError", () => {
     expect(eligible.map(SessionRunnerRetry.isRetryable)).toEqual([true, true, true])
     expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false, false, false, false, false, false])
   })
+
+  test("retries transport failures only when delivery is absent or not sent", () => {
+    const retryable = [
+      llm(new TransportReason({ message: "http transport" })),
+      llm(new TransportReason({ message: "connect failed", delivery: "not-sent", phase: "connect" })),
+    ]
+    const ineligible = [
+      llm(new TransportReason({ message: "send uncertain", delivery: "ambiguous", phase: "send" })),
+      llm(new TransportReason({ message: "response interrupted", delivery: "accepted", phase: "receive" })),
+      llm(
+        new TransportReason({
+          message: "continuation rejected",
+          delivery: "rejected",
+          recovery: "retry-full",
+          phase: "receive",
+        }),
+      ),
+    ]
+
+    expect(retryable.map(SessionRunnerRetry.isRetryable)).toEqual([true, true])
+    expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false, false])
+  })
 })