ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(ai): handle remaining provider failures

Browse source
This commit is contained in:
Aiden Cline 2026-07-17 14:41:38 -05:00
commit 0eacf8d736
8 changed files with 332 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/ai/src/index.ts
View file

@ -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,

59
packages/ai/src/protocols/bedrock-event-stream.ts
View file

@ -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<object> => ({
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<Uint8Array, Uint8Array>({
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"

51
packages/ai/src/provider-error.ts
View file

@ -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<string, unknown> {

2
packages/ai/test/executor.test.ts
View file

@ -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"}')
}),
)

48
packages/ai/test/provider-error.test.ts
View file

@ -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", () => {

52
packages/ai/test/provider/bedrock-converse.test.ts
View file

@ -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<Uint8Array>) => {
const total = frames.reduce((sum, frame) => sum + frame.length, 0)
const out = new Uint8Array(total)
@ -48,6 +59,8 @@ const concat = (frames: ReadonlyArray<Uint8Array>) => {
const eventStreamBody = (...payloads: ReadonlyArray<readonly [string, object]>) =>
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({