From 4428a77acd7e70a209b9fda367678e3eb90348f6 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Sun, 27 Sep 2026 18:44:21 +0530 Subject: [PATCH] fix(ai): classify terminal generation failures by provider error code (#51632) --- packages/ai/AGENTS.md | 2 +- packages/ai/src/protocols/google-video.ts | 15 +++- packages/ai/src/protocols/runway-video.ts | 7 +- packages/ai/src/protocols/xai-video.ts | 8 +++ packages/ai/src/route/media-protocol.ts | 31 ++++++-- packages/ai/test/video.test.ts | 88 +++++++++++++++++------ 6 files changed, 120 insertions(+), 31 deletions(-) diff --git a/packages/ai/AGENTS.md b/packages/ai/AGENTS.md index dde4bb0ebaa..49e3ea0f046 100644 --- a/packages/ai/AGENTS.md +++ b/packages/ai/AGENTS.md @@ -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, 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. diff --git a/packages/ai/src/protocols/google-video.ts b/packages/ai/src/protocols/google-video.ts index a00c4609027..c3ff3a621fe 100644 --- a/packages/ai/src/protocols/google-video.ts +++ b/packages/ai/src/protocols/google-video.ts @@ -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 + // --------------------------------------------------------------------------- // 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. diff --git a/packages/ai/src/protocols/runway-video.ts b/packages/ai/src/protocols/runway-video.ts index c9159c15b3c..bc3a68f064f 100644 --- a/packages/ai/src/protocols/runway-video.ts +++ b/packages/ai/src/protocols/runway-video.ts @@ -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`) diff --git a/packages/ai/src/protocols/xai-video.ts b/packages/ai/src/protocols/xai-video.ts index d9e9f2d9edd..45d1eac0b65 100644 --- a/packages/ai/src/protocols/xai-video.ts +++ b/packages/ai/src/protocols/xai-video.ts @@ -65,6 +65,13 @@ const STATUS = { expired: "expired", } as const satisfies Record +// 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 + // --------------------------------------------------------------------------- // 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") diff --git a/packages/ai/src/route/media-protocol.ts b/packages/ai/src/route/media-protocol.ts index 89fb70013e5..27b8d6de5b2 100644 --- a/packages/ai/src/route/media-protocol.ts +++ b/packages/ai/src/route/media-protocol.ts @@ -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 = ( // 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, message: string) => + ended: ( + status: Exclude, + 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 = >( 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>, 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[1]) => Clock.currentTimeMillis.pipe( diff --git a/packages/ai/test/video.test.ts b/packages/ai/test/video.test.ts index 64d16d0e78a..dc21cc6030a 100644 --- a/packages/ai/test/video.test.ts +++ b/packages/ai/test/video.test.ts @@ -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`, () =>