diff --git a/packages/ai/src/protocols/assemblyai-transcription.ts b/packages/ai/src/protocols/assemblyai-transcription.ts index 423868c15d7..46c07413225 100644 --- a/packages/ai/src/protocols/assemblyai-transcription.ts +++ b/packages/ai/src/protocols/assemblyai-transcription.ts @@ -158,8 +158,7 @@ const decodeResult = Effect.fn("AssemblyAITranscription.decodeResult")(function* const error = transcript.error ?? undefined if (status === "failed") return yield* output.ended("failed", `${route.name} transcription failed${error === undefined ? "" : `: ${error}`}`) - if (status !== "completed") - return yield* output.invalid(`${route.name} transcript ${context.token.transcriptID} has not finished`) + if (status !== "completed") return yield* output.pending(context.token.transcriptID) const duration = transcript.audio_duration ?? undefined return new TranscriptionResponse({ text: transcript.text ?? "", diff --git a/packages/ai/src/protocols/bfl-images.ts b/packages/ai/src/protocols/bfl-images.ts index 37518f5f1bd..b5f7b532bf0 100644 --- a/packages/ai/src/protocols/bfl-images.ts +++ b/packages/ai/src/protocols/bfl-images.ts @@ -181,7 +181,8 @@ const decodeResult = Effect.fn("BlackForestLabsImages.decodeResult")(function* ( if (isModerated(document.status)) return yield* output.contentPolicy(`${route.name} moderated the generation`) if (status === "failed" || status === "expired") return yield* output.ended(status, `${route.name} generation ${context.token.id} ended with ${document.status}`) - if (status !== "completed" || document.result === undefined || document.result === null) + if (status !== "completed") return yield* output.pending(context.token.id) + if (document.result === undefined || document.result === null) return yield* output.invalid(`${route.name} generation ${context.token.id} has no result`) const { sample, seed, prompt, ...rest } = document.result // A settled `cost` on the result supersedes the submit-time cost carried on the token. diff --git a/packages/ai/src/protocols/google-images.ts b/packages/ai/src/protocols/google-images.ts index 1c93fbf6815..b8f7ff1584c 100644 --- a/packages/ai/src/protocols/google-images.ts +++ b/packages/ai/src/protocols/google-images.ts @@ -101,7 +101,7 @@ const generationConfig = (request: Request) => { const fromRequest = Effect.fn("GoogleImages.fromRequest")(function* (request: Request) { if (request.n !== undefined && request.n > 1) return yield* route.unsupported( - "image.n", + "media.n", `${route.name} generates one image per request; call it once per image instead of n=${request.n}`, ) const parts = yield* Effect.forEach(request.images ?? [], (image) => diff --git a/packages/ai/src/protocols/google-video.ts b/packages/ai/src/protocols/google-video.ts index b877d472c75..a00c4609027 100644 --- a/packages/ai/src/protocols/google-video.ts +++ b/packages/ai/src/protocols/google-video.ts @@ -149,8 +149,7 @@ const decodeResult = Effect.fn("GoogleVideo.decodeResult")(function* ( const output = yield* decodeOperation(response) const operation = output.value const status = statusOf(operation) - if (status === "running") - return yield* output.invalid(`${route.name} operation ${context.token.operation} has not finished`) + if (status === "running") return yield* output.pending(context.token.operation) if (status === "failed") return yield* output.ended( "failed", diff --git a/packages/ai/src/protocols/replicate-images.ts b/packages/ai/src/protocols/replicate-images.ts index 6204267c9e0..087bd57df9e 100644 --- a/packages/ai/src/protocols/replicate-images.ts +++ b/packages/ai/src/protocols/replicate-images.ts @@ -132,8 +132,7 @@ const decodeResult = Effect.fn("ReplicateImages.decodeResult")(function* ( status, `${route.name} prediction ${context.token.id} ${prediction.status}${typeof prediction.error === "string" ? `: ${prediction.error}` : ""}`, ) - if (status !== "completed") - return yield* output.invalid(`${route.name} prediction ${context.token.id} has not finished`) + if (status !== "completed") return yield* output.pending(context.token.id) if (prediction.data_removed === true) return yield* output.ended("expired", `${route.name} removed the output of prediction ${context.token.id}`) if (!isOutput(prediction.output)) diff --git a/packages/ai/src/protocols/runway-video.ts b/packages/ai/src/protocols/runway-video.ts index e81b7c62bbc..c9159c15b3c 100644 --- a/packages/ai/src/protocols/runway-video.ts +++ b/packages/ai/src/protocols/runway-video.ts @@ -141,8 +141,7 @@ const decodeResult = Effect.fn("RunwayVideo.decodeResult")(function* ( } if (status === "cancelled") return yield* output.ended("cancelled", `${route.name} task ${context.token.taskID} was cancelled`) - if (status !== "completed") - return yield* output.invalid(`${route.name} task ${context.token.taskID} has not finished`) + if (status !== "completed") return yield* output.pending(context.token.taskID) const urls = task.output ?? [] if (urls.length === 0) return yield* output.invalid(`${route.name} task succeeded without any output`) return new VideoResponse({ diff --git a/packages/ai/src/protocols/stability-images.ts b/packages/ai/src/protocols/stability-images.ts index 9d61fa5db65..2dcc538df8d 100644 --- a/packages/ai/src/protocols/stability-images.ts +++ b/packages/ai/src/protocols/stability-images.ts @@ -175,7 +175,7 @@ const decodeUpscaleResult = Effect.fn("StabilityImages.decodeUpscaleResult")(fun ) { if (response.status === 202) { const output = yield* upscaleRoute.text(response) - return yield* output.invalid(`${upscaleRoute.name} upscale ${context.token.id} has not finished`) + return yield* output.pending(context.token.id) } return yield* decodeUpscaleImage(response) }) diff --git a/packages/ai/src/protocols/xai-video.ts b/packages/ai/src/protocols/xai-video.ts index a71750987c7..d9e9f2d9edd 100644 --- a/packages/ai/src/protocols/xai-video.ts +++ b/packages/ai/src/protocols/xai-video.ts @@ -136,8 +136,7 @@ const decodeResult = Effect.fn("XAIVideo.decodeResult")(function* ( const output = yield* decodeVideoStatus(response) const decoded = output.value const status = yield* MediaProtocol.status(STATUS, decoded.status, output) - if (status === "running") - return yield* output.invalid(`${route.name} request ${context.token.requestID} has not finished`) + if (status === "running") return yield* output.pending(context.token.requestID) if (status === "failed") { const code = decoded.error?.code ?? undefined const message = decoded.error?.message ?? undefined diff --git a/packages/ai/src/route/media-protocol.ts b/packages/ai/src/route/media-protocol.ts index 0876cf0f89d..89fb70013e5 100644 --- a/packages/ai/src/route/media-protocol.ts +++ b/packages/ai/src/route/media-protocol.ts @@ -200,7 +200,8 @@ 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); `contentPolicy` is a moderated result. + * 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. */ const text = Effect.fn("MediaProtocol.text")(function* (response: HttpClientResponse.HttpClientResponse) { const http = context(response) @@ -229,6 +230,14 @@ export const identity = (input: { readonly id: string; readonly name: string; re ? new ProviderInternalError({ message, body, http }) : new InvalidRequestError({ message, body, http }), }), + pending: (id: string) => + new AIError({ + reason: new InvalidRequestError({ + message: `${input.name} generation ${id} has not finished; await it before reading the result`, + body, + http, + }), + }), contentPolicy: (message: string) => new AIError({ reason: new ContentPolicyError({ message, body, http }) }), } }) @@ -290,9 +299,8 @@ export const status = >( raw: string, output: Output, ): Effect.Effect => { - const normalized: Status | undefined = table[raw] - if (normalized === undefined) return Effect.fail(output.invalid(`Unknown generation status "${raw}"`)) - return Effect.succeed(normalized) + if (!Object.hasOwn(table, raw)) return Effect.fail(output.invalid(`Unknown generation status "${raw}"`)) + return Effect.succeed(table[raw]) } /** A `url` asset whose provider-declared retention window starts now. */ diff --git a/packages/ai/test/image.test.ts b/packages/ai/test/image.test.ts index 9701835be80..149edfb2f1e 100644 --- a/packages/ai/test/image.test.ts +++ b/packages/ai/test/image.test.ts @@ -725,6 +725,7 @@ describe("Image", () => { const errors = yield* Effect.all( [ Image.start({ model: Google.configure({ apiKey: "test" }).image("gemini-3.1-flash-image"), prompt }), + Image.generate({ model: Google.configure({ apiKey: "test" }).image("gemini-3.1-flash-image"), prompt, n: 2 }), Image.start({ model: BlackForestLabs.configure({ apiKey: "test" }).image("flux-2-pro"), prompt, @@ -749,6 +750,7 @@ describe("Image", () => { expect(errors.map((error) => [error.reason._tag, "operation" in error.reason && error.reason.operation])).toEqual( [ ["UnsupportedOperation", "image.start"], + ["UnsupportedOperation", "media.n"], ["UnsupportedOperation", "media.aspectRatio"], ["UnsupportedOperation", "media.size"], ["UnsupportedOperation", "media.stream"], @@ -842,6 +844,56 @@ describe("Image", () => { output: { text: "not an image" }, urls: { get: "https://replicate.test/p_1", cancel: "https://replicate.test/p_1/cancel" }, } + for (const pending of [ + { + model: BlackForestLabs.configure({ apiKey: "test" }).image("flux-2-pro"), + token: { id: "req_1", pollingURL: "https://bfl.test/v1/get_result?id=req_1" }, + status: 200, + body: { id: "req_1", status: "Pending" }, + message: "Black Forest Labs generation req_1", + }, + { + model: Replicate.configure({ apiKey: "test" }).image("owner/model"), + token: { id: "p_1", getURL: "https://replicate.test/p_1", cancelURL: "https://replicate.test/p_1/cancel" }, + status: 200, + body: { + id: "p_1", + status: "processing", + urls: { get: "https://replicate.test/p_1", cancel: "https://replicate.test/p_1/cancel" }, + }, + message: "Replicate generation p_1", + }, + { + model: Stability.configure({ apiKey: "test", baseURL: "https://stability.test" }).upscale(), + token: { id: "up_1" }, + status: 202, + body: { id: "up_1", status: "in-progress" }, + message: "Stability AI generation up_1", + }, + ]) { + it.effect(`rejects reading a ${pending.model.provider} result before the generation finishes`, () => + Effect.gen(function* () { + const generation = yield* Image.resume(pending.model, pending.token) + const error = yield* generation.result().pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toBe(`${pending.message} has not finished; await it before reading the result`) + expect(error.reason.body).toBe(JSON.stringify(pending.body)) + expect(error.reason.http?.status).toBe(pending.status) + }).pipe( + Effect.provide( + layer((input) => + Effect.succeed( + input.respond(JSON.stringify(pending.body), { + status: pending.status, + headers: { "content-type": "application/json" }, + }), + ), + ), + ), + ), + ) + } + it.effect("classifies terminal outcomes the recordings never saw", () => Effect.gen(function* () { const bfl = yield* Image.resume(BlackForestLabs.configure({ apiKey: "test" }).image("flux-2-pro"), { diff --git a/packages/ai/test/transcription.test.ts b/packages/ai/test/transcription.test.ts index cdb0e088fc7..e74ed34cf84 100644 --- a/packages/ai/test/transcription.test.ts +++ b/packages/ai/test/transcription.test.ts @@ -191,4 +191,25 @@ describe("Transcription", () => { ]) }), ) + + it.effect("rejects reading an AssemblyAI result before the transcript finishes", () => + Effect.gen(function* () { + const generation = yield* Transcription.resume(assemblyai, { transcriptID: "tr_1" }) + const error = yield* generation.result().pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toBe("AssemblyAI generation tr_1 has not finished; await it before reading the result") + expect(error.reason.body).toBe(JSON.stringify({ id: "tr_1", status: "processing" })) + expect(error.reason.http?.status).toBe(200) + }).pipe( + Effect.provide( + layer((input) => + Effect.succeed( + input.respond(JSON.stringify({ id: "tr_1", status: "processing" }), { + headers: { "content-type": "application/json" }, + }), + ), + ), + ), + ), + ) }) diff --git a/packages/ai/test/video.test.ts b/packages/ai/test/video.test.ts index dfc0d4f8db6..5201c9140ae 100644 --- a/packages/ai/test/video.test.ts +++ b/packages/ai/test/video.test.ts @@ -838,3 +838,55 @@ describe("Video / Runway", () => { }), ) }) + +// --------------------------------------------------------------------------- +// Shared queued behavior +// --------------------------------------------------------------------------- + +describe("Video / queued result", () => { + for (const pending of [ + { + model: Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).video("veo-3.1"), + token: { operation: "models/veo-3.1/operations/op_1" }, + body: { name: "models/veo-3.1/operations/op_1", done: false }, + name: "Google Veo", + }, + { + model: XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }).video("grok-imagine-video-1.5"), + token: { requestID: "req_1" }, + body: { status: "pending", progress: 40 }, + name: "xAI Video", + }, + { + model: Runway.configure({ apiKey: "test", baseURL: "https://runway.test/v1" }).video("gen4.5"), + token: { taskID: "task_1" }, + body: { status: "RUNNING", progress: 0.5 }, + name: "Runway", + }, + ]) { + it.effect(`rejects reading a ${pending.model.provider} result before the generation finishes`, () => + Effect.gen(function* () { + const generation = yield* Video.resume(pending.model, pending.token) + const error = yield* generation.result().pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toBe( + `${pending.name} generation ${generation.id} has not finished; await it before reading the result`, + ) + expect(error.reason.body).toBe(JSON.stringify(pending.body)) + expect(error.reason.http?.status).toBe(200) + }).pipe(Effect.provide(layer((input) => Effect.succeed(json(input, pending.body))))), + ) + } + + it.effect("rejects a status that only matches an inherited property", () => + Effect.gen(function* () { + const error = yield* Video.resume( + XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }).video("grok-imagine-video-1.5"), + { requestID: "req_1" }, + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.message).toBe('Unknown generation status "constructor"') + expect(error.reason.body).toBe(JSON.stringify({ status: "constructor" })) + }).pipe(Effect.provide(layer((input) => Effect.succeed(json(input, { status: "constructor" }))))), + ) +})