diff --git a/packages/ai/test/image.test.ts b/packages/ai/test/image.test.ts index b10e2c97487..9b6768a6cef 100644 --- a/packages/ai/test/image.test.ts +++ b/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", diff --git a/packages/ai/test/transcription.test.ts b/packages/ai/test/transcription.test.ts index 544e505616d..71ac90434d5 100644 --- a/packages/ai/test/transcription.test.ts +++ b/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[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 = [] + 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 = [] + 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 = [] + 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 = [] + 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 = [] + 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", () => diff --git a/packages/ai/test/video.test.ts b/packages/ai/test/video.test.ts index 5201c9140ae..64d16d0e78a 100644 --- a/packages/ai/test/video.test.ts +++ b/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 = [] + const events: Array = [] + 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 = [] + 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 = [] + 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(