fix(ai): tighten media error consistency (#51374)

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 ?? "",

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.

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) =>

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

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))

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({

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)
 })

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

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 = <Table extends Record<string, Status>>(
   raw: string,
   output: Output,
 ): Effect.Effect<Status, AIError> => {
-  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. */

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

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

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