From 5eac2802b7f482c7bb95f3d9505bc6acfbb1e6c2 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 12 Nov 2026 21:56:34 -0600 Subject: [PATCH] feat(ai): graduate video generation with start, status, result and cancel (#63657) --- packages/ai/README.md | 162 ++-- packages/ai/src/effect.ts | 3 +- packages/ai/src/experimental.ts | 11 - packages/ai/src/index.ts | 55 +- packages/ai/src/media-model.ts | 8 +- packages/ai/src/media.ts | 50 +- packages/ai/src/protocols/xai-video.ts | 47 +- packages/ai/src/public.ts | 11 + packages/ai/src/route/media.ts | 26 +- packages/ai/src/video-client.ts | 63 +- packages/ai/src/video.ts | 167 ++--- packages/ai/test/exports.test.ts | 10 +- packages/ai/test/media.test.ts | 27 + packages/ai/test/promise-parity.types.ts | 14 +- packages/ai/test/promise.test.ts | 92 +-- .../test/provider/xai-video.recorded.test.ts | 22 +- packages/ai/test/video.test.ts | 703 +++++++----------- packages/ai/test/video.types.ts | 101 +-- 18 files changed, 762 insertions(+), 810 deletions(-) diff --git a/packages/ai/README.md b/packages/ai/README.md index 25273f576d75..4f66c87f968f 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -1,7 +1,7 @@ # @opencode/ai -Schema-first APIs for text, images, decisions, and embeddings, built with Effect. Video, speech, transcription, extraction, and -realtime voice are [experimental](#experimental). +Schema-first APIs for text, images, video, decisions, and embeddings, built with Effect. Speech, transcription, +extraction, and realtime voice are [experimental](#experimental). `@opencode/ai` is the Promise API: one managed runtime behind the LLM and media methods, plus asset helpers. `ai.file` and `ai.write` load `node:fs/promises` on first use, so no Effect `FileSystem` is needed: @@ -211,7 +211,7 @@ Raw usage is preserved in `usage.providerMetadata.google`. `inputTokens` follows `contextTokens` uses its full `raw_prompt_token` count when supplied. These can differ substantially with server-side continuation. Explicit caches, hosted tools, and generated media are not supported by this initial protocol. -The same configured facade names image models (and the experimental video, speech, and transcription models). +The same configured facade names image and video models (and the experimental speech and transcription models). `ai.image.generate` resolves the provider's image route from the model and returns `Media.Asset`s with lazily decoded bytes: ```ts @@ -1031,6 +1031,72 @@ const program = Effect.gen(function* () { The hosted result is represented as a provider-executed tool call and a tool result whose content carries the generated image as a file. Gemini image-capable models instead emit a first-class `media` `LLMEvent` for inline image output (`response.message` then carries a `media` part). Retaining `response.message` preserves the generated image for continuation on both routes. +## Video generation + +Video generation is a job: `start` submits it and returns at once, and the caller decides when to read its `status`, +fetch its `result`, or `cancel` it. Nothing polls for you. Every call after `start` takes the model and the job's +`token`, plain JSON you can store and read back from another process, so recovering after a restart is the same call +as a normal status read. Models come from `.video(...)` selectors on the `XAI` facade. + +```ts +import { Effect } from "effect" +import { ai, Media } from "@opencode/ai/effect" +import { XAI } from "@opencode/ai/providers" + +const model = XAI.configure({ apiKey: process.env.XAI_API_KEY }).video("grok-imagine-video-1.5") + +const program = Effect.gen(function* () { + const job = yield* ai.video.start({ + model, + prompt: "Panning wide shot of a calico kitten sleeping in the sunshine", + start: "https://example.com/first-frame.png", // exact first frame; `end` pins the last one + references: [], // images, videos or audio that guide subject or style + duration: "8 seconds", + aspect: "16:9", // or "auto" to follow the start image + resolution: "720p", // 480p | 720p | 1080p | 4k, or any provider value + sound: false, + }) + job.token // `{ requestID }` for xAI: store it with the provider and model ID + + const status = yield* ai.video.status(model, job.token) + status.status // "queued" | "running" | "completed" | "failed" | "cancelled" | "expired" + status.progress // 0..1, only when the provider reports it + + const response = yield* ai.video.result(model, job.token) + yield* Media.write(response.video, "./kitten.mp4") // streams the download to disk +}) +``` + +The shared fields are `prompt`, `start`, `end`, `references`, `duration`, `aspect`, `resolution` and `sound`; +everything else is provider-specific and goes under `providerOptions`, typed from the selected model. A request only +fails before sending when its protocol has no way to express it, such as a reference video on xAI. Values the provider +may reject, like an unsupported duration or too many references, go to the provider, and its error comes back typed. + +`status` is one read; `result` fails with an `InvalidRequest` reason when the job hasn't finished, and with the +provider's reason (and terminal document on `reason.body`) when it failed, was cancelled or expired. Moderation +outcomes (xAI `respect_moderation`) fail with a `ContentPolicy` reason when no video comes back. Status and result reads +retry transient failures (rate limits, provider 5xx, network errors) with backoff that honors `retry-after`; `start` +and `cancel` never retry. `cancel` fails with an `UnsupportedOperation` reason on providers without a cancel endpoint, +such as xAI, where the job keeps running and billing. Finished videos are usually URLs that expire (`expiresAt` when the +provider documents it): download them with `Media.write`, `asset.stream()` or `materialize()` before then. + +Provider notes: + +- **xAI** maps `start` to `image`, `end` to `last_frame`, image references to `reference_images`, audio references to + `reference_audios`, and `sound` to `generate_audio`, and leaves `aspect: "auto"` out when a start frame sets the + shape. It has no reference video input. `providerOptions.mode: "edit" | "extend"` sends a native `video: { url }` + source to `/videos/edits` or `/videos/extensions`. + +The promise client has the same calls, each accepting `{ signal }`: + +```ts +import { ai } from "@opencode/ai" + +const job = await ai.video.start({ model, prompt }) +const status = await ai.video.status(model, job.token) +if (status.status === "completed") await ai.write((await ai.video.result(model, job.token)).video, "./kite.mp4") +``` + ## Public API Everything below is exported from `@opencode/ai/effect` unless noted. @@ -1044,108 +1110,30 @@ Everything below is exported from `@opencode/ai/effect` unless noted. - **`LLMEvent.is.*`** — typed guards (`is.textDelta`, `is.toolCall`, `is.finish`, …) for filtering streams. - **`ai.image.request` / `generate` / `stream` / `start` / `resume`** — images over inline, streaming (partial previews), and queued routes through a provider-neutral request and response model. - **`ImageClient`** — Effect service and layer for image execution, parallel to `LLMClient`. +- **`ai.video.request` / `start` / `status` / `result` / `cancel`** — video generation as a job addressed by a stored token; see [Video generation](#video-generation). `VideoClient` is its Effect service and layer. - **`Media`** — the shared asset type (`Media.Asset`, `Media.Source`) and constructors used by messages, tool results, and media requests. - **`Generation`** — provider-neutral handle for an in-flight media generation (`await`, `refresh`, `cancel`, `events`) used by queued media routes. - **`ai.decision.request` / `ai.decision.generate`** — typed predicate, choice, and score questions over text, JSON, and media input; `DecisionClient` is its Effect service and layer. - **`ai.embedding.request` / `ai.embedding.generate`** — one vector per text or media value, with optional `dimensions` and `inputType`; `EmbeddingClient` is its Effect service and layer. - **`AIClient.layer` / `AIClient.layerWith(executor)`** — every modality client plus the request executor in one layer. -- **`@opencode/ai`** (Promise root) — a default `ai` with the text calls (`request`, `generate`, `stream`, `loop`, `compact`, `canCompact`) at the top level plus `image`, `decision` and `embedding` (plus `experimental.video`, `experimental.speech`, and `experimental.transcription`) as Promise / `AsyncIterable` APIs, `run(effect)` for any package Effect, plus `file`, `write`, `bytes`, `base64`, and `materialize` for assets. `AI.make({ layer? })` builds a client over a custom executor. The same `ai` from `@opencode/ai/effect` returns `Effect` and `Stream`. -- **`@opencode/ai/experimental`** — APIs that may change without compatibility guarantees: `Video`, `Speech`, `Transcription`, `Extraction`, and `Realtime`. See [Experimental](#experimental). +- **`@opencode/ai`** (Promise root) — a default `ai` with the text calls (`request`, `generate`, `stream`, `loop`, `compact`, `canCompact`) at the top level plus `image`, `decision`, `embedding` and `video` (plus `experimental.speech` and `experimental.transcription`) as Promise / `AsyncIterable` APIs, `run(effect)` for any package Effect, plus `file`, `write`, `bytes`, `base64`, and `materialize` for assets. `AI.make({ layer? })` builds a client over a custom executor. The same `ai` from `@opencode/ai/effect` returns `Effect` and `Stream`. +- **`@opencode/ai/experimental`** — APIs that may change without compatibility guarantees: `Speech`, `Transcription`, `Extraction`, and `Realtime`. See [Experimental](#experimental). ## Experimental These APIs may change without compatibility guarantees. The Effect API is exported from `@opencode/ai/experimental`, and the Promise client exposes it under `ai.experimental`. -- **`Video`**, **`Speech`**, and **`Transcription`** — media generation and speech-to-text, with `VideoClient`, - `SpeechClient`, and `TranscriptionClient` as their Effect services. Promise: `ai.experimental.video`, `.speech`, and - `.transcription`. +- **`Speech`** and **`Transcription`** — speech generation and speech-to-text, with `SpeechClient` and + `TranscriptionClient` as their Effect services. Promise: `ai.experimental.speech` and `.transcription`. - **`Extraction`** — typed values read out of text. - **`Realtime`** — `Realtime.connect` starts an OpenAI, xAI, or Gemini voice session and returns a WebRTC answer SDP or a short-lived WebSocket URL and token. -The provider selectors for these modalities (`.video(...)`, `.speech(...)`, `.transcription(...)`, +The provider selectors for these modalities (`.speech(...)`, `.transcription(...)`, `.experimental.extraction(...)`) and their protocol entrypoints (such as `@opencode/ai/protocols/openai-speech`) are part of the same experimental surface. -### Video generation - -Video mirrors `Image` with one difference: every provider is asynchronous, so the route is a submit-then-poll -`Generation`. Models come from `.video(...)` selectors on the `XAI` facade. -Common fields (`frames`, `references`, `video`, `durationSeconds`, `aspect`, `resolution`, `audio`, `n`, `seed`, -`negativePrompt`) lower natively or fail with a typed `AIError` before any network call; provider-native controls live -under `providerOptions`, inferred from the selected model. - -```ts -import { Video } from "@opencode/ai/experimental" -import { XAI } from "@opencode/ai/providers" - -const xai = XAI.configure({ apiKey: process.env.XAI_API_KEY }) - -// Simple: submit and wait. -const program = Effect.gen(function* () { - const response = yield* Video.generate( - { - model: xai.video("grok-imagine-video-1.5"), - prompt: "Panning wide shot of a calico kitten sleeping in the sunshine", - aspect: "16:9", - resolution: "720p", - durationSeconds: 10, - }, - { poll: { interval: "10 seconds", timeout: "10 minutes" } }, - ) - // xAI returns a temporary URL: materialize before persisting, or the persisted URL may stop working. - return yield* response.video.materialize() -}) - -// Explicit control: keep the handle, persist the token, resume elsewhere. -const controlled = Effect.gen(function* () { - const generation = yield* Video.start({ model: xai.video("grok-imagine-video-1.5"), prompt }) - generation.id // provider request id - generation.status // "queued" | "running" | "completed" | "failed" | "cancelled" | "expired" - generation.token // route-owned JSON, `{ requestID }` for xAI - // The token carries no route identity: persist the provider and model ID alongside it, since `resume` needs the model. - const saved = JSON.stringify(generation.token) - - const resumed = yield* Video.resume(xai.video("grok-imagine-video-1.5"), JSON.parse(saved)) - return yield* resumed.await({ poll: { interval: "10 seconds" } }) -}) - -// Progress as a stream: generation-queued | generation-progress | video | finish. -const events = Video.stream({ model: xai.video("grok-imagine-video-1.5"), prompt }, { poll }) -``` - -Status polls, result fetches, and cancels all run through the same request executor with the route's -auth. `Generation.await` and `Generation.events` fail with a -`Timeout` reason when `poll.timeout` (default 10 minutes) elapses. Status polls and result fetches retry transient -failures (rate limits, provider 5xx, network errors) with backoff that honors `retry-after`, always within -`poll.timeout`; submits and cancels never retry. Interrupting a wait (or aborting its `signal`) does not cancel the -provider job, which keeps running and billing: call `cancel()` to stop it on providers with a cancel endpoint (xAI has -none, so `cancel()` sends nothing there). Failed, cancelled, and expired generations fail typed with the provider's -terminal document on `reason.body`; moderation outcomes (xAI `respect_moderation`) fail with a `ContentPolicy` reason -when no video comes back; when one still does, the raw outcome stays in `providerMetadata`. - -Provider notes: - -- **xAI** sends a `video` input to `/videos/edits`, or `/videos/extensions` with `providerOptions.mode: "extend"`. - `seed` and `negativePrompt` are not supported. - -The promise client exposes the same surface: `ai.experimental.video.start(...)` resolves to a handle with `await`, `events`, -`result`, `refresh`, `cancel`, and `token`; `ai.experimental.video.generate`, `ai.experimental.video.resume(model, token)`, and -`ai.experimental.video.stream` mirror the Effect API. The handle's `status` and `progress` are a snapshot from when it was -created; `refresh()` resolves to a new handle. Every promise method and stream accepts `{ signal }`: like `fetch`, -aborting rejects the Promise or throws from the `for await` loop with `signal.reason` (an `AbortError` `DOMException` -unless `abort(reason)` passed one), while `break` stops a stream without throwing. - -```ts -import { ai } from "@opencode/ai" - -const generation = await ai.experimental.video.start({ model, prompt }) -for await (const event of generation.events({ poll: { interval: 10_000 } })) console.log(event.type) -const video = await generation.result({ signal }) -await ai.write(video.video, "./kite.mp4") -``` - ### Speech generation Speech (text-to-speech) is one request whose response is parsed incrementally, so every route supports both diff --git a/packages/ai/src/effect.ts b/packages/ai/src/effect.ts index 7f3d88461c4a..ced74ee2e8ef 100644 --- a/packages/ai/src/effect.ts +++ b/packages/ai/src/effect.ts @@ -18,17 +18,18 @@ export const ai = { image: Image, decision: Decision, embedding: Embedding, + video: Video, /** Experimental: may change without compatibility guarantees. */ experimental: { speech: Speech, transcription: Transcription, - video: Video, }, } as const export { AIClient } from "./ai-client.js" export { LLMClient } from "./route/client.js" export { ImageClient } from "./image-client.js" +export { VideoClient } from "./video-client.js" export { Auth } from "./route/auth.js" export { ProviderPackage } from "./provider-package.js" export type { LLMClientService } from "./route/client.js" diff --git a/packages/ai/src/experimental.ts b/packages/ai/src/experimental.ts index 3e2cce4c3598..77b6e541bd65 100644 --- a/packages/ai/src/experimental.ts +++ b/packages/ai/src/experimental.ts @@ -1,14 +1,3 @@ -export { VideoClient } from "./video-client.js" -export { VideoAspectRatio, VideoEvent, VideoFrames, VideoModel, VideoModelSchema, VideoResponse } from "./video.js" -export type { - VideoModelOptions, - VideoOptions, - VideoRequest, - VideoRequestInput, - VideoResolution, - VideoRoute, -} from "./video.js" -export { Video } from "./video.js" export { SpeechClient } from "./speech-client.js" export { SpeechEvent, SpeechModel, SpeechModelSchema, SpeechResponse, SpeechTimestamp, SpeechVoice } from "./speech.js" export type { diff --git a/packages/ai/src/index.ts b/packages/ai/src/index.ts index 32823a8c9a22..e8475955ed7f 100644 --- a/packages/ai/src/index.ts +++ b/packages/ai/src/index.ts @@ -204,20 +204,23 @@ export const make = (options: Options = {}) => { options, ) }, - /** Write an asset's bytes like `Media.write`, downloading `url` sources through the executor. */ + /** Write an asset's bytes like `Media.write`: streamed through a temporary file that replaces `path` on success. */ write: async (asset: Media.Asset, path: string, options?: RunOptions) => { - const { writeFile } = await import("node:fs/promises") - return run( - asset.bytes().pipe( - Effect.flatMap((data) => - Effect.tryPromise({ - try: (signal) => writeFile(path, data, { signal }), - catch: (cause) => fileError(`Failed to write media file ${path}`, cause), - }), - ), - ), - options, - ) + const { open, rename, rm } = await import("node:fs/promises") + const temporary = `${path}.${crypto.randomUUID()}.tmp` + try { + const file = await open(temporary, "w") + try { + for await (const chunk of iterate(asset.stream(), options)) await file.write(chunk) + } finally { + await file.close() + } + await rename(temporary, path) + } catch (cause) { + await rm(temporary, { force: true }) + if (cause instanceof AIError || cause === options?.signal?.reason) throw cause + throw fileError(`Failed to write media file ${path}`, cause) + } }, request: LLM.request, generate, @@ -255,6 +258,17 @@ export const make = (options: Options = {}) => { generate: (input: EmbeddingRequestInput, options?: RunOptions) => run(Embedding.generate(input), options), }, + video: { + request: Video.request, + /** Submit a generation and return its `{ id, token }` at once; nothing polls. */ + start: (input: VideoRequestInput | VideoRequest, options?: RunOptions) => + run(Video.start(input), options), + /** One status read for a stored token. */ + status: (model: VideoModel, token: unknown, options?: RunOptions) => run(Video.status(model, token), options), + /** The finished videos; fails while the job is still running. */ + result: (model: VideoModel, token: unknown, options?: RunOptions) => run(Video.result(model, token), options), + cancel: (model: VideoModel, token: unknown, options?: RunOptions) => run(Video.cancel(model, token), options), + }, /** Experimental: may change without compatibility guarantees. */ experimental: { speech: { @@ -285,21 +299,6 @@ export const make = (options: Options = {}) => { resume: (model: TranscriptionModel, token: unknown, options?: RunOptions) => run(Transcription.resume(model, token), options).then(handle), }, - video: { - request: Video.request, - start: (input: VideoRequestInput | VideoRequest, options?: RunOptions) => - run(Video.start(input), options).then(handle), - generate: ( - input: VideoRequestInput | VideoRequest, - options?: AwaitOptions & RunOptions, - ) => run(Video.generate(input, { poll: options?.poll }), options), - resume: (model: VideoModel, token: unknown, options?: RunOptions) => - run(Video.resume(model, token), options).then(handle), - stream: ( - input: VideoRequestInput | VideoRequest, - options?: AwaitOptions & RunOptions, - ) => iterate(Video.stream(input, { poll: options?.poll }), options), - }, }, dispose: () => runtime.dispose(), } diff --git a/packages/ai/src/media-model.ts b/packages/ai/src/media-model.ts index 563c6eb827b0..64c0c9470e5a 100644 --- a/packages/ai/src/media-model.ts +++ b/packages/ai/src/media-model.ts @@ -84,10 +84,16 @@ export const composeRoute = ) => Effect.Effect, ): MediaRoute.AnyRoute => { if (isStreamInput(route)) return MediaRoute.stream({ ...composition(route, input), collect }) - if (isQueuedInput(route)) return MediaRoute.queued(composition(route, input)) + if (isQueuedInput(route)) return composeQueuedRoute(route, input) return MediaRoute.inline(composition(route, input)) } +/** Compose a queued protocol route input with one deployment through `MediaRoute.queued`. */ +export const composeQueuedRoute = ( + route: MediaModel.QueuedRouteInput, + input: MediaRoute.ModelInput, +) => MediaRoute.queued(composition(route, input)) + const composition = ( route: MediaModel.RouteInput, input: MediaRoute.ModelInput, diff --git a/packages/ai/src/media.ts b/packages/ai/src/media.ts index 9cc15d8b889c..b35da7c89cb5 100644 --- a/packages/ai/src/media.ts +++ b/packages/ai/src/media.ts @@ -1,6 +1,6 @@ export * as Media from "./media.js" -import { Effect, FileSystem, Result, Schema, SchemaGetter } from "effect" +import { Effect, FileSystem, Result, Schema, SchemaGetter, Stream } from "effect" import { Base64 } from "effect/encoding" import { HttpClientRequest } from "effect/http" import { ProviderID } from "./schema/ids.js" @@ -156,6 +156,13 @@ export class Asset { }) } + /** The payload as a byte stream; `url` sources stream from the network without being buffered or cached. */ + stream(): Stream.Stream { + const source = this.source + if (source.type !== "url" || this.#bytes !== undefined) return Stream.fromEffect(this.bytes()) + return Stream.unwrap(downloadStream(source, this.headers)) + } + base64(): Effect.Effect { return Effect.suspend(() => { const source = this.source @@ -255,6 +262,28 @@ const download = Effect.fn("Media.download")(function* ( return new Uint8Array(buffer) }) +const downloadStream = Effect.fn("Media.downloadStream")(function* ( + source: Extract, + headers: Record | undefined, +) { + const executor = yield* Service + const response = yield* executor.execute( + HttpClientRequest.get(source.url).pipe(HttpClientRequest.setHeaders(headers ?? {})), + ) + return response.stream.pipe( + Stream.mapError( + (cause) => + new AIError({ + reason: new InvalidProviderOutputError({ + message: `Failed to read media from ${source.url}`, + http: responseHttp(response), + cause, + }), + }), + ), + ) +}) + // --------------------------------------------------------------------------- // Constructors // --------------------------------------------------------------------------- @@ -344,12 +373,21 @@ export const file = (path: string, options?: AssetOptions): Effect.Effect => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem - const data = yield* asset.bytes() - yield* fs - .writeFile(path, data) - .pipe(Effect.mapError((cause) => invalid(`Failed to write media file ${path}`, cause))) + const temporary = `${path}.${crypto.randomUUID()}.tmp` + yield* asset.stream().pipe( + Stream.run(fs.sink(temporary)), + Effect.andThen(fs.rename(temporary, path)), + Effect.onError(() => fs.remove(temporary).pipe(Effect.ignore)), + Effect.mapError((cause) => + cause instanceof AIError ? cause : invalid(`Failed to write media file ${path}`, cause), + ), + ) }) diff --git a/packages/ai/src/protocols/xai-video.ts b/packages/ai/src/protocols/xai-video.ts index eda336f5891d..8a804543ffd0 100644 --- a/packages/ai/src/protocols/xai-video.ts +++ b/packages/ai/src/protocols/xai-video.ts @@ -5,7 +5,7 @@ import { Media } from "../media.js" import { MediaProtocol } from "../route/media-protocol.js" import { MediaRoute } from "../route/media.js" import { mergeJsonRecords } from "../schema/index.js" -import { VideoModel, VideoResponse, type VideoRequest } from "../video.js" +import { VideoModel, VideoResponse, seconds, type VideoRequest } from "../video.js" import { ProviderShared, optionalNull } from "./shared.js" const route = MediaProtocol.identity({ id: "xai-video", name: "xAI Video", provider: "xai" }) @@ -20,11 +20,13 @@ export const STATUS_PATH = "/videos" // --------------------------------------------------------------------------- /** - * Provider-native options. Common fields (`frames`, `references`, `video`, `durationSeconds`, `aspect`, - * `resolution`, `audio`) live on the request. `mode` selects the endpoint a `video` input is sent to. + * Provider-native options. Common fields (`start`, `end`, `references`, `duration`, `aspect`, `resolution`, + * `sound`) live on the request. `mode` sends the request to the edit or extension endpoint with a native + * `video: { url }` source; `reference_audios` takes preset `voice_id`s. */ export type XAIVideoOptions = { readonly mode?: "edit" | "extend" + readonly video?: { readonly url: string } | { readonly file_id: string } readonly reference_audios?: ReadonlyArray<{ readonly voice_id: string }> } & Record @@ -88,10 +90,22 @@ const nativeOptions = (options: XAIVideoOptions | undefined) => { } const fromRequest = Effect.fn("XAIVideo.fromRequest")(function* (request: Request) { - const image = request.frames?.first === undefined ? undefined : yield* mediaInput(request.frames.first) - const lastFrame = request.frames?.last === undefined ? undefined : yield* mediaInput(request.frames.last) - const video = request.video === undefined ? undefined : yield* mediaInput(request.video) - const references = yield* Effect.forEach(request.references ?? [], mediaInput) + const references = request.references ?? [] + if (references.some((asset) => asset.kind === "video")) + return yield* route.unsupported("video.references", `${route.name} has no reference video input`) + const image = request.start === undefined ? undefined : yield* mediaInput(request.start) + const lastFrame = request.end === undefined ? undefined : yield* mediaInput(request.end) + const images = yield* Effect.forEach( + references.filter((asset) => asset.kind !== "audio"), + mediaInput, + ) + const audios = yield* Effect.forEach( + references.filter((asset) => asset.kind === "audio"), + (asset) => + ProviderShared.mediaReference(asset, route.provider, route.name).pipe( + Effect.map((reference) => ({ url: reference.value })), + ), + ) return MediaProtocol.json( mergeJsonRecords( { @@ -99,12 +113,13 @@ const fromRequest = Effect.fn("XAIVideo.fromRequest")(function* (request: Reques prompt: request.prompt, image, last_frame: lastFrame, - reference_images: references.length === 0 ? undefined : references, - video, - duration: request.durationSeconds, - aspect_ratio: request.aspect, + reference_images: images.length === 0 ? undefined : images, + reference_audios: audios.length === 0 ? undefined : audios, + duration: request.duration === undefined ? undefined : seconds(request.duration), + // xAI follows the input image's shape when a start frame is set, which is what `auto` asks for. + aspect_ratio: request.aspect === "auto" && request.start !== undefined ? undefined : request.aspect, resolution: request.resolution, - generate_audio: request.audio, + generate_audio: request.sound, }, nativeOptions(request.providerOptions), request.http?.body, @@ -187,16 +202,16 @@ const statusPath = (token: Token) => `${STATUS_PATH}/${token.requestID}` export const protocol = MediaProtocol.queued(route, { token: Token, - unsupported: ["n", "seed", "negativePrompt"], start: { body: { from: fromRequest }, decode: decodeStart }, status: { path: statusPath, decode: decodeStatus }, result: { path: statusPath, decode: decodeResult }, }) -// A source video goes to `/videos/edits` unless `providerOptions.mode` asks for an extension. +// `providerOptions.mode` sends the native `video` source to the edit or extension endpoint. const startPath = (request: Request) => { - if (request.video === undefined) return PATH - return request.providerOptions?.mode === "extend" ? EXTEND_PATH : EDIT_PATH + if (request.providerOptions?.mode === "edit") return EDIT_PATH + if (request.providerOptions?.mode === "extend") return EXTEND_PATH + return PATH } export const model = (input: MediaRoute.ModelInput) => diff --git a/packages/ai/src/public.ts b/packages/ai/src/public.ts index d2af57d227cd..d38f6a6046e8 100644 --- a/packages/ai/src/public.ts +++ b/packages/ai/src/public.ts @@ -116,6 +116,17 @@ export type { ImageResolution, ImageRoute, } from "./image.js" +export { VideoAspectRatio, VideoDuration, VideoModel, VideoModelSchema, VideoResponse } from "./video.js" +export type { + VideoJob, + VideoModelOptions, + VideoOptions, + VideoRequest, + VideoRequestInput, + VideoResolution, + VideoRoute, + VideoStatus, +} from "./video.js" export { ChoiceAnswer, ChoiceQuestion, diff --git a/packages/ai/src/route/media.ts b/packages/ai/src/route/media.ts index 200635caefb5..68597abf1df5 100644 --- a/packages/ai/src/route/media.ts +++ b/packages/ai/src/route/media.ts @@ -78,6 +78,12 @@ export interface QueuedRoute { token: unknown, execute: Execute, ) => Effect.Effect, AIError> + /** The status, result and cancel operations for a persisted token, without reading its status first. */ + readonly attach: ( + model: MediaRequest["model"], + token: unknown, + execute: Execute, + ) => Effect.Effect["route"], AIError> } /** One request whose response parses into events; `generate` runs the same stream and collects it. */ @@ -219,12 +225,8 @@ export const queued = ( return new Generation(route, encodeToken(started.token), started.snapshot) }) - const resume = Effect.fn("MediaRoute.resume")(function* ( - model: MediaRequest["model"], - raw: unknown, - execute: Execute, - ) { - const token = yield* decodeToken(raw).pipe( + const decode = (raw: unknown) => + decodeToken(raw).pipe( Effect.mapError( (cause) => new AIError({ @@ -235,11 +237,21 @@ export const queued = ( }), ), ) + + const attach = (model: MediaRequest["model"], raw: unknown, execute: Execute) => + decode(raw).pipe(Effect.map((token) => generationRoute(token, transport.exchange(model), execute))) + + const resume = Effect.fn("MediaRoute.resume")(function* ( + model: MediaRequest["model"], + raw: unknown, + execute: Execute, + ) { + const token = yield* decode(raw) const route = generationRoute(token, transport.exchange(model), execute) return new Generation(route, encodeToken(token), yield* route.status) }) - return { kind: "queued", id: protocol.id, provider: protocol.provider, protocol: protocol.id, start, resume } + return { kind: "queued", id: protocol.id, provider: protocol.provider, protocol: protocol.id, start, resume, attach } } /** Compose a streaming media protocol; `generate` runs the same stream in `generate` mode and folds it with `collect`. */ diff --git a/packages/ai/src/video-client.ts b/packages/ai/src/video-client.ts index 5369167fecf9..960677c86f96 100644 --- a/packages/ai/src/video-client.ts +++ b/packages/ai/src/video-client.ts @@ -1,20 +1,61 @@ -import { Context } from "effect" -import { MediaClient } from "./media-client.js" -import type { VideoEvent, VideoRequest, VideoResponse } from "./video.js" +import { Context, Effect, Layer } from "effect" +import { RequestExecutor } from "./route/executor.js" +import { AIError, UnsupportedOperationError } from "./schema/index.js" +import type { VideoJob, VideoModel, VideoRequest, VideoResponse, VideoStatus } from "./video.js" -export type Interface = MediaClient.Interface +export interface Interface { + readonly start: (request: VideoRequest) => Effect.Effect + readonly status: (model: VideoModel, token: unknown) => Effect.Effect + readonly result: (model: VideoModel, token: unknown) => Effect.Effect + readonly cancel: (model: VideoModel, token: unknown) => Effect.Effect +} export class VideoClientService extends Context.Service()("@opencode/VideoClient") {} export const Service = VideoClientService export type Service = VideoClientService +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const executor = yield* RequestExecutor.Service + const attach = (model: VideoModel, token: unknown) => model.route.attach(model, token, executor.execute) + return Service.of({ + start: (request) => + request.model.route + .start(request, executor.execute) + .pipe(Effect.map((generation) => ({ id: generation.id, token: generation.token }))), + status: (model, token) => + attach(model, token).pipe( + Effect.flatMap((route) => route.status), + Effect.map((snapshot) => ({ status: snapshot.status, progress: snapshot.progress })), + ), + result: (model, token) => attach(model, token).pipe(Effect.flatMap((route) => route.result)), + cancel: (model, token) => + attach(model, token).pipe( + Effect.flatMap( + (route) => + route.cancel ?? + Effect.fail( + new AIError({ + reason: new UnsupportedOperationError({ + operation: "video.cancel", + provider: model.provider, + route: model.route.id, + message: `${model.provider}/${model.route.id} has no way to cancel a video generation`, + }), + }), + ), + ), + ), + }) + }), +) + export const VideoClient = { Service, - ...MediaClient.make(Service, { - modality: "video", - responseEvents: (response: VideoResponse) => [ - ...response.videos.map((video, index) => ({ type: "video" as const, index, video })), - { type: "finish" as const, usage: response.usage, providerMetadata: response.providerMetadata }, - ], - }), + layer, + start: (request: VideoRequest) => Service.use((client) => client.start(request)), + status: (model: VideoModel, token: unknown) => Service.use((client) => client.status(model, token)), + result: (model: VideoModel, token: unknown) => Service.use((client) => client.result(model, token)), + cancel: (model: VideoModel, token: unknown) => Service.use((client) => client.cancel(model, token)), } as const diff --git a/packages/ai/src/video.ts b/packages/ai/src/video.ts index b26c14e9e043..d97f4fae2d8f 100644 --- a/packages/ai/src/video.ts +++ b/packages/ai/src/video.ts @@ -1,7 +1,7 @@ -import { Effect, Schema, Stream } from "effect" -import { Generation, ProgressEvent, QueuedEvent, type AwaitOptions } from "./generation.js" +import { Duration, Effect, Schema } from "effect" +import type { Status } from "./generation.js" import { Media } from "./media.js" -import { MediaModel, composeRoute, tryRequest } from "./media-model.js" +import { MediaModel, composeQueuedRoute, tryRequest } from "./media-model.js" import { MediaRoute } from "./route/media.js" import { AIError, @@ -19,34 +19,22 @@ import { VideoClient, Service } from "./video-client.js" export type VideoOptions = MediaModel.Options -export type VideoRoute = MediaRoute.AnyRoute +/** Video generation is always a queued job: submit, then read its status and result by token. */ +export type VideoRoute = MediaRoute.QueuedRoute export class VideoModel extends MediaModel { declare protected readonly _VideoModel: void - /** The number of type arguments selects the kind: ``, ``, or ``. */ - static fromRoute( - route: MediaModel.InlineRouteInput>, VideoResponse>, - input: MediaRoute.ModelInput, - ): VideoModel - static fromRoute( - route: MediaModel.StreamRouteInput>, VideoEvent, Frame, State>, - input: MediaRoute.ModelInput, - ): VideoModel static fromRoute( route: MediaModel.QueuedRouteInput>, VideoResponse, Token>, input: MediaRoute.ModelInput, - ): VideoModel - static fromRoute( - route: MediaModel.AnyRouteInput>, VideoEvent, VideoResponse, Frame, State, Token>, - input: MediaRoute.ModelInput, ) { return new VideoModel({ id: input.id, provider: route.protocol.provider, http: input.http, providerOptions: input.providerOptions, - route: composeRoute(route, input, collectResponse) as VideoRoute, + route: composeQueuedRoute(route, input) as VideoRoute, }) } } @@ -59,38 +47,38 @@ export const VideoModelSchema = Schema.declare((value): value is VideoModel => v // Request // --------------------------------------------------------------------------- -export type VideoAspectRatio = Media.AspectRatio -export const VideoAspectRatio = Media.AspectRatio +/** A `width:height` ratio, or `auto` to follow the input image. */ +export type VideoAspectRatio = Media.AspectRatio | "auto" +export const VideoAspectRatio = Schema.Union([Media.AspectRatio, Schema.Literal("auto")]) +/** Common output resolutions; protocols convert them to their provider's spelling and pass other values through. */ export type VideoResolution = OpenString<"480p" | "720p" | "1080p" | "4k"> -/** Pinned frames. Routes that accept only a first frame fail typed when `last` is present. */ -export const VideoFrames = Schema.Struct({ - first: Schema.optional(Media.AssetSchema), - last: Schema.optional(Media.AssetSchema), -}).annotate({ identifier: "Video.Frames" }) -export type VideoFrames = Schema.Schema.Type +/** Output length as an Effect duration string in seconds, such as `"8 seconds"`. */ +export type VideoDuration = `${number} second` | `${number} seconds` +export const VideoDuration = Schema.declare( + (value): value is VideoDuration => typeof value === "string" && /^\d+(?:\.\d+)? seconds?$/.test(value), + { title: "Video.Duration" }, +) const VideoRequestSchema = Schema.Struct({ model: VideoModelSchema, prompt: Schema.String, - frames: Schema.optional(VideoFrames), - /** Style or subject references that guide the output without pinning a frame. */ + /** Image the video starts on exactly. */ + start: Schema.optional(Media.AssetSchema), + /** Image the video ends on exactly. */ + end: Schema.optional(Media.AssetSchema), + /** Images, videos or audio that guide the subject or style without pinning a frame. */ references: Schema.optional(Schema.Array(Media.AssetSchema)), - /** Source video for edit or extension routes. */ - video: Schema.optional(Media.AssetSchema), - durationSeconds: Schema.optional(Schema.Number), + duration: Schema.optional(VideoDuration), aspect: Schema.optional(VideoAspectRatio), resolution: Schema.optional(Schema.String), - /** Whether to generate an audio track; routes whose audio is always on fail typed on `false`. */ - audio: Schema.optional(Schema.Boolean), - n: Schema.optional(Schema.Int), - seed: Schema.optional(Schema.Number), - negativePrompt: Schema.optional(Schema.String), + /** Whether the video has a generated soundtrack; unset leaves the model's default. */ + sound: Schema.optional(Schema.Boolean), providerOptions: Schema.optional(ProviderOptions), }) -/** A normalized video request, the plain object `Video.request(...)` returns. */ +/** A normalized video request, the plain object `ai.video.request(...)` returns. */ export type VideoRequest = Omit< typeof VideoRequestSchema.Type, "model" | "providerOptions" | "resolution" @@ -99,15 +87,36 @@ export type VideoRequest = Omit< export type VideoModelOptions = Model extends VideoModel ? Options : never -/** Input accepted by `Video.request` and the calls; a `VideoRequest` is also a valid input. */ -export type VideoRequestInput = Omit, "providerOptions"> & { +/** Input accepted by `ai.video.request` and `ai.video.start`; a `VideoRequest` is also a valid input. */ +export type VideoRequestInput = Omit< + VideoRequest, + "start" | "end" | "references" | "providerOptions" +> & { + readonly start?: Media.Input + readonly end?: Media.Input + readonly references?: ReadonlyArray readonly providerOptions?: NoInfer> } +/** Seconds in a request's `duration`, for protocols that send a number. */ +export const seconds = (duration: VideoDuration) => Duration.toSeconds(Duration.fromInputUnsafe(duration)) + // --------------------------------------------------------------------------- -// Response and events +// Job, status and response // --------------------------------------------------------------------------- +/** A submitted generation. Store `token` (plain JSON) with the model to read the job later, from any process. */ +export interface VideoJob { + readonly id: string + readonly token: unknown +} + +export interface VideoStatus { + readonly status: Status + /** 0..1, only when the provider reports progress. */ + readonly progress?: number +} + export class VideoResponse extends Schema.Class("Video.Response")({ videos: Schema.Array(Media.AssetSchema), usage: Schema.optional(MediaUsage), @@ -118,51 +127,16 @@ export class VideoResponse extends Schema.Class("Video.Response") } } -export const VideoOutputEvent = Schema.Struct({ - type: Schema.tag("video"), - index: Schema.Number, - video: Media.AssetSchema, -}).annotate({ identifier: "Video.Event.Video" }) - -export const VideoFinishEvent = Schema.Struct({ - type: Schema.tag("finish"), - usage: Schema.optional(MediaUsage), - providerMetadata: Schema.optional(ProviderMetadata), -}).annotate({ identifier: "Video.Event.Finish" }) - -const videoEventTagged = Schema.Union([QueuedEvent, ProgressEvent, VideoOutputEvent, VideoFinishEvent]).pipe( - Schema.toTaggedUnion("type"), -) -export const VideoEvent = Object.assign(videoEventTagged, { - is: { - generationQueued: videoEventTagged.guards["generation-queued"], - generationProgress: videoEventTagged.guards["generation-progress"], - video: videoEventTagged.guards.video, - finish: videoEventTagged.guards.finish, - }, -}) -export type VideoEvent = Schema.Schema.Type - -const collectResponse = (events: ReadonlyArray): Effect.Effect => { - const finish = events.find(VideoEvent.is.finish) - // A streaming video protocol's `finish` emits the terminal event or fails, so a completed stream always has one. - if (finish === undefined) return Effect.die(new Error("The video stream completed without a finish event")) - return Effect.succeed( - new VideoResponse({ - videos: events.filter(VideoEvent.is.video).map((event) => event.video), - usage: finish.usage, - providerMetadata: finish.providerMetadata, - }), - ) -} - // --------------------------------------------------------------------------- -// Request-shaped call API +// Call API // --------------------------------------------------------------------------- /** Normalize video request input into a plain `VideoRequest`. */ export const request = (input: VideoRequestInput): VideoRequest => ({ ...input, + start: input.start === undefined ? undefined : Media.make(input.start), + end: input.end === undefined ? undefined : Media.make(input.end), + references: input.references?.map((reference) => Media.make(reference)), }) const validate = Schema.decodeUnknownSync(VideoRequestSchema) @@ -174,37 +148,30 @@ const requestEffect = (input: VideoRequestInput) => return normalized }) +/** Submit a generation and return at once; nothing polls. */ export function start( input: VideoRequestInput, -): Effect.Effect, AIError, Service> +): Effect.Effect export function start(input: VideoRequestInput) { return requestEffect(input).pipe(Effect.flatMap((request) => VideoClient.start(request))) } -export function generate( - input: VideoRequestInput, - options?: AwaitOptions, -): Effect.Effect -export function generate(input: VideoRequestInput, options?: AwaitOptions) { - return requestEffect(input).pipe(Effect.flatMap((request) => VideoClient.generate(request, options))) -} +/** One status read for a stored token. */ +export const status = (model: VideoModel, token: unknown): Effect.Effect => + VideoClient.status(model, token) -/** Rebuild a generation handle from a persisted `Generation.token`, refreshing its status once. */ -export const resume = (model: VideoModel, token: unknown): Effect.Effect, AIError, Service> => - VideoClient.resume(model, token) +/** The finished videos; fails while the job is still running, or with the provider's reason when it failed. */ +export const result = (model: VideoModel, token: unknown): Effect.Effect => + VideoClient.result(model, token) -export function stream( - input: VideoRequestInput, - options?: AwaitOptions, -): Stream.Stream -export function stream(input: VideoRequestInput, options?: AwaitOptions) { - return Stream.unwrap(requestEffect(input).pipe(Effect.map((request) => VideoClient.stream(request, options)))) -} +/** Ask the provider to stop the job; fails with `UnsupportedOperation` when the provider has no cancel call. */ +export const cancel = (model: VideoModel, token: unknown): Effect.Effect => + VideoClient.cancel(model, token) export const Video = { request, start, - generate, - resume, - stream, + status, + result, + cancel, } as const diff --git a/packages/ai/test/exports.test.ts b/packages/ai/test/exports.test.ts index 700c5df4af65..b72999baf754 100644 --- a/packages/ai/test/exports.test.ts +++ b/packages/ai/test/exports.test.ts @@ -8,6 +8,7 @@ import { EmbeddingClient, Generation, ImageClient, + VideoClient, LanguageModel, LLMClient, LLMEvent, @@ -15,7 +16,7 @@ import { RawEvent, } from "@opencode/ai/effect" import { AI } from "@opencode/ai" -import { Speech, SpeechClient, SpeechEvent, TranscriptionClient, Video, VideoClient } from "@opencode/ai/experimental" +import { Speech, SpeechClient, SpeechEvent, TranscriptionClient } from "@opencode/ai/experimental" import { Route, Protocol, RequestExecutor, WebSocketTransport } from "@opencode/ai/route" import { Alibaba, @@ -74,9 +75,12 @@ describe("public exports", () => { expect(LanguageModel.make).toBeFunction() expect(Media.bytes).toBeFunction() expect(ai.image.generate).toBeFunction() - expect(Video.start).toBeFunction() - expect(Video.resume).toBeFunction() + expect(ai.video.start).toBeFunction() + expect(ai.video.status).toBeFunction() + expect(ai.video.result).toBeFunction() + expect(ai.video.cancel).toBeFunction() expect(VideoClient.layer).toBeDefined() + expect(AI.ai.video.start).toBeFunction() expect(Speech.generate).toBeFunction() expect(Speech.stream).toBeFunction() expect(SpeechClient.layer).toBeDefined() diff --git a/packages/ai/test/media.test.ts b/packages/ai/test/media.test.ts index dbf92d54f679..ac8b6bd2a90a 100644 --- a/packages/ai/test/media.test.ts +++ b/packages/ai/test/media.test.ts @@ -310,4 +310,31 @@ describe("Media", () => { expect(missing.reason._tag).toBe("InvalidRequest") }).pipe(Effect.provide(NodeFileSystem.layer), Effect.provide(scriptedResponses(["unused"]))), ) + + it.effect("streams a URL download into place and leaves an existing file alone when the download fails", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem + const dir = yield* fs.makeTempDirectoryScoped() + const target = `${dir}/clip.mp4` + + yield* Media.write(Media.url("https://cdn.example.test/ok.mp4"), target) + expect(yield* fs.readFile(target)).toEqual(Uint8Array.from([7, 8, 9])) + + const error = yield* Media.write(Media.url("https://cdn.example.test/expired.mp4"), target).pipe(Effect.flip) + expect(error.reason.http?.status).toBe(404) + expect(yield* fs.readFile(target)).toEqual(Uint8Array.from([7, 8, 9])) + expect(yield* fs.readDirectory(dir)).toEqual(["clip.mp4"]) + }).pipe( + Effect.provide(NodeFileSystem.layer), + Effect.provide( + dynamicResponse((input) => + Effect.succeed( + input.request.url.endsWith("/ok.mp4") + ? input.respond(Uint8Array.from([7, 8, 9]), { headers: { "content-type": "video/mp4" } }) + : input.respond("file expired", { status: 404 }), + ), + ), + ), + ), + ) }) diff --git a/packages/ai/test/promise-parity.types.ts b/packages/ai/test/promise-parity.types.ts index 365c2071d3b6..ce0b1027dc30 100644 --- a/packages/ai/test/promise-parity.types.ts +++ b/packages/ai/test/promise-parity.types.ts @@ -12,7 +12,7 @@ type Assert> = T type PromiseAI = Omit type EffectAI = typeof ai type Helpers = "run" | "bytes" | "base64" | "materialize" | "file" | "write" -type Namespaces = "image" | "decision" | "embedding" | "experimental" +type Namespaces = "image" | "decision" | "embedding" | "video" | "experimental" type Signatures = F extends { (...args: infer A1): infer R1 @@ -49,7 +49,11 @@ type Promised = : R type EffectShape = - Signatures extends infer S ? (S extends [infer A, infer R] ? [Positional, Options, Promised] : never) : never + Signatures extends infer S + ? S extends [infer A, infer R] + ? [Positional, Options, Promised] + : never + : never type PromiseShape = Signatures extends infer S ? (S extends [infer A, infer R] ? [Positional, Options, R] : never) : never @@ -61,7 +65,9 @@ export type Drift = { : "missing in Promise" } -type Same = [Exclude | Exclude] extends [never] ? true : { readonly drift: Exclude | Exclude } +type Same = [Exclude | Exclude] extends [never] + ? true + : { readonly drift: Exclude | Exclude } export type Checks = [ Assert<{ handle: Same, Exclude, "route">> }>, @@ -71,7 +77,7 @@ export type Checks = [ Assert>, Assert>, Assert>, + Assert>, Assert>, Assert>, - Assert>, ] diff --git a/packages/ai/test/promise.test.ts b/packages/ai/test/promise.test.ts index 8eccf54528aa..da8af3e306d6 100644 --- a/packages/ai/test/promise.test.ts +++ b/packages/ai/test/promise.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test" -import { mkdtemp, rm } from "node:fs/promises" +import { mkdtemp, readdir, rm } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import { Effect, Layer, Schema } from "effect" @@ -15,6 +15,10 @@ import { sseEvents } from "./lib/sse.js" const openai = OpenAI.configure({ apiKey: "test", baseURL: "https://openai.test/v1" }) const xai = XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }) +const assemblyai = AssemblyAI.configure({ apiKey: "test", baseURL: "https://assemblyai.test" }).transcription( + "universal-2", +) +const audio = Media.url("https://audio.test/hello.mp3") const chatBody = sseEvents( { choices: [{ delta: { content: "Hello" } }] }, @@ -25,7 +29,7 @@ const chatBody = sseEvents( /** * Executor layer that answers chat completions with SSE text, image generations with one base64 PNG, xAI video * requests with a pending submission that succeeds on the second poll, speech with raw audio or SSE audio deltas, OpenAI - * transcription with JSON or SSE text deltas, AssemblyAI transcripts that complete on the first poll, and `slow.test` + * transcription with JSON or SSE text deltas, AssemblyAI transcripts that complete on the second poll, and `slow.test` * chat completions that send one text delta and never finish. */ const executor = (seen: Array) => @@ -94,10 +98,17 @@ const executor = (seen: Array) => return input.respond(JSON.stringify({ id: "tr_1", status: "queued" }), { headers: { "content-type": "application/json" }, }) - if (web.url.endsWith("/v2/transcript/tr_1")) - return input.respond(JSON.stringify({ id: "tr_1", status: "completed", text: "Hello there." }), { - headers: { "content-type": "application/json" }, - }) + if (web.url.endsWith("/v2/transcript/tr_1")) { + const polls = seen.filter((url) => url.endsWith("/v2/transcript/tr_1")).length + return input.respond( + JSON.stringify( + polls === 1 + ? { id: "tr_1", status: "processing" } + : { id: "tr_1", status: "completed", text: "Hello there." }, + ), + { headers: { "content-type": "application/json" } }, + ) + } if (web.url.endsWith("/videos/generations")) return input.respond(JSON.stringify({ request_id: "req_1" }), { headers: { "content-type": "application/json" }, @@ -171,36 +182,30 @@ describe("AI promise client", () => { await ai.dispose() }) - test("starts, awaits, and resumes video generations over the same runtime", async () => { + test("starts video jobs and reads their status and result by token over the same runtime", async () => { const seen: Array = [] const ai = AI.make({ layer: executor(seen) }) const model = xai.video("grok-imagine-video-1.5") - const generation = await ai.experimental.video.start({ model, prompt: "A kite" }) - expect(generation.id).toBe("req_1") - expect(generation.status).toBe("running") - expect(generation.token).toEqual({ requestID: "req_1" }) + const job = await ai.video.start({ model, prompt: "A kite" }) + expect(job).toEqual({ id: "req_1", token: { requestID: "req_1" } }) - const refreshed = await generation.refresh() - expect(refreshed.status).toBe("running") - expect(refreshed.progress).toBe(0.5) - - const response = await refreshed.await({ poll: { interval: 10 } }) + const token: unknown = JSON.parse(JSON.stringify(job.token)) + expect(await ai.video.status(model, token)).toEqual({ status: "running", progress: 0.5 }) + expect(await ai.video.status(model, token)).toEqual({ status: "completed" }) + const response = await ai.video.result(model, token) expect(response.video).toBeInstanceOf(Media.Asset) expect(response.video.source).toMatchObject({ type: "url", url: "https://vidgen.x.ai/out.mp4" }) + expect(await ai.video.cancel(model, token).catch((error: unknown) => error)).toMatchObject({ + reason: { _tag: "UnsupportedOperation" }, + }) - const resumed = await ai.experimental.video.resume(model, JSON.parse(JSON.stringify(generation.token))) - expect(resumed.status).toBe("completed") - expect((await resumed.await()).videos).toHaveLength(1) - - const events: Array = [] - for await (const event of ai.experimental.video.stream({ model, prompt: "A kite" }, { poll: { interval: 10 } })) { - events.push(event.type) - } - expect(events).toEqual(["video", "finish"]) - - expect(seen[0]).toBe("https://xai.test/v1/videos/generations") - expect(seen.filter((url) => url.endsWith("/videos/req_1")).length).toBeGreaterThanOrEqual(5) + expect(seen).toEqual([ + "https://xai.test/v1/videos/generations", + "https://xai.test/v1/videos/req_1", + "https://xai.test/v1/videos/req_1", + "https://xai.test/v1/videos/req_1", + ]) await ai.dispose() }) @@ -229,11 +234,8 @@ describe("AI promise client", () => { test("exposes a generation's snapshot and terminal state", async () => { const ai = AI.make({ layer: executor([]) }) - const generation = await ai.experimental.video.start({ - model: xai.video("grok-imagine-video-1.5"), - prompt: "A kite", - }) - expect(generation.snapshot).toEqual({ id: "req_1", status: "running" }) + const generation = await ai.experimental.transcription.start({ model: assemblyai, audio }) + expect(generation.snapshot).toEqual({ id: "tr_1", status: "queued" }) expect(generation.terminal).toBe(false) const refreshed = await generation.refresh() expect(refreshed.terminal).toBe(false) @@ -243,16 +245,13 @@ describe("AI promise client", () => { test("observes a started generation's events and fetches its result", async () => { const ai = AI.make({ layer: executor([]) }) - const generation = await ai.experimental.video.start({ - model: xai.video("grok-imagine-video-1.5"), - prompt: "A kite", - }) + const generation = await ai.experimental.transcription.start({ model: assemblyai, audio }) const events: Array = [] for await (const event of generation.events({ poll: { interval: 10 } })) events.push(event.type) expect(events).toEqual(["generation-progress", "generation-finished"]) - expect(generation.status).toBe("running") - expect((await generation.result()).video.source).toMatchObject({ url: "https://vidgen.x.ai/out.mp4" }) + expect(generation.status).toBe("queued") + expect((await generation.result()).text).toBe("Hello there.") await ai.dispose() }) @@ -267,6 +266,14 @@ describe("AI promise client", () => { expect(asset.mediaType).toBe("image/png") await ai.write(asset, join(dir, "copy.png")) expect(await ai.bytes(await ai.file(join(dir, "copy.png")))).toEqual(png) + + // A failed download leaves the existing file as it was and cleans up its temporary file. + const failed = await ai + .write(Media.url("https://cdn.test/expired.png"), join(dir, "copy.png")) + .catch((error: unknown) => error) + expect(failed instanceof AIError && failed.reason.http?.status).toBe(404) + expect(await ai.bytes(await ai.file(join(dir, "copy.png")))).toEqual(png) + expect((await readdir(dir)).sort()).toEqual(["copy.png", "source.bin"]) } finally { await rm(dir, { recursive: true }) await ai.dispose() @@ -392,10 +399,7 @@ describe("AI promise client", () => { expect(midStreamFailure).toBe(midStream.signal.reason) expect(midStreamFailure).toMatchObject({ name: "AbortError" }) - const generation = await ai.experimental.video.start({ - model: xai.video("grok-imagine-video-1.5"), - prompt: "A kite", - }) + const generation = await ai.experimental.transcription.start({ model: assemblyai, audio }) const polling = new AbortController() const events: Array = [] const eventsFailure = await Array.fromAsync( @@ -598,7 +602,7 @@ describe("AI promise client", () => { test("the default client is created lazily", () => { expect(typeof AI.ai.generate).toBe("function") expect(typeof AI.ai.image.generate).toBe("function") - expect(typeof AI.ai.experimental.video.start).toBe("function") + expect(typeof AI.ai.video.start).toBe("function") expect(typeof AI.ai.experimental.speech.generate).toBe("function") }) }) diff --git a/packages/ai/test/provider/xai-video.recorded.test.ts b/packages/ai/test/provider/xai-video.recorded.test.ts index 31cad80d0cd2..8b6fe07e1f09 100644 --- a/packages/ai/test/provider/xai-video.recorded.test.ts +++ b/packages/ai/test/provider/xai-video.recorded.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" -import { Effect, Stream } from "effect" -import { Video } from "../../src/experimental.js" +import { Effect, Schedule } from "effect" +import { ai, type VideoStatus } from "../../src/effect.js" import { XAI } from "../../src/providers.js" import { recordedTests } from "../recorded-test.js" import { queuedPoll } from "./queued-recording.js" @@ -21,23 +21,29 @@ describe("xAI Video recorded", () => { "generates a video", () => Effect.gen(function* () { - const generation = yield* Video.start({ + const job = yield* ai.video.start({ model, prompt: "A single red balloon drifting slowly upward against a clear blue sky.", aspect: "16:9", resolution: "480p", - durationSeconds: 2, + duration: "2 seconds", }) - const observed = yield* generation.events({ poll: queuedPoll }).pipe(Stream.runCollect) - const progress = Array.from(observed).flatMap((event) => - event.type === "generation-progress" && event.progress !== undefined ? [event.progress] : [], + const observed: Array = [] + yield* ai.video.status(model, job.token).pipe( + Effect.tap((status) => Effect.sync(() => observed.push(status))), + Effect.repeat({ + schedule: Schedule.spaced(queuedPoll.interval ?? "10 millis"), + until: (status) => status.status !== "queued" && status.status !== "running", + }), ) + const progress = observed.flatMap((status) => (status.progress === undefined ? [] : [status.progress])) // xAI reports progress as a percentage; the route normalizes it to a monotonic 0..1 fraction. expect(progress.length).toBeGreaterThan(0) expect(progress.every((value) => value >= 0 && value <= 1)).toBe(true) expect(progress).toEqual([...progress].sort((a, b) => a - b)) + expect(observed.at(-1)?.status).toBe("completed") - const response = yield* generation.result() + const response = yield* ai.video.result(model, job.token) expect(response.videos).toHaveLength(1) expect(response.video.source.type).toBe("url") expect(response.video.info?.durationSeconds).toBe(2) diff --git a/packages/ai/test/video.test.ts b/packages/ai/test/video.test.ts index e0eec6711614..8e78d2e8e03d 100644 --- a/packages/ai/test/video.test.ts +++ b/packages/ai/test/video.test.ts @@ -1,8 +1,7 @@ import { describe, expect } from "bun:test" import { Effect, Fiber, Layer, Stream } from "effect" import * as TestClock from "effect/testing/TestClock" -import { Media, type GenerationEvent } from "../src/effect.js" -import { Video, VideoClient, type VideoEvent } from "../src/experimental.js" +import { Media, VideoClient, ai } from "../src/effect.js" import { XAI } from "../src/providers.js" import { it } from "./lib/effect.js" import { dynamicResponse, json, observe, settle, type Call, type HandlerInput } from "./lib/http.js" @@ -10,94 +9,90 @@ import { dynamicResponse, json, observe, settle, type Call, type HandlerInput } const layer = (handler: Parameters[0]) => VideoClient.layer.pipe(Layer.provideMerge(dynamicResponse(handler))) +const xai = XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }) +const model = xai.video("grok-imagine-video-1.5") +const done = { status: "done", video: { url: "https://vidgen.x.ai/out.mp4" } } + // --------------------------------------------------------------------------- -// xAI +// start // --------------------------------------------------------------------------- -describe("Video / xAI", () => { - const xai = XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }) - const model = xai.video("grok-imagine-video-1.5") - - it.effect("submits a generation, reports pending progress, and returns the temporary URL", () => +describe("ai.video.start / xAI", () => { + it.effect("maps the request onto xAI's body and returns the job without polling", () => Effect.gen(function* () { const calls: Array = [] - const events: Array = [] - const program = Effect.gen(function* () { - const generation = yield* Video.start({ + const job = yield* ai.video + .start({ model, prompt: "Make the water crash down", - frames: { - first: Media.fromDataUrl("data:image/png;base64,AQID"), - last: Media.url("https://example.test/last.png"), - }, - references: [Media.ref("xai", "file_1")], - durationSeconds: 10, + start: Media.fromDataUrl("data:image/png;base64,AQID"), + end: "https://example.test/last.png", + references: [ + Media.ref("xai", "file_1"), + Media.url("https://example.test/voice.mp3", { mediaType: "audio/mpeg" }), + ], + duration: "10 seconds", aspect: "16:9", resolution: "720p", - audio: false, - providerOptions: { reference_audios: [{ voice_id: "eve" }], future_option: true }, + sound: false, + providerOptions: { future_option: true }, headers: { "x-request": "yes" }, http: { query: { trace: "1" } }, }) - expect(generation.token).toEqual({ requestID: "req_1" }) - events.push(...(yield* generation.events({ poll: { interval: "1 second" } }).pipe(Stream.runCollect))) - return yield* generation.await() - }) - const response = yield* settle(program, 2).pipe( - Effect.provide( - layer((input) => - Effect.gen(function* () { - const { call, nth } = yield* observe(calls, input) - expect(call.headers.get("authorization")).toBe("Bearer test") - // The request's own `http` overlay follows the generation into every poll started from it. - expect(call.headers.get("x-request")).toBe("yes") - if (call.method === "POST") { - expect(call.url).toBe("https://xai.test/v1/videos/generations?trace=1") - expect(JSON.parse(call.body)).toEqual({ - model: "grok-imagine-video-1.5", - prompt: "Make the water crash down", - image: { url: "data:image/png;base64,AQID" }, - last_frame: { url: "https://example.test/last.png" }, - reference_images: [{ file_id: "file_1" }], - duration: 10, - aspect_ratio: "16:9", - resolution: "720p", - generate_audio: false, - reference_audios: [{ voice_id: "eve" }], - future_option: true, - }) + .pipe( + Effect.provide( + layer((input) => + Effect.gen(function* () { + const { call } = yield* observe(calls, input) + expect(call.headers.get("authorization")).toBe("Bearer test") + expect(call.headers.get("x-request")).toBe("yes") return json(input, { request_id: "req_1" }) - } - expect(call.url).toBe("https://xai.test/v1/videos/req_1?trace=1") - if (nth === 1) return json(input, { status: "pending", progress: 40 }) - return json(input, { - status: "done", - video: { url: "https://vidgen.x.ai/out.mp4", duration: 10, respect_moderation: true }, - model: "grok-imagine-video-1.5", - }) - }), + }), + ), ), - ), - ) - expect(events).toEqual([ - { type: "generation-progress", id: "req_1", progress: 0.4 }, - { type: "generation-finished", id: "req_1", status: "completed" }, + ) + expect(job).toEqual({ id: "req_1", token: { requestID: "req_1" } }) + expect(calls.map((call) => `${call.method} ${call.url}`)).toEqual([ + "POST https://xai.test/v1/videos/generations?trace=1", ]) - expect(response.video.source).toEqual({ type: "url", url: "https://vidgen.x.ai/out.mp4", mediaType: "video/mp4" }) - expect(response.video.info).toEqual({ durationSeconds: 10 }) - expect(response.providerMetadata).toEqual({ - xai: { requestId: "req_1", model: "grok-imagine-video-1.5", respectModeration: true }, + expect(JSON.parse(calls[0].body)).toEqual({ + model: "grok-imagine-video-1.5", + prompt: "Make the water crash down", + image: { url: "data:image/png;base64,AQID" }, + last_frame: { url: "https://example.test/last.png" }, + reference_images: [{ file_id: "file_1" }], + reference_audios: [{ url: "https://example.test/voice.mp3" }], + duration: 10, + aspect_ratio: "16:9", + resolution: "720p", + generate_audio: false, + future_option: true, }) }), ) - it.effect("routes a source video to edits by default and to extensions on request", () => + it.effect("leaves aspect out for auto when a start image sets the shape, and sends it otherwise", () => Effect.gen(function* () { const calls: Array = [] - const source = Media.url("https://example.test/in.mp4") yield* Effect.gen(function* () { - yield* Video.start({ model, prompt: "brighter", video: source }) - yield* Video.start({ model, prompt: "keep going", video: source, providerOptions: { mode: "extend" } }) + yield* ai.video.start({ model, prompt: "x", start: "https://example.test/first.png", aspect: "auto" }) + yield* ai.video.start({ model, prompt: "x", aspect: "auto" }) + }).pipe( + Effect.provide( + layer((input) => observe(calls, input).pipe(Effect.map(() => json(input, { request_id: "req_1" })))), + ), + ) + expect(calls.map((call) => JSON.parse(call.body).aspect_ratio)).toEqual([undefined, "auto"]) + }), + ) + + it.effect("sends a native source video to edits or extensions through providerOptions.mode", () => + Effect.gen(function* () { + const calls: Array = [] + const video = { url: "https://example.test/in.mp4" } + yield* Effect.gen(function* () { + yield* ai.video.start({ model, prompt: "brighter", providerOptions: { mode: "edit", video } }) + yield* ai.video.start({ model, prompt: "keep going", providerOptions: { mode: "extend", video } }) }).pipe( Effect.provide( layer((input) => observe(calls, input).pipe(Effect.map(() => json(input, { request_id: "req_2" })))), @@ -108,12 +103,132 @@ describe("Video / xAI", () => { "https://xai.test/v1/videos/extensions", ]) expect(calls.map((call) => JSON.parse(call.body))).toEqual([ - { model: "grok-imagine-video-1.5", prompt: "brighter", video: { url: "https://example.test/in.mp4" } }, - { model: "grok-imagine-video-1.5", prompt: "keep going", video: { url: "https://example.test/in.mp4" } }, + { model: "grok-imagine-video-1.5", prompt: "brighter", video }, + { model: "grok-imagine-video-1.5", prompt: "keep going", video }, ]) }), ) + it.effect("rejects reference videos, which xAI has no field for, before sending", () => + Effect.gen(function* () { + const error = yield* ai.video + .start({ + model, + prompt: "x", + references: [Media.url("https://example.test/motion.mp4", { mediaType: "video/mp4" })], + }) + .pipe(Effect.flip) + expect(error.reason._tag).toBe("UnsupportedOperation") + expect(error.message).toContain("reference video") + }).pipe(Effect.provide(layer(() => Effect.die("an unsupported reference reached the network")))), + ) + + it.effect("rejects durations that aren't seconds before sending", () => + Effect.gen(function* () { + // @ts-expect-error Durations are Effect duration strings in seconds. + const error = yield* ai.video.start({ model, prompt: "x", duration: 8 }).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + }).pipe(Effect.provide(layer(() => Effect.die("an invalid request reached the network")))), + ) + + it.effect("never retries a failed submit", () => + Effect.gen(function* () { + const calls: Array = [] + const error = yield* ai.video + .start({ model, prompt: "x" }) + .pipe( + Effect.flip, + Effect.provide(layer((input) => observe(calls, input).pipe(Effect.map(() => failure(input, 503))))), + ) + expect(error.reason._tag).toBe("ProviderInternal") + expect(calls.map((call) => call.method)).toEqual(["POST"]) + }), + ) +}) + +// --------------------------------------------------------------------------- +// status, result, cancel +// --------------------------------------------------------------------------- + +describe("ai.video.status / result / cancel", () => { + it.effect("reads one status per call from a JSON round-tripped token", () => + Effect.gen(function* () { + const calls: Array = [] + const statuses = yield* Effect.gen(function* () { + const job = yield* ai.video.start({ model, prompt: "x" }) + const token: unknown = JSON.parse(JSON.stringify(job.token)) + return [yield* ai.video.status(model, token), yield* ai.video.status(model, token)] + }).pipe( + Effect.provide( + layer((input) => + Effect.gen(function* () { + const { call, nth } = yield* observe(calls, input) + if (call.method === "POST") return json(input, { request_id: "req_1" }) + return nth === 1 ? json(input, { status: "pending", progress: 40 }) : json(input, done) + }), + ), + ), + ) + expect(statuses).toEqual([{ status: "running", progress: 0.4 }, { status: "completed" }]) + expect(calls.map((call) => `${call.method} ${call.url}`)).toEqual([ + "POST https://xai.test/v1/videos/generations", + "GET https://xai.test/v1/videos/req_1", + "GET https://xai.test/v1/videos/req_1", + ]) + }), + ) + + it.effect("keeps unknown statuses running, including inherited property names", () => + Effect.gen(function* () { + const statuses = yield* Effect.forEach(["warming_up", "constructor"], (status) => + ai.video + .status(model, { requestID: "req_1" }) + .pipe(Effect.provide(layer((input) => Effect.succeed(json(input, { status }))))), + ) + expect(statuses.map((value) => value.status)).toEqual(["running", "running"]) + }), + ) + + it.effect("rejects a token from another provider", () => + Effect.gen(function* () { + const error = yield* ai.video.status(model, { operation: "models/x/operations/y" }).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toContain("cannot resume") + }).pipe(Effect.provide(layer(() => Effect.die("a foreign token reached the network")))), + ) + + it.effect("returns the finished video as an expiring URL with xAI metadata", () => + Effect.gen(function* () { + const response = yield* ai.video.result(model, { requestID: "req_1" }) + expect(response.video.source).toEqual({ type: "url", url: "https://vidgen.x.ai/out.mp4", mediaType: "video/mp4" }) + expect(response.video.info).toEqual({ durationSeconds: 10 }) + expect(response.providerMetadata).toEqual({ + xai: { requestId: "req_1", model: "grok-imagine-video-1.5", respectModeration: true }, + }) + }).pipe( + Effect.provide( + layer((input) => + Effect.succeed( + json(input, { + status: "done", + video: { url: "https://vidgen.x.ai/out.mp4", duration: 10, respect_moderation: true }, + model: "grok-imagine-video-1.5", + }), + ), + ), + ), + ), + ) + + it.effect("fails a result read before the job finishes", () => + Effect.gen(function* () { + const error = yield* ai.video.result(model, { requestID: "req_1" }).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toBe("xAI Video generation req_1 has not finished; await it before reading the result") + expect(error.reason.body).toBe(JSON.stringify({ status: "pending", progress: 40 })) + }).pipe(Effect.provide(layer((input) => Effect.succeed(json(input, { status: "pending", progress: 40 }))))), + ) + for (const terminal of [ { body: { status: "failed", error: { code: "invalid_argument", message: "Prompt cannot be empty." } }, @@ -135,11 +250,6 @@ describe("Video / xAI", () => { 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", @@ -147,189 +257,50 @@ describe("Video / xAI", () => { }, { body: { status: "expired" }, tag: "InvalidRequest", message: "xAI Video request req_1 expired" }, ]) { - 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) - expect(error.message).toBe(terminal.message) - expect(error.reason.body).toBe(JSON.stringify(terminal.body)) - }).pipe( - Effect.provide( - layer((input) => - Effect.succeed( - input.request.method === "POST" ? json(input, { request_id: "req_1" }) : json(input, terminal.body), - ), - ), - ), - ), + it.effect( + `fails the result of ${terminal.body.error?.code ?? terminal.body.status} jobs with the provider body`, + () => + Effect.gen(function* () { + const error = yield* ai.video.result(model, { requestID: "req_1" }).pipe(Effect.flip) + expect(error.reason._tag).toBe(terminal.tag) + expect(error.message).toBe(terminal.message) + expect(error.reason.body).toBe(JSON.stringify(terminal.body)) + }).pipe(Effect.provide(layer((input) => Effect.succeed(json(input, terminal.body))))), ) } it.effect("returns moderated results and fails withheld videos as a content policy failure", () => Effect.gen(function* () { - const flagged = yield* Video.generate({ model, prompt: "x" }).pipe( - Effect.provide( - layer((input) => - Effect.succeed( - input.request.method === "POST" - ? json(input, { request_id: "req_1" }) - : json(input, { - status: "done", - video: { url: "https://vidgen.x.ai/o.mp4", respect_moderation: false }, - }), + const flagged = yield* ai.video + .result(model, { requestID: "req_1" }) + .pipe( + Effect.provide( + layer((input) => + Effect.succeed( + json(input, { status: "done", video: { url: "https://vidgen.x.ai/o.mp4", respect_moderation: false } }), + ), ), ), - ), - ) + ) expect(flagged.providerMetadata).toEqual({ xai: { requestId: "req_1", respectModeration: false } }) - const withheld = yield* Video.generate({ model, prompt: "x" }).pipe( - Effect.flip, - Effect.provide( - layer((input) => - Effect.succeed( - input.request.method === "POST" - ? json(input, { request_id: "req_1" }) - : json(input, { status: "done", video: { respect_moderation: false } }), - ), + const withheld = yield* ai.video + .result(model, { requestID: "req_1" }) + .pipe( + Effect.flip, + Effect.provide( + layer((input) => Effect.succeed(json(input, { status: "done", video: { respect_moderation: false } }))), ), - ), - ) + ) expect(withheld.reason._tag).toBe("ContentPolicy") }), ) - it.effect("rejects seed, negativePrompt, and n before sending", () => + it.effect("fails cancel as unsupported without sending a request, since xAI has no cancel endpoint", () => Effect.gen(function* () { - const tags = yield* Effect.forEach( - [ - Video.generate({ model, prompt: "x", seed: 1 }), - Video.generate({ model, prompt: "x", negativePrompt: "blur" }), - Video.generate({ model, prompt: "x", n: 2 }), - ], - (program) => - program.pipe( - Effect.flip, - Effect.map((error) => error.reason._tag), - ), - ) - expect(tags).toEqual(["UnsupportedOperation", "UnsupportedOperation", "UnsupportedOperation"]) - }).pipe(Effect.provide(layer(() => Effect.die("unsupported input reached the network")))), - ) - - it.effect("resumes from a JSON round-tripped token and rejects foreign tokens", () => - Effect.gen(function* () { - const calls: Array = [] - const program = Effect.gen(function* () { - const started = yield* Video.start({ model, prompt: "x" }) - const token: unknown = JSON.parse(JSON.stringify(started.token)) - const resumed = yield* Video.resume(model, token) - expect(resumed.status).toBe("running") - expect(resumed.progress).toBe(0.25) - expect(resumed.token).toEqual({ requestID: "req_1" }) - const response = yield* resumed.await({ poll: { interval: "1 second" } }) - expect(response.videos).toHaveLength(1) - const foreign = yield* Video.resume(model, { operation: "models/x/operations/y" }).pipe(Effect.flip) - expect(foreign.reason._tag).toBe("InvalidRequest") - expect(foreign.message).toContain("cannot resume") - }) - yield* settle(program, 2).pipe( - Effect.provide( - layer((input) => - Effect.gen(function* () { - const { call, nth } = yield* observe(calls, input) - if (call.method === "POST") return json(input, { request_id: "req_1" }) - if (nth <= 2) return json(input, { status: "pending", progress: 25 }) - return json(input, { status: "done", video: { url: "https://vidgen.x.ai/out.mp4" } }) - }), - ), - ), - ) - expect(calls.filter((call) => call.method === "GET")).toHaveLength(4) - }), - ) - - it.effect("streams progress observations followed by the video and finish events", () => - Effect.gen(function* () { - const calls: Array = [] - const program = Video.stream({ model, prompt: "x" }, { poll: { interval: "1 second" } }).pipe(Stream.runCollect) - const events = Array.from( - 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, { request_id: "req_1" }) - if (nth === 1) return json(input, { status: "pending", progress: 20 }) - if (nth === 2) return json(input, { status: "pending", progress: 50 }) - return json(input, { status: "done", video: { url: "https://vidgen.x.ai/out.mp4" } }) - }), - ), - ), - ), - ) - expect(events.map((event) => event.type)).toEqual([ - "generation-progress", - "generation-progress", - "video", - "finish", - ]) - expect(events[1]).toEqual({ type: "generation-progress", id: "req_1", progress: 0.5 }) - }), - ) - - it.effect("streams the observations of a failed generation and then fails with the provider body", () => - Effect.gen(function* () { - const calls: Array = [] - const events: Array = [] - const failed = { status: "failed", error: { code: "internal_error", message: "Generation failed." } } - 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, { request_id: "req_1" }) - if (nth === 1) return json(input, { status: "pending", progress: 20 }) - if (nth === 2) return json(input, { status: "pending", progress: 50 }) - return json(input, failed) - }), - ), - ), - ) - expect(events).toEqual([ - { type: "generation-progress", id: "req_1", progress: 0.2 }, - { type: "generation-progress", id: "req_1", progress: 0.5 }, - ]) - expect(error.reason._tag).toBe("ProviderInternal") - expect(error.message).toBe("xAI Video generation failed (internal_error): Generation failed.") - 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( - { model, prompt: "x" }, - { poll: { interval: "1 second", timeout: "2 seconds" } }, - ).pipe(Stream.runCollect, Effect.flip) - const error = yield* settle(program, 3).pipe( - Effect.provide( - layer((input) => - Effect.succeed( - input.request.method === "POST" - ? json(input, { request_id: "req_1" }) - : json(input, { status: "pending" }), - ), - ), - ), - ) - expect(error.reason._tag).toBe("Timeout") - expect(error.message).toContain("req_1") - }), + const error = yield* ai.video.cancel(model, { requestID: "req_1" }).pipe(Effect.flip) + expect(error.reason._tag).toBe("UnsupportedOperation") + expect(error.message).toContain("cancel") + }).pipe(Effect.provide(layer(() => Effect.die("cancel reached the network")))), ) }) @@ -337,247 +308,103 @@ describe("Video / xAI", () => { // Transient read failures // --------------------------------------------------------------------------- -describe("Video / transient read failures", () => { - const model = XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }).video("grok-imagine-video-1.5") - const succeeded = { status: "done", video: { url: "https://vidgen.x.ai/out.mp4" } } - const failure = (input: HandlerInput, status: number, headers?: Record) => - json(input, { error: `HTTP ${status}` }, { status, headers }) - const methods = (calls: ReadonlyArray) => calls.map((call) => call.method) +const failure = (input: HandlerInput, status: number, headers?: Record) => + json(input, { error: `HTTP ${status}` }, { status, headers }) - it.effect("retries a 503 status poll and a 503 result read, then returns the result", () => +describe("ai.video / transient read failures", () => { + it.effect("retries a 503 status read and a 503 result read", () => Effect.gen(function* () { const calls: Array = [] const response = yield* settle( - Video.generate({ model, prompt: "x" }, { poll: { interval: "1 second" } }), + Effect.gen(function* () { + expect(yield* ai.video.status(model, { requestID: "req_1" })).toEqual({ status: "completed" }) + return yield* ai.video.result(model, { requestID: "req_1" }) + }), 5, ).pipe( Effect.provide( layer((input) => Effect.gen(function* () { - const { call, nth } = yield* observe(calls, input) - if (call.method === "POST") return json(input, { request_id: "req_1" }) + const { nth } = yield* observe(calls, input) // 1: status fails, 2: status succeeds, 3: result fails, 4: result succeeds. - if (nth === 1 || nth === 3) return failure(input, 503) - return json(input, succeeded) + return nth === 1 || nth === 3 ? failure(input, 503) : json(input, done) }), ), ), ) expect(response.video.source).toMatchObject({ type: "url", url: "https://vidgen.x.ai/out.mp4" }) - expect(methods(calls)).toEqual(["POST", "GET", "GET", "GET", "GET"]) + expect(calls).toHaveLength(4) }), ) - it.effect("waits for a 429 retry-after before polling again", () => + it.effect("waits for a 429 retry-after before reading the status again", () => Effect.gen(function* () { const calls: Array = [] const fiber = yield* Effect.forkChild( - Video.generate({ model, prompt: "x" }, { poll: { interval: "1 second" } }).pipe( + ai.video.status(model, { requestID: "req_1" }).pipe( Effect.provide( layer((input) => Effect.gen(function* () { - const { call, nth } = yield* observe(calls, input) - if (call.method === "POST") return json(input, { request_id: "req_1" }) - if (nth === 1) return failure(input, 429, { "retry-after": "10" }) - return json(input, succeeded) + const { nth } = yield* observe(calls, input) + return nth === 1 ? failure(input, 429, { "retry-after": "10" }) : json(input, done) }), ), ), ), ) yield* TestClock.adjust("9 seconds") - expect(methods(calls)).toEqual(["POST", "GET"]) + expect(calls).toHaveLength(1) yield* TestClock.adjust("1 second") - yield* Fiber.join(fiber) - expect(methods(calls)).toEqual(["POST", "GET", "GET", "GET"]) + expect(yield* Fiber.join(fiber)).toEqual({ status: "completed" }) + expect(calls).toHaveLength(2) }), ) - it.effect("fails a 400 status poll without retrying", () => + it.effect("fails a 400 status read without retrying", () => Effect.gen(function* () { const calls: Array = [] - const error = yield* Video.generate({ model, prompt: "x" }).pipe( - Effect.flip, - Effect.provide( - layer((input) => - Effect.gen(function* () { - const { call } = yield* observe(calls, input) - return call.method === "POST" ? json(input, { request_id: "req_1" }) : failure(input, 400) - }), - ), - ), - ) - expect(error.reason._tag).toBe("InvalidRequest") - expect(methods(calls)).toEqual(["POST", "GET"]) - }), - ) - - it.effect("stops retrying at poll.timeout with a Timeout reason", () => - Effect.gen(function* () { - const calls: Array = [] - const error = yield* settle( - Video.generate({ model, prompt: "x" }, { poll: { interval: "1 second", timeout: "5 seconds" } }).pipe( + const error = yield* ai.video + .status(model, { requestID: "req_1" }) + .pipe( Effect.flip, - ), - 6, - ).pipe( - Effect.provide( - layer((input) => - Effect.gen(function* () { - const { call } = yield* observe(calls, input) - return call.method === "POST" ? json(input, { request_id: "req_1" }) : failure(input, 503) - }), - ), - ), - ) - expect(error.reason._tag).toBe("Timeout") - expect(calls.filter((call) => call.method === "GET").length).toBeGreaterThan(1) + Effect.provide(layer((input) => observe(calls, input).pipe(Effect.map(() => failure(input, 400))))), + ) + expect(error.reason._tag).toBe("InvalidRequest") + expect(calls).toHaveLength(1) }), ) +}) - it.effect("bounds a streamed result read's retries by poll.timeout", () => +// --------------------------------------------------------------------------- +// Media streaming +// --------------------------------------------------------------------------- + +describe("Media.Asset.stream", () => { + it.effect("streams a URL asset's bytes from the network with its download headers", () => Effect.gen(function* () { const calls: Array = [] - const error = yield* settle( - Video.stream({ model, prompt: "x" }, { poll: { interval: "1 second", timeout: "5 seconds" } }).pipe( + const asset = Media.url("https://vidgen.x.ai/out.mp4", { mediaType: "video/mp4", headers: { "x-key": "k" } }) + const chunks = yield* asset + .stream() + .pipe( Stream.runCollect, - Effect.flip, - ), - 6, - ).pipe( - Effect.provide( - layer((input) => - Effect.gen(function* () { - const { call, nth } = yield* observe(calls, input) - if (call.method === "POST") return json(input, { request_id: "req_1" }) - return nth === 1 ? json(input, succeeded) : failure(input, 503) - }), + Effect.provide( + dynamicResponse((input) => + observe(calls, input).pipe(Effect.map(() => input.respond(new Uint8Array([1, 2, 3])))), + ), ), - ), - ) - expect(error.reason._tag).toBe("Timeout") - expect(calls.filter((call) => call.method === "GET").length).toBeGreaterThan(2) + ) + expect(new Uint8Array(chunks.flatMap((chunk) => [...chunk]))).toEqual(new Uint8Array([1, 2, 3])) + expect(calls[0].headers.get("x-key")).toBe("k") }), ) - it.effect("never retries a failed submit", () => + it.effect("streams inline bytes without a request", () => Effect.gen(function* () { - const calls: Array = [] - const error = yield* Video.generate({ model, prompt: "x" }).pipe( - Effect.flip, - Effect.provide(layer((input) => observe(calls, input).pipe(Effect.map(() => failure(input, 503))))), - ) - expect(error.reason._tag).toBe("ProviderInternal") - expect(methods(calls)).toEqual(["POST"]) - }), - ) -}) - -// --------------------------------------------------------------------------- -// Shared queued behavior -// --------------------------------------------------------------------------- - -describe("Video / queued result", () => { - const model = XAI.configure({ apiKey: "test", baseURL: "https://xai.test/v1" }).video("grok-imagine-video-1.5") - const running = { status: "pending", progress: 40 } - - it.effect("rejects reading a result before the generation finishes", () => - Effect.gen(function* () { - const generation = yield* Video.resume(model, { requestID: "req_1" }) - const error = yield* generation.result().pipe(Effect.flip) - expect(error.reason._tag).toBe("InvalidRequest") - expect(error.message).toBe( - `xAI Video generation ${generation.id} has not finished; await it before reading the result`, - ) - expect(error.reason.body).toBe(JSON.stringify(running)) - expect(error.reason.http?.status).toBe(200) - }).pipe(Effect.provide(layer((input) => Effect.succeed(json(input, running))))), - ) - - it.effect("resumes a 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, prompt: "x" }) - const resumed = yield* Video.resume(model, JSON.parse(JSON.stringify(started.token))) - expect(resumed.status).toBe("running") - expect(resumed.token).toEqual({ requestID: "req_1" }) - 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, { request_id: "req_1" }) - if (nth === 1) return json(input, running) - return json(input, { - status: "done", - video: { url: "https://vidgen.x.ai/out.mp4", respect_moderation: true }, - }) - }), - ), - ), - ) - expect(response.video.source).toEqual( - expect.objectContaining({ type: "url", url: "https://vidgen.x.ai/out.mp4" }), - ) - expect(calls.map((call) => `${call.method} ${call.url}`)).toEqual([ - "POST https://xai.test/v1/videos/generations", - "GET https://xai.test/v1/videos/req_1", - "GET https://xai.test/v1/videos/req_1", - "GET https://xai.test/v1/videos/req_1", - ]) - }), - ) - - it.effect("cancels a generation without sending a request when the provider has no cancel endpoint", () => - Effect.gen(function* () { - const calls: Array = [] - yield* Effect.gen(function* () { - const generation = yield* Video.start({ 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, { request_id: "req_1" }) - }), - ), - ), - ) - expect(calls.map((call) => call.method)).toEqual(["POST"]) - }), - ) - - it.effect("keeps polling through unknown statuses, including inherited property names", () => - Effect.gen(function* () { - const calls: Array = [] - const response = yield* settle( - Video.generate( - { - model, - prompt: "x", - }, - { poll: { interval: "1 second" } }, - ), - 3, - ).pipe( - Effect.provide( - layer((input) => - Effect.gen(function* () { - const { call, nth } = yield* observe(calls, input) - if (call.method === "POST") return json(input, { request_id: "req_1" }) - if (nth === 1) return json(input, { status: "warming_up" }) - if (nth === 2) return json(input, { status: "constructor" }) - return json(input, { status: "done", video: { url: "https://vidgen.x.ai/out.mp4" } }) - }), - ), - ), - ) - expect(response.video.source).toMatchObject({ type: "url", url: "https://vidgen.x.ai/out.mp4" }) - expect(calls.map((call) => call.method)).toEqual(["POST", "GET", "GET", "GET", "GET"]) - }), + const chunks = yield* Media.bytes(new Uint8Array([4, 5]), "video/mp4") + .stream() + .pipe(Stream.runCollect) + expect(chunks).toEqual([new Uint8Array([4, 5])]) + }).pipe(Effect.provide(dynamicResponse(() => Effect.die("inline bytes reached the network")))), ) }) diff --git a/packages/ai/test/video.types.ts b/packages/ai/test/video.types.ts index f1df70d0f074..ad289ae77b7b 100644 --- a/packages/ai/test/video.types.ts +++ b/packages/ai/test/video.types.ts @@ -1,14 +1,16 @@ import { Effect } from "effect" -import { Generation, Media } from "../src/effect.js" import { - Video, + Media, VideoClient, VideoModel, VideoResponse, + ai, + type VideoJob, type VideoModelOptions, type VideoOptions, type VideoRequest, -} from "../src/experimental.js" + type VideoStatus, +} from "../src/effect.js" import type { Service } from "../src/video-client.js" import { Anthropic, OpenAI, XAI } from "../src/providers.js" @@ -26,70 +28,79 @@ declare const custom: VideoModel const invalidCustomOptions: VideoModelOptions = { personGeneration: "everyone" } void invalidCustomOptions -Video.generate( - { - model: custom, - prompt: "A kitten", - frames: { - first: Media.bytes(Uint8Array.from([1]), "image/png"), - last: Media.fromDataUrl("data:image/png;base64,AQID"), - }, - references: [Media.bytes(Uint8Array.from([1]), "image/png")], - durationSeconds: 8, - aspect: "16:9", - resolution: "1080p", - audio: true, - seed: 7, - negativePrompt: "text", - providerOptions: { personGeneration: "allow_adult", futureOption: true }, - }, - { poll: { interval: "10 seconds", timeout: "10 minutes" } }, -) +ai.video.start({ + model: custom, + prompt: "A kitten", + start: Media.bytes(Uint8Array.from([1]), "image/png"), + end: "data:image/png;base64,AQID", + references: [Media.bytes(Uint8Array.from([1]), "image/png"), "https://example.com/voice.mp3"], + duration: "8 seconds", + aspect: "16:9", + resolution: "1080p", + sound: true, + providerOptions: { personGeneration: "allow_adult", futureOption: true }, +}) +ai.video.start({ model: custom, prompt: "A kitten", duration: "1 second", aspect: "auto", resolution: "2k" }) const xai = XAI.configure({ apiKey: "test" }).video("grok-imagine-video-1.5") -Video.start({ +ai.video.start({ model: xai, prompt: "Waves", - video: Media.url("https://example.com/in.mp4"), - providerOptions: { mode: "extend", reference_audios: [{ voice_id: "eve" }], future_option: true }, + providerOptions: { + mode: "extend", + video: { url: "https://example.com/in.mp4" }, + reference_audios: [{ voice_id: "eve" }], + future_option: true, + }, }) // @ts-expect-error xAI modes are a closed set because the route selects an endpoint from them. -Video.start({ model: xai, prompt: "Waves", providerOptions: { mode: "remix" } }) +ai.video.start({ model: xai, prompt: "Waves", providerOptions: { mode: "remix" } }) // @ts-expect-error Known xAI nested options retain their value kind. -Video.start({ model: xai, prompt: "Waves", providerOptions: { reference_audios: [{ voice_id: 1 }] } }) -Video.generate({ model: xai, prompt: "Waves", resolution: "future-resolution" }) +ai.video.start({ model: xai, prompt: "Waves", providerOptions: { reference_audios: [{ voice_id: 1 }] } }) // @ts-expect-error Known custom options are inferred from the selected model. -Video.generate({ model: custom, prompt: "A kitten", providerOptions: { personGeneration: "everyone" } }) +ai.video.start({ model: custom, prompt: "A kitten", providerOptions: { personGeneration: "everyone" } }) // @ts-expect-error Language models cannot be used for video requests. -Video.generate({ model: Anthropic.configure({ apiKey: "test" }).model("claude-sonnet-4-5"), prompt: "A kitten" }) +ai.video.start({ model: Anthropic.configure({ apiKey: "test" }).model("claude-sonnet-4-5"), prompt: "A kitten" }) // @ts-expect-error Image models cannot be used for video requests. -Video.generate({ model: OpenAI.configure({ apiKey: "test" }).image("gpt-image-2"), prompt: "A kitten" }) +ai.video.start({ model: OpenAI.configure({ apiKey: "test" }).image("gpt-image-2"), prompt: "A kitten" }) -// @ts-expect-error Frames are media assets, not strings. -Video.generate({ model: xai, prompt: "A kitten", frames: { first: "https://example.com/first.png" } }) -// @ts-expect-error Aspect ratios are `${w}:${h}` strings. -Video.generate({ model: xai, prompt: "A kitten", aspect: "wide" }) -// @ts-expect-error Durations are numbers of seconds. -Video.generate({ model: xai, prompt: "A kitten", durationSeconds: "8s" }) -// @ts-expect-error Provider-native controls live under `providerOptions`. -Video.generate({ model: xai, prompt: "A kitten", options: { mode: "extend" } }) +// @ts-expect-error Aspect ratios are `${w}:${h}` strings or "auto". +ai.video.start({ model: xai, prompt: "A kitten", aspect: "wide" }) +// @ts-expect-error Durations are Effect duration strings in seconds, not numbers. +ai.video.start({ model: xai, prompt: "A kitten", duration: 8 }) +// @ts-expect-error Durations are seconds for now. +ai.video.start({ model: xai, prompt: "A kitten", duration: "1 minute" }) +// @ts-expect-error Sound is on or off. +ai.video.start({ model: xai, prompt: "A kitten", sound: "native" }) +// @ts-expect-error The shared request has no seed; it lives under `providerOptions`. +ai.video.start({ model: xai, prompt: "A kitten", seed: 7 }) +// @ts-expect-error Start and end replace `frames`. +ai.video.start({ model: xai, prompt: "A kitten", frames: { first: "https://example.com/first.png" } }) declare const generic: VideoModel const widenVideo = (model: VideoModel): VideoModel => model void widenVideo -Video.generate({ model: generic, prompt: "A kitten", providerOptions: { arbitrary: true } }) +ai.video.start({ model: generic, prompt: "A kitten", providerOptions: { arbitrary: true } }) -const request = Video.request({ model: custom, prompt: "A kitten", providerOptions: { personGeneration: "allow_all" } }) +const request = ai.video.request({ + model: custom, + prompt: "A kitten", + providerOptions: { personGeneration: "allow_all" }, +}) const typedRequest: VideoRequest> = request void typedRequest + const started = VideoClient.start(request) type StartRequirements = Assert, Service>> -type StartSuccess = Assert, Generation>> +type StartSuccess = Assert, VideoJob>> void (true satisfies StartRequirements) void (true satisfies StartSuccess) -const resumed = Video.resume(xai, { requestID: "req_1" }) -type ResumeSuccess = Assert, Generation>> -void (true satisfies ResumeSuccess) +type StatusSuccess = Assert>, VideoStatus>> +type ResultSuccess = Assert>, VideoResponse>> +type CancelSuccess = Assert>, void>> +void (true satisfies StatusSuccess) +void (true satisfies ResultSuccess) +void (true satisfies CancelSuccess)