mirror of
https://gitee.com/zhangsken/opencode.git
synced 2026-10-16 10:30:01 +08:00
fix(ai): classify terminal generation failures by provider error code (#51632)
This commit is contained in:
Shoubhit Dash
@@ -98,7 +98,7 @@ When a provider supports multiple physical transports, selection remains executi
|
||||
|
||||
Media does not fit the SSE-frames-to-event-state-machine LLM route. `MediaRoute.inline(...)` / `queued(...)` / `stream(...)` (`src/route/media.ts`) compose a `MediaProtocol` kind with `Endpoint` and `Auth` and own the transport plumbing: `http` option merging, URL/query rendering, auth headers, JSON vs multipart encoding, and handing the response back to the protocol. `MediaProtocol.inline` (`src/route/media-protocol.ts`) is `body.from(request)` plus `response.decode(response, context)`; each protocol declares `const route = MediaProtocol.identity({ id, name, provider })` once and decodes through `route.decodeJson` / `route.text` / `route.decodeStarted` so decode failures retain the raw body and HTTP context, raising `route.unsupported(operation, message)` for requests it cannot lower, and passes `route` as the first argument to `MediaProtocol.inline` / `queued` / `stream`. `Generation` (`src/generation.ts`) is the provider-neutral handle for a queued generation over a `GenerationRoute` (`status`, `result`, `cancel`). Image protocol files follow the same section order as LLM protocols and declare unsupported common fields once through the protocol's `unsupported` list.
|
||||
|
||||
`MediaProtocol.queued` is the submit-then-poll kind every video route uses: `start` (body + decode into `{ token, snapshot }`), `status`, `result`, and optional `cancel` (with `activeOnly` when the provider's cancel endpoint deletes finished work, as Runway's does: the route refreshes status first and skips terminal generations), each addressed by a route-owned `token` whose `Schema.Codec` makes it serializable. `MediaRoute.inline` and `MediaRoute.queued` compose the two kinds with `Endpoint` and `Auth`; the queued route decodes the token once at the boundary (`start` output or `resume` input) and closes over it in a token-free `GenerationRoute` (`status`/`result`/`cancel` are plain Effects), so `Generation` never sees the token's shape and only carries the encoded JSON for persistence. Polls reuse the route's auth and deployment headers plus the request's `http` overlay after `start`, and resolve relative paths against the route base URL (provider-issued absolute URLs such as fal's `status_url` pass through). `result` is always its own GET even when the provider returns output inside the status document, so `Generation.await` behaves the same after `start` and after `resume`. `PollContext.auth` carries only what `Auth` added or changed so protocols can hand download credentials to output assets as transient `Media.Asset.headers` (Veo) — never part of `source` or JSON. Status strings map through a per-protocol `STATUS` table via `MediaProtocol.status`; terminal generations without output fail through `output.ended` / `output.contentPolicy` with the provider document on `reason.body`. `GenerationAwaitOptions` (`AwaitOptions` in `src/generation.ts`, `{ poll?: Poll }`) is the one options type for `await`, `events`, `Video.generate`, and `Video.stream`.
|
||||
`MediaProtocol.queued` is the submit-then-poll kind every video route uses: `start` (body + decode into `{ token, snapshot }`), `status`, `result`, and optional `cancel` (with `activeOnly` when the provider's cancel endpoint deletes finished work, as Runway's does: the route refreshes status first and skips terminal generations), each addressed by a route-owned `token` whose `Schema.Codec` makes it serializable. `MediaRoute.inline` and `MediaRoute.queued` compose the two kinds with `Endpoint` and `Auth`; the queued route decodes the token once at the boundary (`start` output or `resume` input) and closes over it in a token-free `GenerationRoute` (`status`/`result`/`cancel` are plain Effects), so `Generation` never sees the token's shape and only carries the encoded JSON for persistence. Polls reuse the route's auth and deployment headers plus the request's `http` overlay after `start`, and resolve relative paths against the route base URL (provider-issued absolute URLs such as fal's `status_url` pass through). `result` is always its own GET even when the provider returns output inside the status document, so `Generation.await` behaves the same after `start` and after `resume`. `PollContext.auth` carries only what `Auth` added or changed so protocols can hand download credentials to output assets as transient `Media.Asset.headers` (Veo) — never part of `source` or JSON. Status strings map through a per-protocol `STATUS` table via `MediaProtocol.status`; terminal generations without output fail through `output.ended` / `output.contentPolicy` with the provider document on `reason.body`; a `failed` generation maps the provider's error code through a per-protocol `FAILURE` table via `MediaProtocol.failure` so rejected inputs are not reported as retryable `ProviderInternal`. `GenerationAwaitOptions` (`AwaitOptions` in `src/generation.ts`, `{ poll?: Poll }`) is the one options type for `await`, `events`, `Video.generate`, and `Video.stream`.
|
||||
|
||||
`MediaProtocol.stream` is the incremental kind every speech route uses, with the same discipline as LLM protocols. `MediaRoute.stream` submits the caller's request as `MediaProtocol.Addressed<Request>` (`{ ...request, mode }`, `mode: "generate" | "stream"`), so one provider stays one protocol: `body.from`, the endpoint path, and `frames` read `request.mode` to pick the body, path, and framing. `frames(bytes, context)` returns frames — `Framing.sse`, `Framing.lines`, `Framing.document` (a single-document response shaped like a streamed record), or the raw `bytes` for chunked audio. `initial()` is fresh per-response parser state; `step` folds each frame into it and emits modality events; `finish(state, context)` runs once after the last frame with the request, body, and observed `http` (header-only usage lives there) and emits exactly one terminal event or fails with `route.incomplete()`. Keep parser state to real accumulators and derive anything the request or body determines in `finish`. `generate` runs the same stream and folds it with the modality's `collect`. Request-derived URL parameters go on the body's `query` (array values repeat the parameter), applied before route and caller `http.query`. Decode frames with `route.decodeFrame` and raise stream-time failures with `route.frameError` (the frame stays on `reason.body`); protocols never thread HTTP context, because the route fills `reason.http` on stream errors that lack it. Speech protocols share `protocols/utils/speech-stream.ts` for deltas, timestamps, voice ids, PCM and container descriptions, and the terminal asset.
|
||||
|
||||
|
||||
@@ -36,7 +36,9 @@ const StartResponse = Schema.Struct({ name: Schema.String })
|
||||
|
||||
const Operation = Schema.Struct({
|
||||
done: Schema.optional(Schema.Boolean),
|
||||
error: Schema.optional(Schema.Struct({ message: Schema.optional(Schema.String) })),
|
||||
error: Schema.optional(
|
||||
Schema.Struct({ code: Schema.optional(Schema.Number), message: Schema.optional(Schema.String) }),
|
||||
),
|
||||
response: Schema.optional(
|
||||
Schema.Struct({
|
||||
generateVideoResponse: Schema.optional(
|
||||
@@ -60,6 +62,16 @@ const Operation = Schema.Struct({
|
||||
metadata: Schema.optional(Schema.Unknown),
|
||||
})
|
||||
|
||||
// Operation errors are `google.rpc.Status`; unlisted codes (INTERNAL, UNAVAILABLE, ...) are provider-side.
|
||||
const FAILURE = {
|
||||
3: "InvalidRequest", // INVALID_ARGUMENT
|
||||
7: "Authentication", // PERMISSION_DENIED
|
||||
8: "RateLimit", // RESOURCE_EXHAUSTED
|
||||
9: "InvalidRequest", // FAILED_PRECONDITION
|
||||
11: "InvalidRequest", // OUT_OF_RANGE
|
||||
16: "Authentication", // UNAUTHENTICATED
|
||||
} as const satisfies Record<number, MediaProtocol.Failure>
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 5. Request body construction
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -154,6 +166,7 @@ const decodeResult = Effect.fn("GoogleVideo.decodeResult")(function* (
|
||||
return yield* output.ended(
|
||||
"failed",
|
||||
`${route.name} operation failed${operation.error?.message === undefined ? "" : `: ${operation.error.message}`}`,
|
||||
MediaProtocol.failure(FAILURE, operation.error?.code),
|
||||
)
|
||||
const generated = operation.response?.generateVideoResponse
|
||||
// Downloads require the same API key as the poll; the asset carries it transiently and follows the redirect.
|
||||
|
||||
@@ -137,7 +137,12 @@ const decodeResult = Effect.fn("RunwayVideo.decodeResult")(function* (
|
||||
const message = `${route.name} task failed${code === undefined ? "" : ` (${code})`}${task.failure ? `: ${task.failure}` : ""}`
|
||||
// Runway failure codes are dotted paths; every moderation outcome carries a SAFETY segment.
|
||||
if (code !== undefined && /(^|\.)SAFETY(\.|$)/.test(code)) return yield* output.contentPolicy(message)
|
||||
return yield* output.ended("failed", message)
|
||||
// ASSET.INVALID rejects the caller's input media; Runway documents it as not retryable.
|
||||
return yield* output.ended(
|
||||
"failed",
|
||||
message,
|
||||
code !== undefined && /^ASSET\.INVALID(\.|$)/.test(code) ? "InvalidRequest" : "ProviderInternal",
|
||||
)
|
||||
}
|
||||
if (status === "cancelled")
|
||||
return yield* output.ended("cancelled", `${route.name} task ${context.token.taskID} was cancelled`)
|
||||
|
||||
@@ -65,6 +65,13 @@ const STATUS = {
|
||||
expired: "expired",
|
||||
} as const satisfies Record<string, Status>
|
||||
|
||||
// Documented video error codes; `service_unavailable`, `internal_error`, and unknown codes are provider-side.
|
||||
const FAILURE = {
|
||||
invalid_argument: "InvalidRequest",
|
||||
failed_precondition: "InvalidRequest",
|
||||
permission_denied: "Authentication",
|
||||
} as const satisfies Record<string, MediaProtocol.Failure>
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 5. Request body construction
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -143,6 +150,7 @@ const decodeResult = Effect.fn("XAIVideo.decodeResult")(function* (
|
||||
return yield* output.ended(
|
||||
"failed",
|
||||
`${route.name} generation failed${code === undefined ? "" : ` (${code})`}${message === undefined ? "" : `: ${message}`}`,
|
||||
MediaProtocol.failure(FAILURE, code),
|
||||
)
|
||||
}
|
||||
if (status !== "completed")
|
||||
|
||||
@@ -5,12 +5,14 @@ import { Media } from "../media.js"
|
||||
import type { AuthInput } from "./auth.js"
|
||||
import {
|
||||
AIError,
|
||||
AuthenticationError,
|
||||
ContentPolicyError,
|
||||
HttpContext,
|
||||
InvalidProviderOutputError,
|
||||
InvalidRequestError,
|
||||
ProviderID,
|
||||
ProviderInternalError,
|
||||
RateLimitError,
|
||||
UnsupportedOperationError,
|
||||
} from "../schema/index.js"
|
||||
|
||||
@@ -188,6 +190,16 @@ export const stream = <Request, Event, Frame, State>(
|
||||
// Response helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Reasons a provider can report for a `failed` generation; anything it does not classify is `ProviderInternal`. */
|
||||
const FAILURES = {
|
||||
InvalidRequest: InvalidRequestError,
|
||||
Authentication: AuthenticationError,
|
||||
RateLimit: RateLimitError,
|
||||
ProviderInternal: ProviderInternalError,
|
||||
}
|
||||
|
||||
export type Failure = keyof typeof FAILURES
|
||||
|
||||
const context = (response: HttpClientResponse.HttpClientResponse) =>
|
||||
new HttpContext({ url: response.request.url, status: response.status, headers: response.headers })
|
||||
|
||||
@@ -199,9 +211,10 @@ export const identity = (input: { readonly id: string; readonly name: string; re
|
||||
|
||||
/**
|
||||
* Read a text body while retaining the original payload and HTTP context on every downstream error. `invalid` is a
|
||||
* malformed provider document; `ended` is a generation that reached a terminal status without output (`failed` is
|
||||
* provider-side, `cancelled`/`expired` mean the result will never exist); `pending` is a `result()` read before the
|
||||
* generation finished, which is caller misuse; `contentPolicy` is a moderated result.
|
||||
* malformed provider document; `ended` is a generation that reached a terminal status without output (`failed`
|
||||
* carries the provider's classification, defaulting to `ProviderInternal`; `cancelled`/`expired` mean the result
|
||||
* will never exist); `pending` is a `result()` read before the generation finished, which is caller misuse;
|
||||
* `contentPolicy` is a moderated result.
|
||||
*/
|
||||
const text = Effect.fn("MediaProtocol.text")(function* (response: HttpClientResponse.HttpClientResponse) {
|
||||
const http = context(response)
|
||||
@@ -223,11 +236,15 @@ export const identity = (input: { readonly id: string; readonly name: string; re
|
||||
http,
|
||||
invalid: (message: string, cause?: unknown) =>
|
||||
new AIError({ reason: new InvalidProviderOutputError({ route: input.id, message, body, http, cause }) }),
|
||||
ended: (status: Exclude<Status, "queued" | "running" | "completed">, message: string) =>
|
||||
ended: (
|
||||
status: Exclude<Status, "queued" | "running" | "completed">,
|
||||
message: string,
|
||||
failure: Failure = "ProviderInternal",
|
||||
) =>
|
||||
new AIError({
|
||||
reason:
|
||||
status === "failed"
|
||||
? new ProviderInternalError({ message, body, http })
|
||||
? new FAILURES[failure]({ message, body, http })
|
||||
: new InvalidRequestError({ message, body, http }),
|
||||
}),
|
||||
pending: (id: string) =>
|
||||
@@ -303,6 +320,10 @@ export const status = <Table extends Record<string, Status>>(
|
||||
return Effect.succeed(table[raw])
|
||||
}
|
||||
|
||||
/** Map a provider error code through the protocol's table; missing or unmapped codes are `ProviderInternal`. */
|
||||
export const failure = (table: Readonly<Record<string, Failure>>, code: string | number | undefined): Failure =>
|
||||
code !== undefined && Object.hasOwn(table, code) ? table[code] : "ProviderInternal"
|
||||
|
||||
/** A `url` asset whose provider-declared retention window starts now. */
|
||||
export const expiringUrl = (url: string, retention: Duration.Duration, options?: Parameters<typeof Media.url>[1]) =>
|
||||
Clock.currentTimeMillis.pipe(
|
||||
|
||||
@@ -162,27 +162,39 @@ describe("Video / Google Veo", () => {
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("surfaces an operation error as a failed generation with the provider body", () =>
|
||||
Effect.gen(function* () {
|
||||
const failure = {
|
||||
name: operation,
|
||||
done: true,
|
||||
error: { code: 3, message: "Prompt violates policy", status: "INVALID_ARGUMENT" },
|
||||
}
|
||||
const error = yield* Video.generate({ model, prompt: "nope" }).pipe(
|
||||
Effect.flip,
|
||||
Effect.provide(
|
||||
layer((input) =>
|
||||
Effect.succeed(input.request.method === "POST" ? json(input, { name: operation }) : json(input, failure)),
|
||||
),
|
||||
),
|
||||
)
|
||||
expect(error.reason._tag).toBe("ProviderInternal")
|
||||
expect(error.message).toBe("Google Veo operation failed: Prompt violates policy")
|
||||
expect(error.reason.body).toBe(JSON.stringify(failure))
|
||||
expect(error.reason.http?.status).toBe(200)
|
||||
}),
|
||||
)
|
||||
for (const terminal of [
|
||||
{ error: { code: 3, message: "Prompt violates policy", status: "INVALID_ARGUMENT" }, tag: "InvalidRequest" },
|
||||
{ error: { code: 9, message: "Unsupported resolution", status: "FAILED_PRECONDITION" }, tag: "InvalidRequest" },
|
||||
{ error: { code: 11, message: "Duration out of range", status: "OUT_OF_RANGE" }, tag: "InvalidRequest" },
|
||||
{ error: { code: 7, message: "Permission denied", status: "PERMISSION_DENIED" }, tag: "Authentication" },
|
||||
{ error: { code: 16, message: "Invalid credentials", status: "UNAUTHENTICATED" }, tag: "Authentication" },
|
||||
{ error: { code: 8, message: "Quota exceeded", status: "RESOURCE_EXHAUSTED" }, tag: "RateLimit" },
|
||||
{ error: { code: 13, message: "Internal error", status: "INTERNAL" }, tag: "ProviderInternal" },
|
||||
{ error: { code: 14, message: "Service unavailable", status: "UNAVAILABLE" }, tag: "ProviderInternal" },
|
||||
{ error: { message: "Something broke" }, tag: "ProviderInternal" },
|
||||
]) {
|
||||
it.effect(
|
||||
`surfaces ${terminal.error.status ?? "an uncoded"} operation error as ${terminal.tag} with the provider body`,
|
||||
() =>
|
||||
Effect.gen(function* () {
|
||||
const failure = { name: operation, done: true, error: terminal.error }
|
||||
const error = yield* Video.generate({ model, prompt: "nope" }).pipe(
|
||||
Effect.flip,
|
||||
Effect.provide(
|
||||
layer((input) =>
|
||||
Effect.succeed(
|
||||
input.request.method === "POST" ? json(input, { name: operation }) : json(input, failure),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
expect(error.reason._tag).toBe(terminal.tag)
|
||||
expect(error.message).toBe(`Google Veo operation failed: ${terminal.error.message}`)
|
||||
expect(error.reason.body).toBe(JSON.stringify(failure))
|
||||
expect(error.reason.http?.status).toBe(200)
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
it.effect("reports fully filtered output as a content policy failure", () =>
|
||||
Effect.gen(function* () {
|
||||
@@ -332,12 +344,37 @@ describe("Video / xAI", () => {
|
||||
for (const terminal of [
|
||||
{
|
||||
body: { status: "failed", error: { code: "invalid_argument", message: "Prompt cannot be empty." } },
|
||||
tag: "ProviderInternal",
|
||||
tag: "InvalidRequest",
|
||||
message: "xAI Video generation failed (invalid_argument): Prompt cannot be empty.",
|
||||
},
|
||||
{
|
||||
body: { status: "failed", error: { code: "failed_precondition", message: "Extension is not supported." } },
|
||||
tag: "InvalidRequest",
|
||||
message: "xAI Video generation failed (failed_precondition): Extension is not supported.",
|
||||
},
|
||||
{
|
||||
body: { status: "failed", error: { code: "permission_denied", message: "Team lacks access." } },
|
||||
tag: "Authentication",
|
||||
message: "xAI Video generation failed (permission_denied): Team lacks access.",
|
||||
},
|
||||
{
|
||||
body: { status: "failed", error: { code: "service_unavailable", message: "Overloaded." } },
|
||||
tag: "ProviderInternal",
|
||||
message: "xAI Video generation failed (service_unavailable): Overloaded.",
|
||||
},
|
||||
{
|
||||
body: { status: "failed", error: { code: "internal_error", message: "Generation failed." } },
|
||||
tag: "ProviderInternal",
|
||||
message: "xAI Video generation failed (internal_error): Generation failed.",
|
||||
},
|
||||
{
|
||||
body: { status: "failed", error: { code: "constructor", message: "Future code." } },
|
||||
tag: "ProviderInternal",
|
||||
message: "xAI Video generation failed (constructor): Future code.",
|
||||
},
|
||||
{ body: { status: "expired" }, tag: "InvalidRequest", message: "xAI Video request req_1 expired" },
|
||||
]) {
|
||||
it.effect(`surfaces ${terminal.body.status} generations with the provider body`, () =>
|
||||
it.effect(`surfaces ${terminal.body.error?.code ?? terminal.body.status} generations with the provider body`, () =>
|
||||
Effect.gen(function* () {
|
||||
const error = yield* Video.generate({ model, prompt: "x" }).pipe(Effect.flip)
|
||||
expect(error.reason._tag).toBe(terminal.tag)
|
||||
@@ -765,6 +802,11 @@ describe("Video / Runway", () => {
|
||||
tag: "ProviderInternal",
|
||||
message: "Runway task failed (INTERNAL.BAD_OUTPUT.CODE01): Something broke",
|
||||
},
|
||||
{
|
||||
body: { status: "FAILED", failure: "Unsupported dimensions", failureCode: "ASSET.INVALID" },
|
||||
tag: "InvalidRequest",
|
||||
message: "Runway task failed (ASSET.INVALID): Unsupported dimensions",
|
||||
},
|
||||
{ body: { status: "CANCELLED" }, tag: "InvalidRequest", message: "Runway task task_1 was cancelled" },
|
||||
]) {
|
||||
it.effect(`surfaces ${terminal.body.failureCode ?? terminal.body.status} with the task body`, () =>
|
||||
|
||||
Reference in New Issue
Block a user