test(ai): cover queued media failures, resume, cancel, and transcription sources (#51626)

packages/ai/test/image.test.ts

@@ -842,6 +842,30 @@ describe("Image", () => {
     ),
   )
 
+  const falDetail = { detail: [{ loc: ["body", "prompt"], msg: "Invalid input", type: "value_error" }] }
+  it.effect(
+    "fails a fal await whose COMPLETED status carries an error with the response_url body and HTTP context",
+    () =>
+      Effect.gen(function* () {
+        const generation = yield* Image.resume(Fal.configure({ apiKey: "test" }).image("fal-ai/flux/schnell"), falToken)
+        expect(generation.status).toBe("failed")
+        const error = yield* generation.await().pipe(Effect.flip)
+        expect(error.reason._tag).toBe("InvalidRequest")
+        expect(error.reason.body).toBe(JSON.stringify(falDetail))
+        expect(error.reason.http).toMatchObject({ url: falToken.responseURL, status: 422 })
+      }).pipe(
+        Effect.provide(
+          layer((input) =>
+            Effect.succeed(
+              input.request.url === falToken.statusURL
+                ? json(input, { status: "COMPLETED", error: "Invalid input", error_type: "ValidationError" })
+                : json(input, falDetail, { status: 422 }),
+            ),
+          ),
+        ),
+      ),
+  )
+
   const moderated = { id: "req_1", status: "Content Moderated" }
   const prediction = {
     id: "p_1",

packages/ai/test/transcription.test.ts

@@ -2,10 +2,11 @@ import { describe, expect } from "bun:test"
 import { Effect, Fiber, Layer, Stream } from "effect"
 import * as TestClock from "effect/testing/TestClock"
 import { HttpClientRequest } from "effect/unstable/http"
-import { Media, Transcription, TranscriptionClient } from "../src/index.js"
+import { Media, Transcription, TranscriptionClient, type TranscriptionEvent } from "../src/index.js"
 import { AssemblyAI, Deepgram, Google, OpenAI } from "../src/providers.js"
 import { it } from "./lib/effect.js"
 import { dynamicResponse, json, observe, type Call } from "./lib/http.js"
+import { sseEvents } from "./lib/sse.js"
 
 const layer = (handler: Parameters<typeof dynamicResponse>[0]) =>
   TranscriptionClient.layer.pipe(Layer.provideMerge(dynamicResponse(handler)))
@@ -16,6 +17,21 @@ const deepgram = Deepgram.configure({ apiKey: "test", baseURL: "https://deepgram
 const google = Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).transcription(
   "gemini-3.5-transcribe",
 )
+/**
+ * Multipart fields of a recorded request, with repeated names collected in order. The boundary comes from the body:
+ * each conversion of a FormData request to a web request picks a fresh one, so the recorded headers may not match.
+ */
+const formFields = (call: Call) =>
+  Effect.promise(() =>
+    new Response(call.body, {
+      headers: { "content-type": `multipart/form-data; boundary=${call.body.slice(2, call.body.indexOf("\r\n"))}` },
+    }).formData(),
+  ).pipe(
+    Effect.map((form) =>
+      Object.fromEntries([...new Set(form.keys())].map((key) => [key, form.getAll(key).map((value) => String(value))])),
+    ),
+  )
+
 const assemblyai = AssemblyAI.configure({ apiKey: "aai-key", baseURL: "https://assemblyai.test" }).transcription(
   "universal-3-5-pro",
 )
@@ -129,6 +145,198 @@ describe("Transcription", () => {
     }),
   )
 
+  it.effect("streams diarized segments and finishes with the accumulated segments", () =>
+    Effect.gen(function* () {
+      const calls: Array<Call> = []
+      const body = sseEvents(
+        { type: "transcript.text.segment", id: "seg_0", text: " Hello", start: 0.25, end: 0.7, speaker: "A" },
+        { type: "transcript.text.segment", id: "seg_1", text: " there.", start: 0.7, end: 1.25, speaker: "B" },
+        { type: "transcript.text.done", text: "Hello there.", usage: { type: "duration", seconds: 2 } },
+      )
+      const events = Array.from(
+        yield* Stream.runCollect(
+          Transcription.stream({ model: openai.transcription("gpt-4o-transcribe-diarize"), audio, diarize: true }),
+        ).pipe(
+          Effect.provide(
+            layer((input) =>
+              observe(calls, input).pipe(
+                Effect.as(input.respond(body, { headers: { "content-type": "text/event-stream" } })),
+              ),
+            ),
+          ),
+        ),
+      )
+
+      const form = yield* formFields(calls[0])
+      expect(form).toMatchObject({
+        model: ["gpt-4o-transcribe-diarize"],
+        response_format: ["diarized_json"],
+        chunking_strategy: ["auto"],
+        stream: ["true"],
+      })
+      const segments = [
+        { text: "Hello", startSeconds: 0.25, endSeconds: 0.7, speaker: "A" },
+        { text: "there.", startSeconds: 0.7, endSeconds: 1.25, speaker: "B" },
+      ]
+      expect(events).toEqual([
+        { type: "segment", segment: segments[0] },
+        { type: "segment", segment: segments[1] },
+        expect.objectContaining({
+          type: "finish",
+          text: "Hello there.",
+          segments,
+          usage: { type: "seconds", seconds: 2 },
+        }),
+      ])
+    }),
+  )
+
+  it.effect("requests whisper-1 segment timestamps as verbose_json", () =>
+    Effect.gen(function* () {
+      const calls: Array<Call> = []
+      const response = yield* Transcription.generate({
+        model: openai.transcription("whisper-1"),
+        audio,
+        timestamps: "segment",
+      }).pipe(
+        Effect.provide(
+          layer((input) =>
+            observe(calls, input).pipe(
+              Effect.as(
+                json(input, {
+                  text: "Hello there.",
+                  language: "English",
+                  duration: 1.25,
+                  segments: [
+                    { id: 0, text: " Hello", start: 0.25, end: 0.7 },
+                    { id: 1, text: " there.", start: 0.7, end: 1.25 },
+                  ],
+                  usage: { type: "duration", seconds: 2 },
+                }),
+              ),
+            ),
+          ),
+        ),
+      )
+
+      const form = yield* formFields(calls[0])
+      expect(form).toMatchObject({
+        model: ["whisper-1"],
+        response_format: ["verbose_json"],
+        "timestamp_granularities[]": ["segment"],
+      })
+      expect(form.stream).toBeUndefined()
+      expect(response).toMatchObject({
+        text: "Hello there.",
+        segments: [
+          { text: "Hello", startSeconds: 0.25, endSeconds: 0.7 },
+          { text: "there.", startSeconds: 0.7, endSeconds: 1.25 },
+        ],
+        language: "english",
+        durationSeconds: 1.25,
+        usage: { type: "seconds", seconds: 2 },
+      })
+    }),
+  )
+
+  it.effect("fails an OpenAI stream that ends without transcript.text.done as incomplete", () =>
+    Effect.gen(function* () {
+      const events: Array<TranscriptionEvent> = []
+      const error = yield* Transcription.stream({ model: openai.transcription("gpt-4o-mini-transcribe"), audio }).pipe(
+        Stream.runForEach((event) => Effect.sync(() => events.push(event))),
+        Effect.flip,
+        Effect.provide(
+          layer((input) =>
+            Effect.succeed(
+              input.respond(sseEvents({ type: "transcript.text.delta", delta: "Hel" }), {
+                headers: { "content-type": "text/event-stream" },
+              }),
+            ),
+          ),
+        ),
+      )
+
+      expect(events).toEqual([{ type: "text-delta", delta: "Hel" }])
+      expect(error.reason).toMatchObject({ _tag: "InvalidProviderOutput", classification: "incomplete-stream" })
+      expect(error.reason.http?.status).toBe(200)
+    }),
+  )
+
+  it.effect("sends a Deepgram URL source as a JSON body and repeats array query parameters", () =>
+    Effect.gen(function* () {
+      const calls: Array<Call> = []
+      const response = yield* Transcription.generate({
+        model: deepgram,
+        audio: Media.url("https://a.test/call.mp3", { mediaType: "audio/mpeg" }),
+        language: "en",
+        providerOptions: { keyterm: ["OpenCode", "Effect"] },
+      }).pipe(
+        Effect.provide(
+          layer((input) =>
+            observe(calls, input).pipe(
+              Effect.as(
+                json(input, {
+                  metadata: { request_id: "dg_1", duration: 2 },
+                  results: { channels: [{ alternatives: [{ transcript: "Hello there." }] }] },
+                }),
+              ),
+            ),
+          ),
+        ),
+      )
+
+      expect(calls).toHaveLength(1)
+      const url = new URL(calls[0].url)
+      expect(url.origin + url.pathname).toBe("https://deepgram.test/v1/listen")
+      expect([...url.searchParams]).toEqual([
+        ["model", "nova-3"],
+        ["smart_format", "true"],
+        ["language", "en"],
+        ["keyterm", "OpenCode"],
+        ["keyterm", "Effect"],
+      ])
+      expect(calls[0].headers.get("content-type")).toBe("application/json")
+      expect(JSON.parse(calls[0].body)).toEqual({ url: "https://a.test/call.mp3" })
+      expect(response).toMatchObject({
+        text: "Hello there.",
+        usage: { type: "seconds", seconds: 2 },
+        providerMetadata: { deepgram: { requestId: "dg_1" } },
+      })
+    }),
+  )
+
+  it.effect("transcribes an AssemblyAI URL source without uploading it first", () =>
+    Effect.gen(function* () {
+      const calls: Array<Call> = []
+      const response = yield* Transcription.generate({
+        model: assemblyai,
+        audio: Media.url("https://a.test/call.mp3", { mediaType: "audio/mpeg" }),
+      }).pipe(
+        Effect.provide(
+          layer((input) =>
+            Effect.gen(function* () {
+              const { call } = yield* observe(calls, input)
+              if (call.method === "POST") return json(input, { id: "tr_1", status: "queued" })
+              return json(input, { id: "tr_1", status: "completed", text: "Hello there.", audio_duration: 2 })
+            }),
+          ),
+        ),
+      )
+
+      expect(calls.map((call) => `${call.method} ${call.url}`)).toEqual([
+        "POST https://assemblyai.test/v2/transcript",
+        "GET https://assemblyai.test/v2/transcript/tr_1",
+        "GET https://assemblyai.test/v2/transcript/tr_1",
+      ])
+      expect(JSON.parse(calls[0].body)).toEqual({
+        audio_url: "https://a.test/call.mp3",
+        speech_models: ["universal-3-5-pro"],
+        language_detection: true,
+      })
+      expect(response).toMatchObject({ text: "Hello there.", usage: { type: "seconds", seconds: 2 } })
+    }),
+  )
+
   it.effect(
     "uploads inline audio to AssemblyAI, resumes polling from a persisted token, and surfaces failed transcripts",
     () =>

packages/ai/test/video.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect } from "bun:test"
 import { Effect, Layer, Stream } from "effect"
-import { Media, Video, VideoClient, type GenerationEvent } from "../src/index.js"
+import { Media, Video, VideoClient, type GenerationEvent, type VideoEvent } from "../src/index.js"
 import { Fal, Google, Runway, XAI } from "../src/providers.js"
 import { it } from "./lib/effect.js"
 import { dynamicResponse, json, observe, settle, type Call } from "./lib/http.js"
@@ -542,6 +542,42 @@ describe("Video / fal", () => {
     ),
   )
 
+  for (const failure of [
+    {
+      name: "a COMPLETED status carrying an error",
+      status: { status: "COMPLETED", error: "Invalid input", error_type: "ValidationError" },
+      result: { status: 422, body: { detail: [{ loc: ["body", "prompt"], msg: "Invalid input" }] } },
+      tag: "InvalidRequest",
+    },
+    {
+      name: "a failing response_url",
+      status: { status: "COMPLETED" },
+      result: { status: 500, body: { detail: "Internal error" } },
+      tag: "ProviderInternal",
+    },
+  ]) {
+    it.effect(`fails await for ${failure.name} with the response_url body and HTTP context`, () =>
+      Effect.gen(function* () {
+        const error = yield* Video.generate({ model, prompt: "x" }).pipe(Effect.flip)
+        expect(error.reason._tag).toBe(failure.tag)
+        expect(error.reason.body).toBe(JSON.stringify(failure.result.body))
+        expect(error.reason.http).toMatchObject({ url: urls.response, status: failure.result.status })
+      }).pipe(
+        Effect.provide(
+          layer((input) =>
+            Effect.succeed(
+              input.request.method === "POST"
+                ? json(input, submitted)
+                : input.request.url === urls.response
+                  ? json(input, failure.result.body, { status: failure.result.status })
+                  : json(input, failure.status),
+            ),
+          ),
+        ),
+      ),
+    )
+  }
+
   it.effect("rejects model-specific common fields and points at providerOptions", () =>
     Effect.gen(function* () {
       const errors = yield* Effect.forEach(
@@ -818,6 +854,39 @@ describe("Video / Runway", () => {
     }),
   )
 
+  it.effect("streams the observations of a failed task and then fails with the task body", () =>
+    Effect.gen(function* () {
+      const calls: Array<Call> = []
+      const events: Array<VideoEvent> = []
+      const failed = { status: "FAILED", failure: "Something broke", failureCode: "INTERNAL.BAD_OUTPUT.CODE01" }
+      const program = Video.stream({ model, prompt: "x" }, { poll: { interval: "1 second" } }).pipe(
+        Stream.runForEach((event) => Effect.sync(() => events.push(event))),
+        Effect.flip,
+      )
+      const error = yield* settle(program, 3).pipe(
+        Effect.provide(
+          layer((input) =>
+            Effect.gen(function* () {
+              const { call, nth } = yield* observe(calls, input)
+              if (call.method === "POST") return json(input, { id: "task_1" })
+              if (nth === 1) return json(input, { status: "PENDING" })
+              if (nth === 2) return json(input, { status: "RUNNING", progress: 0.5 })
+              return json(input, failed)
+            }),
+          ),
+        ),
+      )
+      expect(events).toEqual([
+        { type: "generation-queued", id: "task_1", position: undefined },
+        { type: "generation-progress", id: "task_1", progress: 0.5 },
+      ])
+      expect(error.reason._tag).toBe("ProviderInternal")
+      expect(error.message).toBe("Runway task failed (INTERNAL.BAD_OUTPUT.CODE01): Something broke")
+      expect(error.reason.body).toBe(JSON.stringify(failed))
+      expect(error.reason.http?.status).toBe(200)
+    }),
+  )
+
   it.effect("fails a stream with a Timeout reason once polling passes the poll deadline", () =>
     Effect.gen(function* () {
       const program = Video.stream(
@@ -878,6 +947,123 @@ describe("Video / queued result", () => {
     )
   }
 
+  const veoOperation = "models/veo-3.1/operations/op_1"
+  const falURLs = {
+    status: "https://queue.fal.test/fal-ai/veo3.1/requests/r1/status",
+    response: "https://queue.fal.test/fal-ai/veo3.1/requests/r1",
+    cancel: "https://queue.fal.test/fal-ai/veo3.1/requests/r1/cancel",
+  }
+  for (const queued of [
+    {
+      model: Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).video("veo-3.1"),
+      submitted: { name: veoOperation },
+      token: { operation: veoOperation },
+      submitURL: "https://google.test/v1beta/models/veo-3.1:predictLongRunning",
+      statusURL: `https://google.test/v1beta/${veoOperation}`,
+      resultURL: `https://google.test/v1beta/${veoOperation}`,
+      running: { name: veoOperation, done: false },
+      done: {
+        name: veoOperation,
+        done: true,
+        response: { generateVideoResponse: { generatedSamples: [{ video: { uri: "https://google.test/out.mp4" } }] } },
+      },
+      result: undefined,
+      url: "https://google.test/out.mp4",
+    },
+    {
+      model: XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }).video("grok-imagine-video-1.5"),
+      submitted: { request_id: "req_1" },
+      token: { requestID: "req_1" },
+      submitURL: "https://xai.test/v1/videos/generations",
+      statusURL: "https://xai.test/v1/videos/req_1",
+      resultURL: "https://xai.test/v1/videos/req_1",
+      running: { status: "pending", progress: 40 },
+      done: { status: "done", video: { url: "https://vidgen.x.ai/out.mp4", respect_moderation: true } },
+      result: undefined,
+      url: "https://vidgen.x.ai/out.mp4",
+    },
+    {
+      model: Fal.configure({ apiKey: "test", baseURL: "https://queue.fal.test" }).video("fal-ai/veo3.1"),
+      submitted: {
+        request_id: "r1",
+        status_url: falURLs.status,
+        response_url: falURLs.response,
+        cancel_url: falURLs.cancel,
+      },
+      token: { requestID: "r1", statusURL: falURLs.status, responseURL: falURLs.response, cancelURL: falURLs.cancel },
+      submitURL: "https://queue.fal.test/fal-ai/veo3.1",
+      statusURL: falURLs.status,
+      resultURL: falURLs.response,
+      running: { status: "IN_PROGRESS" },
+      done: { status: "COMPLETED" },
+      result: { video: { url: "https://v3.fal.media/out.mp4" } },
+      url: "https://v3.fal.media/out.mp4",
+    },
+  ]) {
+    it.effect(`resumes a ${queued.model.provider} generation from a JSON round-tripped token`, () =>
+      Effect.gen(function* () {
+        const calls: Array<Call> = []
+        const response = yield* Effect.gen(function* () {
+          const started = yield* Video.start({ model: queued.model, prompt: "x" })
+          const resumed = yield* Video.resume(queued.model, JSON.parse(JSON.stringify(started.token)))
+          expect(resumed.status).toBe("running")
+          expect(resumed.token).toEqual(queued.token)
+          return yield* resumed.await()
+        }).pipe(
+          Effect.provide(
+            layer((input) =>
+              Effect.gen(function* () {
+                const { call, nth } = yield* observe(calls, input)
+                if (call.method === "POST") return json(input, queued.submitted)
+                if (call.url === queued.resultURL && queued.result !== undefined) return json(input, queued.result)
+                return json(input, nth === 1 ? queued.running : queued.done)
+              }),
+            ),
+          ),
+        )
+        expect(response.video.source).toEqual(expect.objectContaining({ type: "url", url: queued.url }))
+        expect(calls.map((call) => `${call.method} ${call.url}`)).toEqual([
+          `POST ${queued.submitURL}`,
+          `GET ${queued.statusURL}`,
+          `GET ${queued.statusURL}`,
+          `GET ${queued.resultURL}`,
+        ])
+      }),
+    )
+  }
+
+  for (const queued of [
+    {
+      model: Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).video("veo-3.1"),
+      submitted: { name: veoOperation },
+    },
+    {
+      model: XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }).video("grok-imagine-video-1.5"),
+      submitted: { request_id: "req_1" },
+    },
+  ]) {
+    it.effect(`cancels a ${queued.model.provider} generation without sending a request`, () =>
+      Effect.gen(function* () {
+        const calls: Array<Call> = []
+        yield* Effect.gen(function* () {
+          const generation = yield* Video.start({ model: queued.model, prompt: "x" })
+          yield* generation.cancel()
+        }).pipe(
+          Effect.provide(
+            layer((input) =>
+              Effect.gen(function* () {
+                const { call } = yield* observe(calls, input)
+                if (call.method !== "POST") return yield* Effect.die(`cancel sent ${call.method} ${call.url}`)
+                return json(input, queued.submitted)
+              }),
+            ),
+          ),
+        )
+        expect(calls.map((call) => call.method)).toEqual(["POST"])
+      }),
+    )
+  }
+
   it.effect("rejects a status that only matches an inherited property", () =>
     Effect.gen(function* () {
       const error = yield* Video.resume(