vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-11-14 00:53:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat(ai): graduate video generation with start, status, result and cancel (#63657)

Browse source
This commit is contained in:
Aiden Cline 2026-11-12 21:56:34 -06:00 • committed by GitHub
parent 265be7a96d
commit 5eac2802b7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 762 additions and 810 deletions
Download patch file Download diff file Expand all files Collapse all files

162
packages/ai/README.md
View file

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

3
packages/ai/src/effect.ts
View file

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

11
packages/ai/src/experimental.ts
View file

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

55
packages/ai/src/index.ts
View file

@ -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: <const Model extends EmbeddingModel>(input: EmbeddingRequestInput<Model>, options?: RunOptions) =>
run(Embedding.generate(input), options),
},
video: {
request: Video.request,
/** Submit a generation and return its `{ id, token }` at once; nothing polls. */
start: <const Model extends VideoModel>(input: VideoRequestInput<Model> | 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: <const Model extends VideoModel>(input: VideoRequestInput<Model> | VideoRequest, options?: RunOptions) =>
run(Video.start(input), options).then(handle),
generate: <const Model extends VideoModel>(
input: VideoRequestInput<Model> | 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: <const Model extends VideoModel>(
input: VideoRequestInput<Model> | VideoRequest,
options?: AwaitOptions & RunOptions,
) => iterate(Video.stream(input, { poll: options?.poll }), options),
},
},
dispose: () => runtime.dispose(),
}

8
packages/ai/src/media-model.ts
View file

@ -84,10 +84,16 @@ export const composeRoute = <Request extends MediaRoute.MediaRequest, Event, Res
collect: (events: ReadonlyArray<Event>) => Effect.Effect<Response, AIError>,
): MediaRoute.AnyRoute<Request, Event, Response> => {
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 = <Request extends MediaRoute.MediaRequest, Response, Token>(
route: MediaModel.QueuedRouteInput<Request, Response, Token>,
input: MediaRoute.ModelInput,
) => MediaRoute.queued(composition(route, input))
const composition = <Request extends MediaRoute.MediaRequest, Protocol>(
route: MediaModel.RouteInput<Request, Protocol>,
input: MediaRoute.ModelInput,

50
packages/ai/src/media.ts
View file

@ -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<Uint8Array, AIError, Service> {
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<string, AIError, Service> {
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<Source, { readonly type: "url" }>,
headers: Record<string, string> | 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<Asset,
return bytes(data, fileMediaType(data, path), options)
})
/** Materialize an asset and write its bytes through `FileSystem`. */
/**
* Stream an asset's bytes into a file through `FileSystem` without buffering `url` downloads. The bytes go to a temporary
* file next to `path` that replaces it only once the download finished, so a failed download leaves an existing file as
* it was.
*/
export const write = (asset: Asset, path: string): Effect.Effect<void, AIError, FileSystem.FileSystem | Service> =>
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),
),
)
})

47
packages/ai/src/protocols/xai-video.ts
View file

@ -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<string, unknown>
@ -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<Request, VideoResponse, Token>(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) =>

11
packages/ai/src/public.ts
View file

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

26
packages/ai/src/route/media.ts
View file

@ -78,6 +78,12 @@ export interface QueuedRoute<Request extends MediaRequest, Response> {
token: unknown,
execute: Execute,
) => Effect.Effect<Generation<Response>, 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<Generation<Response>["route"], AIError>
}
/** One request whose response parses into events; `generate` runs the same stream and collects it. */
@ -219,12 +225,8 @@ export const queued = <Request extends MediaRequest, Response, Token>(
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 = <Request extends MediaRequest, Response, Token>(
}),
),
)
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`. */

63
packages/ai/src/video-client.ts
View file

@ -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<VideoRequest, VideoEvent, VideoResponse>
export interface Interface {
readonly start: (request: VideoRequest) => Effect.Effect<VideoJob, AIError>
readonly status: (model: VideoModel, token: unknown) => Effect.Effect<VideoStatus, AIError>
readonly result: (model: VideoModel, token: unknown) => Effect.Effect<VideoResponse, AIError>
readonly cancel: (model: VideoModel, token: unknown) => Effect.Effect<void, AIError>
}
export class VideoClientService extends Context.Service<VideoClientService, Interface>()("@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

167
packages/ai/src/video.ts
View file

@ -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<VideoRequest, VideoEvent, VideoResponse>
/** Video generation is always a queued job: submit, then read its status and result by token. */
export type VideoRoute = MediaRoute.QueuedRoute<VideoRequest, VideoResponse>
export class VideoModel<Options extends VideoOptions = VideoOptions> extends MediaModel<VideoRoute, Options> {
declare protected readonly _VideoModel: void
/** The number of type arguments selects the kind: `<Options>`, `<Options, Frame, State>`, or `<Options, Token>`. */
static fromRoute<Options extends VideoOptions>(
route: MediaModel.InlineRouteInput<VideoRequest<VideoModel<Options>>, VideoResponse>,
input: MediaRoute.ModelInput,
): VideoModel<Options>
static fromRoute<Options extends VideoOptions, Frame, State>(
route: MediaModel.StreamRouteInput<VideoRequest<VideoModel<Options>>, VideoEvent, Frame, State>,
input: MediaRoute.ModelInput,
): VideoModel<Options>
static fromRoute<Options extends VideoOptions, Token>(
route: MediaModel.QueuedRouteInput<VideoRequest<VideoModel<Options>>, VideoResponse, Token>,
input: MediaRoute.ModelInput,
): VideoModel<Options>
static fromRoute<Options extends VideoOptions, Frame, State, Token>(
route: MediaModel.AnyRouteInput<VideoRequest<VideoModel<Options>>, VideoEvent, VideoResponse, Frame, State, Token>,
input: MediaRoute.ModelInput,
) {
return new VideoModel<Options>({
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<typeof VideoFrames>
/** 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<VideoDuration>(
(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<Model extends VideoModel = VideoModel> = Omit<
typeof VideoRequestSchema.Type,
"model" | "providerOptions" | "resolution"
@ -99,15 +87,36 @@ export type VideoRequest<Model extends VideoModel = VideoModel> = Omit<
export type VideoModelOptions<Model> = Model extends VideoModel<infer Options> ? Options : never
/** Input accepted by `Video.request` and the calls; a `VideoRequest` is also a valid input. */
export type VideoRequestInput<Model extends VideoModel = VideoModel> = Omit<VideoRequest<Model>, "providerOptions"> & {
/** Input accepted by `ai.video.request` and `ai.video.start`; a `VideoRequest` is also a valid input. */
export type VideoRequestInput<Model extends VideoModel = VideoModel> = Omit<
VideoRequest<Model>,
"start" | "end" | "references" | "providerOptions"
> & {
readonly start?: Media.Input
readonly end?: Media.Input
readonly references?: ReadonlyArray<Media.Input>
readonly providerOptions?: NoInfer<VideoModelOptions<Model>>
}
/** 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<VideoResponse>("Video.Response")({
videos: Schema.Array(Media.AssetSchema),
usage: Schema.optional(MediaUsage),
@ -118,51 +127,16 @@ export class VideoResponse extends Schema.Class<VideoResponse>("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<typeof videoEventTagged>
const collectResponse = (events: ReadonlyArray<VideoEvent>): Effect.Effect<VideoResponse> => {
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 = <const Model extends VideoModel>(input: VideoRequestInput<Model>): VideoRequest<Model> => ({
...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<const Model extends VideoModel>(
input: VideoRequestInput<Model>,
): Effect.Effect<Generation<VideoResponse>, AIError, Service>
): Effect.Effect<VideoJob, AIError, Service>
export function start(input: VideoRequestInput) {
return requestEffect(input).pipe(Effect.flatMap((request) => VideoClient.start(request)))
}
export function generate<const Model extends VideoModel>(
input: VideoRequestInput<Model>,
options?: AwaitOptions,
): Effect.Effect<VideoResponse, AIError, Service>
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<VideoStatus, AIError, Service> =>
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<Generation<VideoResponse>, 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<VideoResponse, AIError, Service> =>
VideoClient.result(model, token)
export function stream<const Model extends VideoModel>(
input: VideoRequestInput<Model>,
options?: AwaitOptions,
): Stream.Stream<VideoEvent, AIError, Service>
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<void, AIError, Service> =>
VideoClient.cancel(model, token)
export const Video = {
request,
start,
generate,
resume,
stream,
status,
result,
cancel,
} as const

10
packages/ai/test/exports.test.ts
View file

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

27
packages/ai/test/media.test.ts
View file

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

14
packages/ai/test/promise-parity.types.ts
View file

@ -12,7 +12,7 @@ type Assert<T extends Record<string, true>> = T
type PromiseAI = Omit<Client, "dispose">
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> = F extends {
(...args: infer A1): infer R1
@ -49,7 +49,11 @@ type Promised<R> =
: R
type EffectShape<F> =
Signatures<F> extends infer S ? (S extends [infer A, infer R] ? [Positional<A>, Options<A>, Promised<R>] : never) : never
Signatures<F> extends infer S
? S extends [infer A, infer R]
? [Positional<A>, Options<A>, Promised<R>]
: never
: never
type PromiseShape<F> =
Signatures<F> extends infer S ? (S extends [infer A, infer R] ? [Positional<A>, Options<A>, R] : never) : never
@ -61,7 +65,9 @@ export type Drift<Promise, Effect> = {
: "missing in Promise"
}
type Same<A, B> = [Exclude<A, B> | Exclude<B, A>] extends [never] ? true : { readonly drift: Exclude<A, B> | Exclude<B, A> }
type Same<A, B> = [Exclude<A, B> | Exclude<B, A>] extends [never]
? true
: { readonly drift: Exclude<A, B> | Exclude<B, A> }
export type Checks = [
Assert<{ handle: Same<keyof GenerationHandle<unknown>, Exclude<keyof Generation<unknown>, "route">> }>,
@ -71,7 +77,7 @@ export type Checks = [
Assert<Drift<PromiseAI["image"], EffectAI["image"]>>,
Assert<Drift<PromiseAI["decision"], EffectAI["decision"]>>,
Assert<Drift<PromiseAI["embedding"], EffectAI["embedding"]>>,
Assert<Drift<PromiseAI["video"], EffectAI["video"]>>,
Assert<Drift<PromiseAI["experimental"]["speech"], EffectAI["experimental"]["speech"]>>,
Assert<Drift<PromiseAI["experimental"]["transcription"], EffectAI["experimental"]["transcription"]>>,
Assert<Drift<PromiseAI["experimental"]["video"], EffectAI["experimental"]["video"]>>,
]

92
packages/ai/test/promise.test.ts
View file

@ -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<string>) =>
@ -94,10 +98,17 @@ const executor = (seen: Array<string>) =>
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<string> = []
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<string> = []
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<string> = []
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<string> = []
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")
})
})

22
packages/ai/test/provider/xai-video.recorded.test.ts
View file

@ -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<VideoStatus> = []
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)

703
packages/ai/test/video.test.ts
View file

@ -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<typeof dynamicResponse>[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<Call> = []
const events: Array<GenerationEvent> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
const events: Array<VideoEvent> = []
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<string, string>) =>
json(input, { error: `HTTP ${status}` }, { status, headers })
const methods = (calls: ReadonlyArray<Call>) => calls.map((call) => call.method)
const failure = (input: HandlerInput, status: number, headers?: Record<string, string>) =>
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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<Call> = []
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")))),
)
})

101
packages/ai/test/video.types.ts
View file

@ -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<CustomOptions>
const invalidCustomOptions: VideoModelOptions<typeof custom> = { 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<VideoOptions>
const widenVideo = <Options extends VideoOptions>(model: VideoModel<Options>): 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<VideoModel<CustomOptions>> = request
void typedRequest
const started = VideoClient.start(request)
type StartRequirements = Assert<Equal<Requirements<typeof started>, Service>>
type StartSuccess = Assert<Equal<Success<typeof started>, Generation<VideoResponse>>>
type StartSuccess = Assert<Equal<Success<typeof started>, VideoJob>>
void (true satisfies StartRequirements)
void (true satisfies StartSuccess)
const resumed = Video.resume(xai, { requestID: "req_1" })
type ResumeSuccess = Assert<Equal<Success<typeof resumed>, Generation<VideoResponse>>>
void (true satisfies ResumeSuccess)
type StatusSuccess = Assert<Equal<Success<ReturnType<typeof ai.video.status>>, VideoStatus>>
type ResultSuccess = Assert<Equal<Success<ReturnType<typeof ai.video.result>>, VideoResponse>>
type CancelSuccess = Assert<Equal<Success<ReturnType<typeof ai.video.cancel>>, void>>
void (true satisfies StatusSuccess)
void (true satisfies ResultSuccess)
void (true satisfies CancelSuccess)