diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index c898d96e4c..61416f7ee4 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -211,11 +211,43 @@ export type StreamItem = Schema.Schema.Type // 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] const NO_EVENTS: StepResult["1"] = [] // `response.completed` / `response.incomplete` are clean finishes that emit a -// `finish` event; `response.failed` is a hard failure. All three end the stream, -// so keep this set aligned with `step` and the protocol's terminal predicate. -const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"]) +// `finish` event; `response.failed` and `error` are hard failures. All four end +// the stream, so keep this set aligned with `step` and the protocol's terminal predicate. +const TERMINAL_TYPES = new Set(["error", "response.completed", "response.incomplete", "response.failed"]) export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type) const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => { @@ -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([state, NO_EVENTS]) } diff --git a/packages/ai/src/provider-error.ts b/packages/ai/src/provider-error.ts index ed8976788f..71cadab041 100644 --- a/packages/ai/src/provider-error.ts +++ b/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"]) diff --git a/packages/ai/src/route/transport/websocket.ts b/packages/ai/src/route/transport/websocket.ts index b1d8854de1..dd3c4d8242 100644 --- a/packages/ai/src/route/transport/websocket.ts +++ b/packages/ai/src/route/transport/websocket.ts @@ -26,10 +26,18 @@ type WebSocketConstructorWithHeaders = new ( export class Service extends Context.Service()("@opencode/AI/WebSocketExecutor") {} +const MAX_FRAME_BYTES = 16 * 1024 * 1024 + const transportError = ( method: string, message: string, - input: { readonly operation: TransportOperation; readonly url?: string; readonly code?: string }, + input: { + readonly operation: TransportOperation + readonly url?: string + readonly code?: string + readonly phase?: TransportReason["phase"] + readonly delivery?: TransportReason["delivery"] + }, ) => new AIError({ module: "WebSocketExecutor", @@ -40,9 +48,33 @@ const transportError = ( operation: input.operation, url: input.url, code: input.code, + 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, + transport: error.reason.transport, + operation: error.reason.operation, + code: error.reason.code, + 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 @@ -63,6 +95,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => { url: input.url, operation: "request", code: "closed", + phase: "connect", + delivery: "not-sent", }), ) } @@ -89,6 +123,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => { transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, operation: "request", + phase: "connect", + delivery: "not-sent", }), ), ) @@ -101,6 +137,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => { url: input.url, operation: "request", code: String(event.code), + phase: "connect", + delivery: "not-sent", }), ), ) @@ -131,6 +169,8 @@ const webSocketUrl = (value: string) => url: value, operation: "request", code: "invalid-url", + phase: "prepare", + delivery: "not-sent", }), }) @@ -142,6 +182,8 @@ export const open = (input: WebSocketRequest) => transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", { url: input.url, operation: "request", + phase: "connect", + delivery: "not-sent", }), }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input))) @@ -155,16 +197,43 @@ export const fromWebSocket = ( yield* waitOpen(ws, input) const messages = yield* Queue.bounded>(128) + const oversized = (message: string | Uint8Array) => + typeof message === "string" ? new Blob([message]).size > MAX_FRAME_BYTES : message.byteLength > MAX_FRAME_BYTES + const rejectOversized = (message: string | Uint8Array) => { + if (!oversized(message)) return false + Queue.failCauseUnsafe( + messages, + Cause.fail( + transportError("message", "WebSocket message exceeds the 16 MiB limit", { + url: input.url, + operation: "read", + code: "message-too-large", + phase: "receive", + }), + ), + ) + if (ws.readyState === globalThis.WebSocket.OPEN) ws.close(1009, "Message too large") + return true + } + const onMessage = (event: MessageEvent) => { - if (typeof event.data === "string") return Queue.offerUnsafe(messages, event.data) + if (typeof event.data === "string") { + if (rejectOversized(event.data)) return + return Queue.offerUnsafe(messages, event.data) + } const binary = binaryMessage(event.data) - if (binary) return Queue.offerUnsafe(messages, binary) + if (binary) { + if (rejectOversized(binary)) return + return Queue.offerUnsafe(messages, binary) + } Queue.failCauseUnsafe( messages, Cause.fail( transportError("message", "Unsupported WebSocket message payload", { url: input.url, operation: "read", + code: "message", + phase: "receive", }), ), ) @@ -176,12 +245,13 @@ export const fromWebSocket = ( transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, operation: "read", + code: "message", + phase: "receive", }), ), ) } const onClose = (event: CloseEvent) => { - if (event.code === 1000 || event.code === 1005) return Queue.endUnsafe(messages) Queue.failCauseUnsafe( messages, Cause.fail( @@ -189,6 +259,7 @@ export const fromWebSocket = ( url: input.url, operation: "read", code: String(event.code), + phase: "close", }), ), ) @@ -211,6 +282,8 @@ export const fromWebSocket = ( transportError("sendText", error instanceof Error ? error.message : "Failed to send WebSocket message", { url: input.url, operation: "write", + phase: "send", + delivery: "not-sent", }), }), messages: Stream.fromQueue(messages), @@ -267,6 +340,8 @@ export const json = (input: JsonInput): JsonTransp url: prepared.url, operation: "request", code: "unavailable", + phase: "prepare", + delivery: "not-sent", }), ) } @@ -274,11 +349,27 @@ export const json = (input: JsonInput): 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", + }), + ), + ) }), ) }, diff --git a/packages/ai/src/schema/errors.ts b/packages/ai/src/schema/errors.ts index d03f2f3cf7..aeeabb4b98 100644 --- a/packages/ai/src/schema/errors.ts +++ b/packages/ai/src/schema/errors.ts @@ -106,6 +106,13 @@ export class TransportReason extends Schema.Class("AI.Error.Tra code: 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( diff --git a/packages/ai/test/provider-error.test.ts b/packages/ai/test/provider-error.test.ts index cbf145067a..14be937d59 100644 --- a/packages/ai/test/provider-error.test.ts +++ b/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", () => { diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index cfef867420..6749d8024e 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -11,6 +11,7 @@ import { ToolCallPart, ToolDefinition, ToolResultPart, + TransportReason, Usage, } from "../../src/index.js" import { Auth, LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route.js" @@ -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" }) }), ) diff --git a/packages/ai/test/schema.test.ts b/packages/ai/test/schema.test.ts index 4a01be238a..e0833c2ed8 100644 --- a/packages/ai/test/schema.test.ts +++ b/packages/ai/test/schema.test.ts @@ -11,6 +11,7 @@ import { LanguageModel, ModelID, ProviderID, + TransportReason, Usage, } from "../src/schema/index.js" import { ProviderShared } from "../src/protocols/shared.js" @@ -108,3 +109,25 @@ 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", + transport: "websocket", + operation: "read", + phase: "receive", + delivery: "ambiguous", + recovery: "fail", + }) + + expect(Schema.encodeSync(TransportReason)(reason)).toEqual({ + _tag: "Transport", + message: "connection closed", + transport: "websocket", + operation: "read", + phase: "receive", + delivery: "ambiguous", + recovery: "fail", + }) + expect(Schema.decodeUnknownSync(TransportReason)(Schema.encodeSync(TransportReason)(reason))).toEqual(reason) +}) diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index efc2004433..a4526dc4f4 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/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": diff --git a/packages/core/test/session-error.test.ts b/packages/core/test/session-error.test.ts index d0e14b8fa7..d79050745a 100644 --- a/packages/core/test/session-error.test.ts +++ b/packages/core/test/session-error.test.ts @@ -128,4 +128,52 @@ 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", transport: "http", operation: "request" })), + llm( + new TransportReason({ + message: "connect failed", + transport: "websocket", + operation: "request", + delivery: "not-sent", + phase: "connect", + }), + ), + ] + const ineligible = [ + llm( + new TransportReason({ + message: "send uncertain", + transport: "websocket", + operation: "write", + delivery: "ambiguous", + phase: "send", + }), + ), + llm( + new TransportReason({ + message: "response interrupted", + transport: "websocket", + operation: "read", + delivery: "accepted", + phase: "receive", + }), + ), + llm( + new TransportReason({ + message: "continuation rejected", + transport: "websocket", + operation: "read", + delivery: "rejected", + recovery: "retry-full", + phase: "receive", + }), + ), + ] + + expect(retryable.map(SessionRunnerRetry.isRetryable)).toEqual([true, true]) + expect(ineligible.map(SessionRunnerRetry.isRetryable)).toEqual([false, false, false]) + }) })