From 0eacf8d73667ef441ea1b1f822129e48e9a10cfc Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Fri, 17 Jul 2026 14:41:38 -0500 Subject: [PATCH] fix(ai): handle remaining provider failures --- packages/ai/src/index.ts | 1 + .../ai/src/protocols/bedrock-event-stream.ts | 59 ++++++++- packages/ai/src/provider-error.ts | 51 ++++++-- packages/ai/test/executor.test.ts | 2 +- packages/ai/test/provider-error.test.ts | 48 ++++++- .../ai/test/provider/bedrock-converse.test.ts | 52 +++++++- packages/core/src/aisdk.ts | 38 ++++-- packages/core/test/aisdk.test.ts | 121 +++++++++++++++++- 8 files changed, 332 insertions(+), 40 deletions(-) diff --git a/packages/ai/src/index.ts b/packages/ai/src/index.ts index 6d950aabed..b551393496 100644 --- a/packages/ai/src/index.ts +++ b/packages/ai/src/index.ts @@ -3,6 +3,7 @@ export { Auth } from "./route/auth" export { Provider } from "./provider" export { ProviderPackage } from "./provider-package" export { classifyProviderFailure, isContextOverflow, isContextOverflowFailure } from "./provider-error" +export type { ProviderFailure } from "./provider-error" export type { RouteModelInput, RouteRoutedModelInput, diff --git a/packages/ai/src/protocols/bedrock-event-stream.ts b/packages/ai/src/protocols/bedrock-event-stream.ts index 0312ea7d57..b854f8a9b5 100644 --- a/packages/ai/src/protocols/bedrock-event-stream.ts +++ b/packages/ai/src/protocols/bedrock-event-stream.ts @@ -1,6 +1,6 @@ import { EventStreamCodec } from "@smithy/eventstream-codec" import { fromUtf8, toUtf8 } from "@smithy/util-utf8" -import { Effect, Stream } from "effect" +import { Effect, Option, Schema, Stream } from "effect" import { Framing } from "../route/framing" import { ProviderShared } from "./shared" @@ -53,8 +53,13 @@ const consumeFrames = (route: string) => (state: FrameBufferState, chunk: Uint8A }) cursor = { buffer: cursor.buffer, offset: cursor.offset + totalLength } - if (decoded.headers[":message-type"]?.value !== "event") continue - const eventType = decoded.headers[":event-type"]?.value + const messageType = decoded.headers[":message-type"]?.value + const eventType = + messageType === "event" + ? decoded.headers[":event-type"]?.value + : messageType === "exception" + ? decoded.headers[":exception-type"]?.value + : undefined if (typeof eventType !== "string") continue const payload = utf8.decode(decoded.body) if (!payload) continue @@ -84,4 +89,52 @@ export const framing = (route: string): Framing.Definition => ({ frame: (bytes) => bytes.pipe(Stream.mapAccumEffect(() => initialFrameBuffer, consumeFrames(route))), }) +class StreamExceptionError extends Error { + constructor( + message: string, + readonly code: string, + ) { + super(message) + } +} + +// The AI SDK Bedrock decoder ignores AWS exception frames before its language +// model stream can expose them. Fail the byte stream first so the shared AI SDK +// adapter can classify the transport error instead of accepting a false finish. +export function monitorExceptions(response: Response) { + if (!response.body || !response.headers.get("content-type")?.includes("application/vnd.amazon.eventstream")) + return response + let state = initialFrameBuffer + const body = response.body.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + state = appendChunk(state, chunk) + while (state.buffer.length - state.offset >= 4) { + const view = state.buffer.subarray(state.offset) + const totalLength = new DataView(view.buffer, view.byteOffset, view.byteLength).getUint32(0, false) + if (view.length < totalLength) break + const decoded = eventCodec.decode(view.subarray(0, totalLength)) + state = { buffer: state.buffer, offset: state.offset + totalLength } + const exceptionType = decoded.headers[":exception-type"]?.value + if (decoded.headers[":message-type"]?.value === "exception" && typeof exceptionType === "string") { + const payload = Option.getOrUndefined( + Schema.decodeUnknownOption(Schema.UnknownFromJsonString)(utf8.decode(decoded.body)), + ) + const message = + ProviderShared.isRecord(payload) && typeof payload.message === "string" ? payload.message : undefined + controller.error(new StreamExceptionError(message ?? `Bedrock ${exceptionType}`, exceptionType)) + return + } + } + controller.enqueue(chunk) + }, + }), + ) + return new Response(body, { + headers: new Headers(response.headers), + status: response.status, + statusText: response.statusText, + }) +} + export * as BedrockEventStream from "./bedrock-event-stream" diff --git a/packages/ai/src/provider-error.ts b/packages/ai/src/provider-error.ts index 57c2b5604b..70a3f82631 100644 --- a/packages/ai/src/provider-error.ts +++ b/packages/ai/src/provider-error.ts @@ -46,26 +46,40 @@ export const isContextOverflowFailure = (failure: unknown) => : Schema.is(ProviderErrorEvent)(failure) && failure.classification === "context-overflow" const decodeJson = Schema.decodeUnknownOption(Schema.UnknownFromJsonString) +const CONTENT_POLICY_CODES = new Set(["content_filter", "content_policy_violation", "safety"]) const QUOTA_CODES = new Set(["insufficient_quota", "usage_not_included", "billing_error"]) +const RATE_LIMIT_CODES = new Set(["resource_exhausted", "throttlingexception", "too_many_requests"]) const SERVER_CODES = new Set([ "api_error", + "internal", "internal_error", + "internal_server_error", "internalserverexception", "modelstreamerrorexception", + "modeltimeoutexception", "overloaded_error", + "response_error", "server_error", "server_is_overloaded", "serviceunavailableexception", ]) const INVALID_REQUEST_CODES = new Set([ + "invalid_argument", "invalid_prompt", "invalid_request_error", + "model_not_found", + "not_found", + "not_found_error", + "resourcenotfoundexception", "request_too_large", "validationexception", ]) -const RATE_LIMIT_TEXT = /rate increased too quickly|rate[-_\s]?limit|too[_\s]?many[_\s]?requests/i +const RATE_LIMIT_TEXT = /rate increased too quickly|rate[-_\s]?limit|throttl|too[_\s]?many[_\s]?requests/i const QUOTA_TEXT = /insufficient[-_\s]?quota|quota[-_\s]?exceeded/i -const CONTENT_POLICY_TEXT = /content[-_\s]?policy|content_filter|safety/i +const CONTENT_POLICY_TEXT = + /content[-_\s]?(?:filter|policy)|safety (?:filter|policy|rating)|blocked (?:by|due to) safety/i +const INVALID_REQUEST_TEXT = /validation (?:error|exception)/i +const SERVER_TEXT = /internal server error|service unavailable/i export interface ProviderFailure { readonly message: string @@ -98,7 +112,8 @@ export function classifyProviderFailure(input: ProviderFailure): LLMError["reaso texts.some(isContextOverflow)) ) return new InvalidRequestReason({ ...common, classification: "context-overflow" }) - if (texts.some((text) => CONTENT_POLICY_TEXT.test(text))) return new ContentPolicyReason(common) + if (codes.some((code) => CONTENT_POLICY_CODES.has(code)) || texts.some((text) => CONTENT_POLICY_TEXT.test(text))) + return new ContentPolicyReason(common) if ( codes.some((code) => QUOTA_CODES.has(code)) || (input.status === 429 && texts.some((text) => QUOTA_TEXT.test(text))) @@ -106,12 +121,15 @@ export function classifyProviderFailure(input: ProviderFailure): LLMError["reaso return new QuotaExceededReason(common) if (input.status === 401) return new AuthenticationReason({ ...common, kind: "invalid" }) if (input.status === 403) return new AuthenticationReason({ ...common, kind: "insufficient-permissions" }) - if (codes.includes("authentication_error")) return new AuthenticationReason({ ...common, kind: "invalid" }) - if (codes.includes("permission_error")) - return new AuthenticationReason({ ...common, kind: "insufficient-permissions" }) + if (codes.some((code) => code === "authentication_error" || code === "unauthenticated")) + return new AuthenticationReason({ ...common, kind: "invalid" }) if ( - codes.some((code) => code.includes("rate_limit") || code === "too_many_requests" || code === "throttlingexception") + codes.some( + (code) => code === "accessdeniedexception" || code === "permission_error" || code === "permission_denied", + ) ) + return new AuthenticationReason({ ...common, kind: "insufficient-permissions" }) + if (codes.some((code) => code.includes("rate_limit") || RATE_LIMIT_CODES.has(code))) return new RateLimitReason({ ...common, retryAfterMs: input.retryAfterMs, @@ -123,8 +141,12 @@ export function classifyProviderFailure(input: ProviderFailure): LLMError["reaso retryAfterMs: input.retryAfterMs, rateLimit: input.rateLimit, }) - if (codes.some((code) => INVALID_REQUEST_CODES.has(code))) return new InvalidRequestReason(common) - if (codes.some((code) => SERVER_CODES.has(code) || code.includes("exhausted") || code.includes("unavailable"))) + if (codes.some((code) => INVALID_REQUEST_CODES.has(code)) || texts.some((text) => INVALID_REQUEST_TEXT.test(text))) + return new InvalidRequestReason(common) + if ( + codes.some((code) => SERVER_CODES.has(code) || code.includes("unavailable")) || + texts.some((text) => SERVER_TEXT.test(text)) + ) return new ProviderInternalReason({ ...common, status: input.status, @@ -160,8 +182,15 @@ function providerCodes(value: string) { const error = isRecord(decoded.error) ? decoded.error : undefined const response = isRecord(decoded.response) ? decoded.response : undefined const responseError = isRecord(response?.error) ? response.error : undefined - return [decoded.code, decoded.status, error?.code, error?.type, error?.status, responseError?.code, responseError?.type] - .filter((value): value is string => typeof value === "string") + return [ + decoded.code, + decoded.status, + error?.code, + error?.type, + error?.status, + responseError?.code, + responseError?.type, + ].filter((value): value is string => typeof value === "string") } function isRecord(value: unknown): value is Record { diff --git a/packages/ai/test/executor.test.ts b/packages/ai/test/executor.test.ts index 2a0056896f..46c5480cc2 100644 --- a/packages/ai/test/executor.test.ts +++ b/packages/ai/test/executor.test.ts @@ -121,6 +121,7 @@ describe("RequestExecutor", () => { yield* classify("Request rate increased too quickly") yield* classify('{"type":"error","error":{"type":"too_many_requests"}}') yield* classify('{"type":"error","error":{"code":"rate_limit_exceeded"}}') + yield* classify('{"code":"resource_exhausted"}') }), ) @@ -135,7 +136,6 @@ describe("RequestExecutor", () => { expect(error.reason).toMatchObject({ _tag: "ProviderInternal" }) }).pipe(Effect.provide(responsesLayer([new Response(body, { status: 400 })]))) - yield* classify('{"code":"resource_exhausted"}') yield* classify('{"code":"service_unavailable"}') }), ) diff --git a/packages/ai/test/provider-error.test.ts b/packages/ai/test/provider-error.test.ts index f0920126a5..a4e59c9d6a 100644 --- a/packages/ai/test/provider-error.test.ts +++ b/packages/ai/test/provider-error.test.ts @@ -26,6 +26,14 @@ describe("provider error classification", () => { ) }) + test("does not treat incidental safety text as a content-policy failure", () => { + expect(classifyProviderFailure({ message: "Internal safety check failed", status: 500 })._tag).toBe( + "ProviderInternal", + ) + expect(classifyProviderFailure({ message: "Blocked by safety policy", status: 400 })._tag).toBe("ContentPolicy") + expect(classifyProviderFailure({ message: "Blocked", status: 400, code: "SAFETY" })._tag).toBe("ContentPolicy") + }) + test("classifies V1 plain-text rate limit fallbacks", () => { expect( [ @@ -47,12 +55,48 @@ describe("provider error classification", () => { ).toEqual(["RateLimit", "RateLimit", "RateLimit", "RateLimit"]) }) - test("classifies V1 overloaded provider codes", () => { + test("classifies canonical provider retry codes", () => { expect( ['{"code":"resource_exhausted"}', '{"code":"service_unavailable"}'].map( (message) => classifyProviderFailure({ message })._tag, ), - ).toEqual(["ProviderInternal", "ProviderInternal"]) + ).toEqual(["RateLimit", "ProviderInternal"]) + }) + + test("keeps temporary per-minute quota wording retryable", () => { + expect(classifyProviderFailure({ message: "You exceeded your per-minute quota", status: 429 })._tag).toBe( + "RateLimit", + ) + }) + + test("classifies canonical Google error codes", () => { + expect( + ["UNAUTHENTICATED", "PERMISSION_DENIED", "INVALID_ARGUMENT", "NOT_FOUND", "INTERNAL"].map( + (code) => classifyProviderFailure({ message: "Provider failed", code })._tag, + ), + ).toEqual(["Authentication", "Authentication", "InvalidRequest", "InvalidRequest", "ProviderInternal"]) + }) + + test("classifies stripped Bedrock stream errors from their messages", () => { + expect( + ["Internal server error", "Throttling exception", "Validation error: invalid input"].map( + (message) => classifyProviderFailure({ message })._tag, + ), + ).toEqual(["ProviderInternal", "RateLimit", "InvalidRequest"]) + }) + + test("classifies documented Bedrock exception codes", () => { + expect( + ["accessDeniedException", "modelTimeoutException", "resourceNotFoundException"].map( + (code) => classifyProviderFailure({ message: "Bedrock failed", code })._tag, + ), + ).toEqual(["Authentication", "ProviderInternal", "InvalidRequest"]) + }) + + test("classifies Anthropic not-found stream errors as invalid requests", () => { + expect(classifyProviderFailure({ message: "Model unavailable", code: "not_found_error" })._tag).toBe( + "InvalidRequest", + ) }) test("classifies nested provider codes when a top-level code is also present", () => { diff --git a/packages/ai/test/provider/bedrock-converse.test.ts b/packages/ai/test/provider/bedrock-converse.test.ts index c7b519d368..2fbfc4eaae 100644 --- a/packages/ai/test/provider/bedrock-converse.test.ts +++ b/packages/ai/test/provider/bedrock-converse.test.ts @@ -6,6 +6,7 @@ import { CacheHint, LLM, Message, ToolCallPart, ToolChoice } from "../../src" import { LLMClient } from "../../src/route" import { AmazonBedrock } from "../../src/providers" import * as BedrockConverse from "../../src/protocols/bedrock-converse" +import { BedrockEventStream } from "../../src/protocols/bedrock-event-stream" import { it } from "../lib/effect" import { fixedResponse } from "../lib/http" import { @@ -34,6 +35,16 @@ const eventFrame = (type: string, payload: object) => body: utf8Encoder.encode(JSON.stringify(payload)), }) +const exceptionFrame = (type: string, payload: object) => + codec.encode({ + headers: { + ":message-type": { type: "string", value: "exception" }, + ":exception-type": { type: "string", value: type }, + ":content-type": { type: "string", value: "application/json" }, + }, + body: utf8Encoder.encode(JSON.stringify(payload)), + }) + const concat = (frames: ReadonlyArray) => { const total = frames.reduce((sum, frame) => sum + frame.length, 0) const out = new Uint8Array(total) @@ -48,6 +59,8 @@ const concat = (frames: ReadonlyArray) => { const eventStreamBody = (...payloads: ReadonlyArray) => concat(payloads.map(([type, payload]) => eventFrame(type, payload))) +const exceptionStreamBody = (type: string, payload: object) => exceptionFrame(type, payload) + // Override the default SSE content-type with the binary event-stream type so // the cassette layer treats the body as bytes when recording. const fixedBytes = (bytes: Uint8Array) => @@ -357,10 +370,10 @@ describe("Bedrock Converse route", () => { it.effect("classifies throttlingException as a rate limit", () => Effect.gen(function* () { - const body = eventStreamBody( - ["messageStart", { role: "assistant" }], - ["throttlingException", { message: "Slow down" }], - ) + const body = concat([ + eventStreamBody(["messageStart", { role: "assistant" }]), + exceptionStreamBody("throttlingException", { message: "Slow down" }), + ]) const error = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body)), Effect.flip) expect(error.reason).toMatchObject({ _tag: "RateLimit", message: "Slow down" }) @@ -371,7 +384,7 @@ describe("Bedrock Converse route", () => { Effect.gen(function* () { const error = yield* LLMClient.generate(baseRequest).pipe( Effect.provide( - fixedBytes(eventStreamBody(["validationException", { message: "Input is too long for requested model" }])), + fixedBytes(exceptionStreamBody("validationException", { message: "Input is too long for requested model" })), ), Effect.flip, ) @@ -384,6 +397,35 @@ describe("Bedrock Converse route", () => { }), ) + it.effect("fails monitored AI SDK bodies on exception wire frames", () => + Effect.gen(function* () { + const response = BedrockEventStream.monitorExceptions( + new Response(exceptionStreamBody("throttlingException", { message: "Slow down" }), { + headers: { "content-type": "application/vnd.amazon.eventstream" }, + }), + ) + const error = yield* Effect.tryPromise({ + try: () => response.arrayBuffer(), + catch: (error) => error, + }).pipe(Effect.flip) + + expect(error).toMatchObject({ code: "throttlingException", message: "Slow down" }) + }), + ) + + it.effect("classifies serviceUnavailableException wire frames as provider failures", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate(baseRequest).pipe( + Effect.provide( + fixedBytes(exceptionStreamBody("serviceUnavailableException", { message: "Service unavailable" })), + ), + Effect.flip, + ) + + expect(error.reason).toMatchObject({ _tag: "ProviderInternal", message: "Service unavailable" }) + }), + ) + it.effect("rejects requests with no auth path", () => Effect.gen(function* () { const unsignedModel = AmazonBedrock.configure({ diff --git a/packages/core/src/aisdk.ts b/packages/core/src/aisdk.ts index 9ddc7fe5b6..15a84924f0 100644 --- a/packages/core/src/aisdk.ts +++ b/packages/core/src/aisdk.ts @@ -47,6 +47,7 @@ import { type UsageInput, } from "@opencode-ai/ai" import { Auth, Endpoint, type AnyRoute } from "@opencode-ai/ai/route" +import { BedrockEventStream } from "@opencode-ai/ai/protocols/bedrock-event-stream" import { Cause, Context, Effect, Layer, Option, Schema, Scope, Stream } from "effect" import { ModelV2 } from "./model" import { ProviderV2 } from "./provider" @@ -172,8 +173,9 @@ function prepareOptions(model: ModelV2.Info, pkg: string) { ...opts, timeout: false, }) - if (!chunkAbortCtl || typeof chunkTimeout !== "number") return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) + const response = pkg === "@ai-sdk/amazon-bedrock" ? BedrockEventStream.monitorExceptions(res) : res + if (!chunkAbortCtl || typeof chunkTimeout !== "number") return response + return wrapSSE(response, chunkTimeout, chunkAbortCtl) } return options @@ -739,10 +741,12 @@ function llmError(method: string, error: unknown) { const cause = error instanceof Error ? error.cause : undefined const failures = [error, cause] const code = failures.map(machineCode).find((value) => value !== undefined) + const providerCode = apiFailureCode(error)?.toLowerCase() const reason = (() => { if ( error instanceof ChunkTimeoutError || failures.some((failure) => failure instanceof Error && failure.name === "TimeoutError") || + providerCode === "timeout_error" || (code !== undefined && TRANSPORT_TIMEOUT_CODES.has(code)) ) return new TransportReason({ message: errorMessage(error), kind: code ?? "Timeout" }) @@ -751,8 +755,7 @@ function llmError(method: string, error: unknown) { return new TransportReason({ message: errorMessage(error), kind: code }) const malformed = failures.find(isMalformedError) if (malformed) return new InvalidProviderOutputReason({ message: malformed.message }) - if (LoadAPIKeyError.isInstance(error)) - return new AuthenticationReason({ message: error.message, kind: "missing" }) + if (LoadAPIKeyError.isInstance(error)) return new AuthenticationReason({ message: error.message, kind: "missing" }) if (NoSuchModelError.isInstance(error)) return new InvalidRequestReason({ message: error.message }) if ( LoadSettingError.isInstance(error) || @@ -780,12 +783,13 @@ function apiCallReason(error: APICallError) { const malformed = isMalformedError(error.cause) ? error.cause : undefined if (error.statusCode !== undefined && error.statusCode < 400 && malformed) return new InvalidProviderOutputReason({ message: malformed.message }) - const code = apiFailureCode(error.data) ?? apiFailureCode(error.responseBody) + const evidence = apiFailureEvidence(error.responseBody) + const code = apiFailureCode(error.data) ?? apiFailureCode(evidence) if (error.statusCode === undefined) { const reason = - code === undefined && error.responseBody === undefined + code === undefined && evidence === undefined ? undefined - : classifyProviderFailure({ message: error.message, code, evidence: error.responseBody }) + : classifyProviderFailure({ message: error.message, code, evidence }) if (reason && reason._tag !== "UnknownProvider") return reason if (error.isRetryable) return new TransportReason({ message: error.message }) return reason ?? new UnknownProviderReason({ message: error.message }) @@ -793,16 +797,13 @@ function apiCallReason(error: APICallError) { const retryAfter = retryAfterMs(error.responseHeaders) const reason = classifyProviderFailure({ message: error.message, - evidence: error.responseBody, + evidence, status: error.statusCode, code, retryAfterMs: retryAfter, }) if (!error.isRetryable || (reason._tag !== "UnknownProvider" && reason._tag !== "InvalidRequest")) return reason - if ( - classifyProviderFailure({ message: error.message, evidence: error.responseBody, code })._tag !== "UnknownProvider" - ) - return reason + if (classifyProviderFailure({ message: error.message, evidence, code })._tag !== "UnknownProvider") return reason return new ProviderInternalReason({ message: error.message, status: error.statusCode, retryAfterMs: retryAfter }) } @@ -825,6 +826,7 @@ const TRANSPORT_CONNECTION_CODES = new Set([ "FAILEDTOOPENSOCKET", "UND_ERR_SOCKET", ]) +const FAILURE_EVIDENCE_LIMIT = 65_536 function field(error: unknown, name: string) { return typeof error === "object" && error !== null ? Reflect.get(error, name) : undefined @@ -843,6 +845,8 @@ function apiFailureCode(error: unknown): string | undefined { const nested = field(error, "error") ?? field(field(error, "response"), "error") const nestedCode = nested === undefined || nested === error ? undefined : apiFailureCode(nested) if (nestedCode) return nestedCode + if (typeof field(error, "originalMessage") === "string" && typeof field(error, "originalStatusCode") === "number") + return "modelstreamerrorexception" const code = field(error, "code") if (typeof code === "string" || typeof code === "number") return String(code) const status = field(error, "status") @@ -854,14 +858,18 @@ function apiFailureCode(error: unknown): string | undefined { function apiFailureMessage(error: unknown): string { const message = field(error, "message") if (typeof message === "string") return message + const originalMessage = field(error, "originalMessage") + if (typeof originalMessage === "string") return originalMessage const nested = field(error, "error") ?? field(field(error, "response"), "error") return nested === undefined || nested === error ? String(error) : apiFailureMessage(nested) } function apiFailureStatus(error: unknown): number | undefined { const status = field(error, "status") + const statusCode = field(error, "statusCode") + const originalStatusCode = field(error, "originalStatusCode") const code = field(error, "code") - const value = [status, code] + const value = [status, statusCode, originalStatusCode, code] .map((value) => (typeof value === "number" ? value : typeof value === "string" ? Number(value) : undefined)) .find((value) => value !== undefined && Number.isInteger(value) && value >= 400 && value < 600) if (value !== undefined) return value @@ -869,6 +877,10 @@ function apiFailureStatus(error: unknown): number | undefined { return nested === undefined || nested === error ? undefined : apiFailureStatus(nested) } +function apiFailureEvidence(error: string | undefined) { + return error === undefined ? undefined : error.slice(0, FAILURE_EVIDENCE_LIMIT) +} + function retryAfterMs(headers: Record | undefined) { if (!headers) return undefined const normalized = Object.fromEntries(Object.entries(headers).map(([name, value]) => [name.toLowerCase(), value])) diff --git a/packages/core/test/aisdk.test.ts b/packages/core/test/aisdk.test.ts index 53647a2b38..b6bf0c84d9 100644 --- a/packages/core/test/aisdk.test.ts +++ b/packages/core/test/aisdk.test.ts @@ -492,12 +492,53 @@ it.effect("classifies structured AI SDK stream errors", () => }), ) -it.effect("classifies structured stream messages without codes", () => +it.effect("classifies AI Gateway errors with statusCode", () => Effect.gen(function* () { const error = yield* streamFailure( - streamingLanguage({ type: "error", error: { message: "Rate limit exceeded" } }), + streamingLanguage({ + type: "error", + error: { type: "internal_server_error", message: "Gateway failed", statusCode: 503 }, + }), ) + expect(error).toMatchObject({ reason: { _tag: "ProviderInternal", status: 503 } }) + }), +) + +it.effect("classifies AI Gateway timeouts as retryable", () => + Effect.gen(function* () { + const error = yield* streamFailure( + streamingLanguage({ + type: "error", + error: { type: "timeout_error", message: "Gateway timed out", statusCode: 408 }, + }), + ) + + expect(error).toMatchObject({ reason: { _tag: "Transport", kind: "Timeout" } }) + }), +) + +it.effect("classifies stripped Bedrock model stream errors", () => + Effect.gen(function* () { + const error = yield* streamFailure( + streamingLanguage({ + type: "error", + error: { + message: "The model stream failed", + originalMessage: "Upstream provider failed", + originalStatusCode: 424, + }, + }), + ) + + expect(error).toMatchObject({ reason: { _tag: "ProviderInternal", status: 424 } }) + }), +) + +it.effect("classifies structured stream messages without codes", () => + Effect.gen(function* () { + const error = yield* streamFailure(streamingLanguage({ type: "error", error: { message: "Rate limit exceeded" } })) + expect(error).toMatchObject({ reason: { _tag: "RateLimit" } }) }), ) @@ -528,11 +569,81 @@ it.effect("classifies numeric string stream statuses", () => }), ) +it.effect("classifies readable stream failures", () => + Effect.gen(function* () { + const error = yield* streamFailure({ + ...failingLanguage(new Error("unused")), + doStream: async () => ({ + stream: new ReadableStream({ + start(controller) { + controller.error(Object.assign(new Error("connection reset"), { code: "ECONNRESET" })) + }, + }), + request: { body: {} }, + }), + }) + + expect(error).toMatchObject({ method: "readStream", reason: { _tag: "Transport", kind: "ECONNRESET" } }) + }), +) + +it.live("times out stalled SSE chunks", () => + Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + port: 0, + fetch: () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(": connected\n\n")) + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ), + }), + ), + (server) => + Effect.gen(function* () { + const aisdk = yield* AISDK.Service + let wrappedFetch: typeof fetch | undefined + yield* aisdk.hook.sdk((event) => { + wrappedFetch = event.options.fetch + event.sdk = {} + }) + yield* aisdk.hook.language((event) => { + event.language = { + ...failingLanguage(new Error("unused")), + doStream: async () => { + const fetcher = wrappedFetch + if (!fetcher) throw new Error("AI SDK fetch was not configured") + const response = await fetcher(server.url, { method: "POST" }) + if (!response.body) throw new Error("AI SDK response body was missing") + return { + stream: response.body.pipeThrough( + new TransformStream({ transform() {} }), + ), + request: { body: {} }, + } + }, + } + }) + const resolved = yield* aisdk.model(model("@ai-sdk/openai", { chunkTimeout: 10 })) + const request = LLM.request({ model: resolved, prompt: "Hello" }) + const prepared = yield* LLMClient.prepare(request) + const error = yield* resolved.route + .streamPrepared(prepared.body, request, { http: { execute: () => Effect.die("unused") } }) + .pipe(Stream.runDrain, Effect.flip) + + expect(error).toMatchObject({ method: "readStream", reason: { _tag: "Transport", kind: "Timeout" } }) + }), + (server) => Effect.promise(() => server.stop(true)), + ), +) + it.effect("classifies missing AI SDK API keys", () => Effect.gen(function* () { - const error = yield* streamFailure( - failingLanguage(new LoadAPIKeyError({ message: "API key is missing" })), - ) + const error = yield* streamFailure(failingLanguage(new LoadAPIKeyError({ message: "API key is missing" }))) expect(error).toMatchObject({ reason: { _tag: "Authentication", kind: "missing" } }) }),