feat(ai): add queued image routes, partial image streaming, and BFL, fal, Replicate, Stability (#50959)

This commit is contained in:
Shoubhit Dash
2026-09-24 00:54:37 +05:30
committed by GitHub
parent c98d9c65dc
commit 56db9053e8
43 changed files with 2403 additions and 275 deletions
+1 -1
View File
@@ -102,7 +102,7 @@ Media does not fit the SSE-frames-to-event-state-machine LLM route. `MediaRoute.
`MediaProtocol.stream` is the incremental kind every speech route uses, with the same discipline as LLM protocols. `MediaRoute.stream` submits the caller's request as `MediaProtocol.Addressed<Request>` (`{ ...request, mode }`, `mode: "generate" | "stream"`), so one provider stays one protocol: `body.from`, the endpoint path, and `frames` read `request.mode` to pick the body, path, and framing. `frames(bytes, context)` returns frames — `Framing.sse`, `Framing.lines`, `Framing.document` (a single-document response shaped like a streamed record), or the raw `bytes` for chunked audio. `initial()` is fresh per-response parser state; `step` folds each frame into it and emits modality events; `finish(state, context)` runs once after the last frame with the request, body, and observed `http` (header-only usage lives there) and emits exactly one terminal event or fails with `MediaProtocol.incomplete`. Keep parser state to real accumulators and derive anything the request or body determines in `finish`. `generate` runs the same stream and folds it with the modality's `collect`. Request-derived URL parameters go on the body's `query` (array values repeat the parameter), applied before route and caller `http.query`. Decode frames with `MediaProtocol.decodeFrame` and raise stream-time failures with `MediaProtocol.frameError` (the frame stays on `reason.body`); protocols never thread HTTP context, because the route fills `reason.http` on stream errors that lack it. Speech protocols share `protocols/utils/speech-stream.ts` for deltas, timestamps, voice ids, PCM and container descriptions, and the terminal asset.
Transcription uses all three kinds (OpenAI and Gemini stream, Deepgram is inline, AssemblyAI is queued): every route carries its `kind` and `TranscriptionClient` dispatches on it. Bodies are `json`, `multipart`, or `binary` (a raw upload), and a queued protocol that must upload media before submitting implements `start.prepare` (`MediaProtocol.Prepare`; AssemblyAI `/v2/upload`).
Transcription uses all three kinds (OpenAI and Gemini stream, Deepgram is inline, AssemblyAI is queued): every route carries its `kind` and `TranscriptionClient` dispatches on it. `ImageRoute` is the same union; both clients dispatch through `MediaRoute.dispatch` and models compose through `composeAnyRoute`, and fal queue protocols come from `protocols/utils/fal-queue.ts`. Bodies are `json`, `multipart`, or `binary` (a raw upload), and a queued protocol that must upload media before submitting implements `start.prepare` (`MediaProtocol.Prepare`; AssemblyAI `/v2/upload`).
### URL Construction
+56 -1
View File
@@ -584,6 +584,61 @@ Z.ai does not include trustworthy MIME metadata for output URLs, so generated im
`application/octet-stream` until materialized. Output URLs expire after 30 days; call `asset.materialize()` and
persist the bytes promptly if they must remain available.
### Partial images
OpenAI's GPT image models stream previews. `Image.stream` sends `stream: true` with `partialImages` (0–3, default 2)
and emits `image-partial` events before each final `image`; `Image.generate` keeps the plain JSON request.
`dall-e-*` models do not stream and fail typed:
```ts
yield *
Image.stream({
model: openai.image("gpt-image-2"),
prompt: "A lighthouse at dusk",
providerOptions: { partialImages: 2 },
}).pipe(Stream.runForEach((event) => (ImageEvent.is.imagePartial(event) ? showPreview(event.image) : Effect.void)))
```
The provider may send fewer previews than requested when the final image is ready first.
### Queued image providers
Black Forest Labs, fal, Replicate, and Stability's creative upscaler are submit-then-poll routes. `Image.generate`
and `Image.stream` poll for you (pass `{ poll }` to tune the interval and timeout); `Image.start` returns a
`Generation` whose `token` is serializable JSON for `Image.resume` in another process:
```ts
import { BlackForestLabs, Stability } from "@opencode/ai/providers"
const bfl = BlackForestLabs.configure({ apiKey: process.env.BFL_API_KEY })
const generation = yield * Image.start({ model: bfl.image("flux-2-pro"), prompt, size: "1024x768" })
persist(generation.token)
const resumed = yield * Image.resume(bfl.image("flux-2-pro"), loadToken())
const response = yield * resumed.await({ poll: { interval: "2 seconds" } })
```
- **Black Forest Labs** — results are downloaded before returning, because `result.sample` expires in 10 minutes.
- **Replicate** — inputs are model-defined, so only `prompt` lowers: sizing, count, seed, format, and files go in
`providerOptions` under the model's names, with files as `Media.Asset` (data URLs up to 256 KB, larger by URL).
Outputs are removed an hour after the prediction completes. `Prefer: wait=60` in `headers` or `http.headers` holds
the submission open so a fast prediction costs one result read.
- **Stability** — `stability.image(id)` generates inline; `stability.upscale()` is the creative upscaler, queued:
```ts
const stability = Stability.configure({ apiKey: process.env.STABILITY_API_KEY })
const upscaled =
yield *
Image.generate(
{ model: stability.upscale(), prompt: "A lighthouse", images: [yield * Media.file("./small.png")] },
{ poll: { interval: "5 seconds" } },
)
```
Imagen is not available: Google shut it down on the Gemini API, and Vertex discontinued the Imagen 4 models on
2026-06-30. `Google.image(...)` uses Gemini-native image models.
Conversational image generation remains part of the LLM interaction. OpenAI Responses exposes it through its hosted image tool:
```ts
@@ -836,7 +891,7 @@ const transcript = await generation.await({ poll: { interval: 3_000 } })
- **`Message.user(...)` / `Message.assistant(...)` / `Message.tool(...)`** — message constructors from the canonical schema model.
- **`LanguageModel.make(...)` / `ToolCallPart.make(...)` / `ToolResultPart.make(...)` / `ToolDefinition.make(...)`** — model and tool-related constructors from the canonical schema model.
- **`LLMEvent.is.*`** — typed guards (`is.textDelta`, `is.toolCall`, `is.finish`, …) for filtering streams.
- **`Image.request` / `Image.generate` / `Image.stream`** — generate images through a provider-neutral image request and response model.
- **`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`.
- **`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.
+14 -8
View File
@@ -11,11 +11,11 @@ The design below is derived from a survey of the raw provider APIs (OpenAI, Gemi
## What the survey forces
1. **Three execution shapes, everywhere.** Inline sync (OpenAI images, all TTS, Gemini), async job with polling or webhook (every video provider, BFL, fal, Replicate, AssemblyAI), and bidirectional streams (ElevenLabs/Cartesia/Deepgram WS, realtime). Video has no sync provider at all.
2. **Output is never just bytes.** base64, signed URLs with TTLs from 10 minutes (BFL) to 2 days (Veo), URLs that need auth plus redirect (Veo), separate download endpoints (Sora `/content?variant=`), raw bodies (Stability, TTS). Multi-output is the norm.
2. **Output is never just bytes.** base64, signed URLs with TTLs from 10 minutes (BFL, so its route downloads before returning via `PollContext.materialize`) to 2 days (Veo), URLs that need auth plus redirect (Veo), separate download endpoints (Sora `/content?variant=`), raw bodies (Stability, TTS). Multi-output is the norm.
3. **Inputs have roles.** First/last frame, mask, style/subject reference, source video for edit/extend, reference audio, prior generation id, provider-side file handles (`file_id`, `gs://`, `runway://`, `mm_file://`).
4. **Partial streaming is modality-specific.** Images: a few whole partial frames. Audio: ordered chunks plus timestamp events. Jobs: status/progress/logs. Video: none.
5. **Usage is a union**: tokens, seconds, characters (often only in headers), credits, compute time.
6. **Moderation can be partial success** (Veo strips audio but returns video). Deprecations are constant (Sora API shuts down 2026-09-24, Imagen on Gemini API 2026-08-17).
6. **Moderation can be partial success** (Veo strips audio but returns video). Deprecations are constant (Sora API shuts down 2026-09-24; Imagen is shut down on the Gemini API and past its 2026-06-30 discontinuation date on Vertex).
## Where existing SDKs are weak and we should not be
@@ -54,7 +54,7 @@ Speech.request({ model: openai.speech("gpt-4o-mini-tts"), text })
Transcription.request({ model: openai.transcription("gpt-4o-transcribe"), audio })
```
The request namespace and the selector share one word (`Image.request` + `.image(...)`). That redundancy is accepted: a callable facade returning a lazily resolved ref would be a second way to construct the same model, and the type machinery to infer `providerOptions` through it is not worth one word. Where a provider has two APIs for one modality, the selectors stay explicit (`openai.chat`, a future `google.imagen`), and one default per modality per provider is part of the facade definition (OpenAI image → Images API, Google image → Gemini-native since Imagen on the Gemini API shuts down 2026-08-17). Provider package entrypoints keep `model(modelID, settings)` per modality-specific path, e.g. `@opencode/ai/providers/openai/responses`.
The request namespace and the selector share one word (`Image.request` + `.image(...)`). That redundancy is accepted: a callable facade returning a lazily resolved ref would be a second way to construct the same model, and the type machinery to infer `providerOptions` through it is not worth one word. Where a provider has two routes for one modality, the selectors stay explicit (`openai.chat`, `stability.image` inline vs `stability.upscale()` queued), and one default per modality per provider is part of the facade definition (OpenAI image → Images API, Google image → Gemini-native; Imagen is shut down, so there is no `google.imagen`). Provider package entrypoints keep `model(modelID, settings)` per modality-specific path, e.g. `@opencode/ai/providers/openai/responses`.
### `Media` — the asset type
@@ -128,6 +128,8 @@ yield* Image.stream(request) // Stream<ImageEvent>
Editing is not a separate function; `images`/`mask` on the request select the edit path in the route (OpenAI `/images/edits`, Gemini multimodal parts, xAI `/images/edits`). Routes that cannot honor `mask` fail with `Unsupported`.
`ImageRoute` is the inline | stream | queued union, dispatched on `route.kind`. `Image.stream` on a streaming route emits `image-partial` previews before each `image`; on a queued route it emits `generation-queued` / `generation-progress` observations, then the result's `image` and `finish` events.
#### Video
Shipped in phase 2 (`src/video.ts`, `src/video-client.ts`, protocols `google-video`, `xai-video`, `fal-video`, `runway-video`).
@@ -385,15 +387,19 @@ Existing facades gain per-modality selectors; the modality routes each facade pr
| Facade | llm | image | video | speech | transcription | other |
|---|---|---|---|---|---|---|
| `OpenAI` | responses (default), chat | Images API | Sora (deprecated 2026-09-24) | ✓ | ✓ | |
| `Google` | Gemini | Gemini-native (default), `imagen` | Veo | Gemini TTS | `gemini-3.5-transcribe` | |
| `OpenAI` | responses (default), chat | Images API (stream) | Sora (deprecated 2026-09-24) | ✓ | ✓ | |
| `Google` | Gemini | Gemini-native | Veo | Gemini TTS | `gemini-3.5-transcribe` | |
| `XAI` | ✓ | ✓ | ✓ | | | |
| `ElevenLabs` | | | | ✓ | Scribe | soundEffect, music |
| `Cartesia` | | | | ✓ | | |
| `Deepgram` | | | | Aura | ✓ | |
| `Fal` | | ✓ | ✓ | | | |
| `Fal` | | ✓ (queued) | ✓ | | | |
| `AssemblyAI` | | | | | ✓ (queued) | |
| `Replicate`, `Runway`, `Luma`, `Kling`, `MiniMax`, `BlackForestLabs`, `Stability` | | per provider | | | | |
| `BlackForestLabs` | | ✓ (queued) | | | | |
| `Replicate` | | ✓ (queued) | | | | |
| `Stability` | | `image` (inline), `upscale()` (queued) | | | | |
| `Runway` | | | ✓ | | | |
| `Luma`, `Kling`, `MiniMax` | | per provider | | | | |
New facades follow the existing one-file-per-provider rule. Package entrypoints are modality-specific, such as `@opencode/ai/providers/openai/images`, and return the concrete model.
@@ -438,7 +444,7 @@ Foundation + Image ship together as the reference implementation, serially. Vide
1. **Foundation** — per-modality selectors, `Media`, `Generation`, `Poll`, `Usage` union, `MediaProtocol` kinds, `@opencode/ai/promise` with `llm` + `image`. Port the five existing image protocols onto it. Unify `MediaPart` and add the `media` LLM event (fixes Gemini image output being dropped).
2. **Video** — ✅ Veo, xAI, fal, Runway shipped (`MediaProtocol.queued`, `Video.start/generate/resume/stream`, promise `ai.video`). Deferred: `Video.complete` (webhooks), Luma, Kling, MiniMax, Replicate.
3. **Speech + Transcription** — ✅ Speech: OpenAI, Gemini TTS, ElevenLabs, Cartesia, Deepgram shipped (`MediaProtocol.stream`, `Speech.generate/stream`, promise `ai.speech`). ✅ Transcription: OpenAI, Gemini, Deepgram, AssemblyAI shipped across all three route kinds (`Transcription.generate/stream/start/resume`, promise `ai.transcription`). Pending: ElevenLabs Scribe. Deferred: `Speech.session` and `Transcription.session` (WebSocket streaming).
4. **Image queued routes and partials** — BFL, fal, Replicate, Stability; OpenAI `partial_images` streaming.
4. **Image queued routes and partials** — ✅ BFL, fal, Replicate, and Stability creative upscale queued; Stability generate inline; OpenAI `partial_images` streaming (`image-partial` restored). Imagen dropped: shut down on the Gemini API and discontinued on Vertex (2026-06-30). Deferred: Stability's synchronous edit and fast/conservative upscale endpoints.
5. **Later** — ElevenLabs music/SFX, Lyria, `Speech.session` / `Transcription.session`, realtime.
Core adoption (session attachments beyond png/jpeg/gif/webp/pdf, image-generation tool, TUI rendering) comes after phase 1 and is a Core concern.
+29 -1
View File
@@ -108,7 +108,7 @@ const PROVIDERS: ReadonlyArray<Provider> = [
id: "fal",
label: "fal",
tier: "canary",
note: "fal queue video recorded tests",
note: "fal queue image and video recorded tests",
vars: [{ name: "FAL_KEY" }],
// fal has no free authenticated list endpoint; a 404 for an unknown request id proves the key was accepted.
validate: (env) =>
@@ -123,6 +123,34 @@ const PROVIDERS: ReadonlyArray<Provider> = [
return yield* responseError(response)
}),
},
{
id: "black-forest-labs",
label: "Black Forest Labs",
tier: "canary",
note: "BFL FLUX queued image recorded tests",
vars: [{ name: "BFL_API_KEY" }],
validate: (env) =>
HttpClientRequest.get("https://api.bfl.ai/v1/credits").pipe(
HttpClientRequest.setHeader("x-key", Redacted.value(Redacted.make(env.BFL_API_KEY))),
executeRequest,
),
},
{
id: "replicate",
label: "Replicate",
tier: "canary",
note: "Replicate prediction image recorded tests",
vars: [{ name: "REPLICATE_API_TOKEN" }],
validate: (env) => validateBearer("https://api.replicate.com/v1/account", Redacted.make(env.REPLICATE_API_TOKEN)),
},
{
id: "stability",
label: "Stability AI",
tier: "canary",
note: "Stability inline generate and queued upscale recorded tests",
vars: [{ name: "STABILITY_API_KEY" }],
validate: (env) => validateBearer("https://api.stability.ai/v1/user/balance", Redacted.make(env.STABILITY_API_KEY)),
},
{
id: "runway",
label: "Runway",
+44 -8
View File
@@ -1,9 +1,12 @@
import { Context, Effect, Layer, Stream } from "effect"
import type { AwaitOptions, Generation } from "./generation.js"
import { RequestExecutor } from "./route/executor.js"
import { MediaRoute } from "./route/media.js"
import type { AIError } from "./schema/index.js"
import {
responseEvents,
type ImageEvent,
type ImageModel,
type ImageOptions,
type ImageRequestFor,
type ImageResponse,
@@ -12,43 +15,74 @@ import {
export interface Interface {
readonly generate: <Options extends ImageOptions>(
request: ImageRequestFor<Options>,
options?: AwaitOptions,
) => Effect.Effect<ImageResponse, AIError>
readonly stream: <Options extends ImageOptions>(
request: ImageRequestFor<Options>,
options?: AwaitOptions,
) => Stream.Stream<ImageEvent, AIError>
readonly start: <Options extends ImageOptions>(
request: ImageRequestFor<Options>,
) => Effect.Effect<Generation<ImageResponse>, AIError>
readonly resume: <Options extends ImageOptions>(
model: ImageModel<Options>,
token: unknown,
) => Effect.Effect<Generation<ImageResponse>, AIError>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/ImageClient") {}
export const generate = <Options extends ImageOptions>(
request: ImageRequestFor<Options>,
options?: AwaitOptions,
): Effect.Effect<ImageResponse, AIError, Service> =>
Effect.gen(function* () {
const client = yield* Service
return yield* client.generate(request)
return yield* client.generate(request, options)
})
export const stream = <Options extends ImageOptions>(
request: ImageRequestFor<Options>,
options?: AwaitOptions,
): Stream.Stream<ImageEvent, AIError, Service> =>
Stream.unwrap(
Effect.gen(function* () {
const client = yield* Service
return client.stream(request)
return client.stream(request, options)
}),
)
export const start = <Options extends ImageOptions>(
request: ImageRequestFor<Options>,
): Effect.Effect<Generation<ImageResponse>, AIError, Service> =>
Effect.gen(function* () {
const client = yield* Service
return yield* client.start(request)
})
export const resume = <Options extends ImageOptions>(
model: ImageModel<Options>,
token: unknown,
): Effect.Effect<Generation<ImageResponse>, AIError, Service> =>
Effect.gen(function* () {
const client = yield* Service
return yield* client.resume(model, token)
})
export const layer: Layer.Layer<Service, never, RequestExecutor.Service> = Layer.effect(
Service,
Effect.gen(function* () {
const executor = yield* RequestExecutor.Service
const generate = <Options extends ImageOptions>(request: ImageRequestFor<Options>) =>
request.model.route.generate(request, executor.execute)
const dispatch = MediaRoute.dispatch<ImageEvent, ImageResponse>({
modality: "image",
execute: executor.execute,
responseEvents,
})
return Service.of({
generate,
// Inline routes have no partial frames yet; the stream is the completed response expanded into events.
stream: (request) =>
Stream.fromIterableEffect(Effect.map(generate(request), responseEvents)),
start: (request) => dispatch.start(request.model.route, request),
resume: (model, token) => dispatch.resume(model.route, model, token),
generate: (request, options) => dispatch.generate(request.model.route, request, options),
stream: (request, options) => dispatch.stream(request.model.route, request, options),
})
}),
)
@@ -58,4 +92,6 @@ export const ImageClient = {
layer,
generate,
stream,
start,
resume,
} as const
+97 -15
View File
@@ -1,6 +1,7 @@
import { Effect, Schema, Stream } from "effect"
import { Generation, ProgressEvent, QueuedEvent, type AwaitOptions } from "./generation.js"
import { Media } from "./media.js"
import { MediaModel, composeRoute, tryRequest } from "./media-model.js"
import { MediaModel, composeAnyRoute, tryRequest } from "./media-model.js"
import { MediaRoute } from "./route/media.js"
import type { MediaProtocol } from "./route/media-protocol.js"
import { AIError, HttpOptions, MediaUsage, ProviderMetadata } from "./schema/index.js"
@@ -12,8 +13,9 @@ import { ImageClient, Service } from "./image-client.js"
export type ImageOptions = Record<string, unknown>
export type ImageRoute<Options extends ImageOptions = ImageOptions> = MediaRoute.Route<
export type ImageRoute<Options extends ImageOptions = ImageOptions> = MediaRoute.AnyRoute<
ImageRequestFor<Options>,
ImageEvent,
ImageResponse
>
@@ -24,25 +26,58 @@ export class ImageModel<Options extends ImageOptions = ImageOptions> extends Med
return new ImageModel<Options>(input)
}
/** Compose an inline image protocol with its canonical path into a model for one deployment. */
static fromRoute<Options extends ImageOptions = ImageOptions>(
route: ImageModel.RouteInput<Options>,
/** The number of type arguments selects the kind: `<Options>`, `<Options, Frame, State>`, or `<Options, Token>`. */
static fromRoute<Options extends ImageOptions>(
route: ImageModel.InlineRouteInput<Options>,
input: MediaRoute.ModelInput,
): ImageModel<Options>
static fromRoute<Options extends ImageOptions, Frame, State>(
route: ImageModel.StreamRouteInput<Options, Frame, State>,
input: MediaRoute.ModelInput,
): ImageModel<Options>
static fromRoute<Options extends ImageOptions, Token>(
route: ImageModel.QueuedRouteInput<Options, Token>,
input: MediaRoute.ModelInput,
): ImageModel<Options>
static fromRoute<Options extends ImageOptions, Frame, State, Token>(
route: ImageModel.RouteInput<Options, Frame, State, Token>,
input: MediaRoute.ModelInput,
) {
return new ImageModel<Options>({
id: input.id,
provider: route.provider,
http: input.http,
route: composeRoute(MediaRoute.inline, route, input),
route: composeAnyRoute(route, input, collectResponse),
})
}
}
export namespace ImageModel {
export type RouteInput<Options extends ImageOptions = ImageOptions> = MediaModel.RouteInput<
export type InlineRouteInput<Options extends ImageOptions = ImageOptions> = MediaModel.RouteInput<
ImageRequestFor<Options>,
MediaProtocol.Inline<ImageRequestFor<Options>, ImageResponse>
>
export type StreamRouteInput<
Options extends ImageOptions = ImageOptions,
Frame = unknown,
State = unknown,
> = MediaModel.RouteInput<
MediaProtocol.Addressed<ImageRequestFor<Options>>,
MediaProtocol.Streamed<ImageRequestFor<Options>, ImageEvent, Frame, State>
>
export type QueuedRouteInput<Options extends ImageOptions = ImageOptions, Token = unknown> = MediaModel.RouteInput<
ImageRequestFor<Options>,
MediaProtocol.Queued<ImageRequestFor<Options>, ImageResponse, Token>
>
export type RouteInput<
Options extends ImageOptions = ImageOptions,
Frame = unknown,
State = unknown,
Token = unknown,
> = MediaModel.AnyRouteInput<ImageRequestFor<Options>, ImageEvent, ImageResponse, Frame, State, Token>
}
export const ImageModelSchema = Schema.declare((value): value is ImageModel => value instanceof ImageModel, {
@@ -117,6 +152,13 @@ export class ImageResponse extends Schema.Class<ImageResponse>("Image.Response")
}
}
/** The `index`-th lower-fidelity preview of an image still generating; the `image` event supersedes it. */
export const ImagePartialEvent = Schema.Struct({
type: Schema.tag("image-partial"),
index: Schema.Number,
image: Media.AssetSchema,
}).annotate({ identifier: "Image.Event.Partial" })
export const ImageOutputEvent = Schema.Struct({
type: Schema.tag("image"),
index: Schema.Number,
@@ -130,16 +172,24 @@ export const ImageFinishEvent = Schema.Struct({
providerMetadata: Schema.optional(ProviderMetadata),
}).annotate({ identifier: "Image.Event.Finish" })
const imageEventTagged = Schema.Union([ImageOutputEvent, ImageFinishEvent]).pipe(Schema.toTaggedUnion("type"))
const imageEventTagged = Schema.Union([
QueuedEvent,
ProgressEvent,
ImagePartialEvent,
ImageOutputEvent,
ImageFinishEvent,
]).pipe(Schema.toTaggedUnion("type"))
export const ImageEvent = Object.assign(imageEventTagged, {
is: {
generationQueued: imageEventTagged.guards["generation-queued"],
generationProgress: imageEventTagged.guards["generation-progress"],
imagePartial: imageEventTagged.guards["image-partial"],
image: imageEventTagged.guards.image,
finish: imageEventTagged.guards.finish,
},
})
export type ImageEvent = Schema.Schema.Type<typeof imageEventTagged>
/** Inline routes produce every image at once; expand the response into the streaming event shape. */
export const responseEvents = (response: ImageResponse): ReadonlyArray<ImageEvent> => [
...response.images.map((image, index) => ImageOutputEvent.make({ index, image })),
ImageFinishEvent.make({
@@ -149,6 +199,20 @@ export const responseEvents = (response: ImageResponse): ReadonlyArray<ImageEven
}),
]
const collectResponse = (events: ReadonlyArray<ImageEvent>): Effect.Effect<ImageResponse> => {
const finish = events.find(ImageEvent.is.finish)
// Every image 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 image stream completed without a finish event"))
return Effect.succeed(
new ImageResponse({
images: events.filter(ImageEvent.is.image).map((event) => event.image),
usage: finish.usage,
notices: finish.notices,
providerMetadata: finish.providerMetadata,
}),
)
}
// ---------------------------------------------------------------------------
// Request-shaped call API
// ---------------------------------------------------------------------------
@@ -169,22 +233,40 @@ const requestEffect = (input: ImageRequest | ImageRequestInput) => tryRequest(()
export function generate<const Model extends ImageModel>(
input: ImageRequestInput<Model>,
options?: AwaitOptions,
): Effect.Effect<ImageResponse, AIError, Service>
export function generate(input: ImageRequest): Effect.Effect<ImageResponse, AIError, Service>
export function generate(input: ImageRequest | ImageRequestInput) {
return requestEffect(input).pipe(Effect.flatMap((request) => ImageClient.generate(request)))
export function generate(input: ImageRequest, options?: AwaitOptions): Effect.Effect<ImageResponse, AIError, Service>
export function generate(input: ImageRequest | ImageRequestInput, options?: AwaitOptions) {
return requestEffect(input).pipe(Effect.flatMap((request) => ImageClient.generate(request, options)))
}
export function stream<const Model extends ImageModel>(
input: ImageRequestInput<Model>,
options?: AwaitOptions,
): Stream.Stream<ImageEvent, AIError, Service>
export function stream(input: ImageRequest): Stream.Stream<ImageEvent, AIError, Service>
export function stream(input: ImageRequest | ImageRequestInput) {
return Stream.unwrap(requestEffect(input).pipe(Effect.map((request) => ImageClient.stream(request))))
export function stream(input: ImageRequest, options?: AwaitOptions): Stream.Stream<ImageEvent, AIError, Service>
export function stream(input: ImageRequest | ImageRequestInput, options?: AwaitOptions) {
return Stream.unwrap(requestEffect(input).pipe(Effect.map((request) => ImageClient.stream(request, options))))
}
/** Inline and streaming routes fail with `UnsupportedOperation`. */
export function start<const Model extends ImageModel>(
input: ImageRequestInput<Model>,
): Effect.Effect<Generation<ImageResponse>, AIError, Service>
export function start(input: ImageRequest): Effect.Effect<Generation<ImageResponse>, AIError, Service>
export function start(input: ImageRequest | ImageRequestInput) {
return requestEffect(input).pipe(Effect.flatMap((request) => ImageClient.start(request)))
}
export const resume = <Options extends ImageOptions>(
model: ImageModel<Options>,
token: unknown,
): Effect.Effect<Generation<ImageResponse>, AIError, Service> => ImageClient.resume(model, token)
export const Image = {
request,
generate,
stream,
start,
resume,
} as const
+29 -1
View File
@@ -1,6 +1,6 @@
import { Effect } from "effect"
import { Endpoint } from "./route/endpoint.js"
import type { MediaRoute } from "./route/media.js"
import { MediaRoute } from "./route/media.js"
import type { MediaProtocol } from "./route/media-protocol.js"
import { AIError, HttpOptions, InvalidRequestError, ModelID, ProviderID } from "./schema/index.js"
@@ -42,6 +42,11 @@ export namespace MediaModel {
/** Headers the protocol requires on every call, such as a pinned API version; deployment headers win. */
readonly headers?: Record<string, string>
}
export type AnyRouteInput<Request extends MediaRoute.MediaRequest, Event, Response, Frame, State, Token> =
| RouteInput<Request, MediaProtocol.Inline<Request, Response>>
| RouteInput<MediaProtocol.Addressed<Request>, MediaProtocol.Streamed<Request, Event, Frame, State>>
| RouteInput<Request, MediaProtocol.Queued<Request, Response, Token>>
}
/** Compose a protocol route input with one deployment through `MediaRoute.inline`, `queued`, or `stream`. */
@@ -60,6 +65,29 @@ export const composeRoute = <Request extends MediaRoute.MediaRequest, Protocol,
route.headers === undefined && input.headers === undefined ? undefined : { ...route.headers, ...input.headers },
})
export const composeAnyRoute = <Request extends MediaRoute.MediaRequest, Event, Response, Frame, State, Token>(
route: MediaModel.AnyRouteInput<Request, Event, Response, Frame, State, Token>,
input: MediaRoute.ModelInput,
collect: (events: ReadonlyArray<Event>) => Effect.Effect<Response, AIError>,
): MediaRoute.AnyRoute<Request, Event, Response> => {
if (isStreamInput(route))
return composeRoute((composition) => MediaRoute.stream({ ...composition, collect }), route, input)
if (isQueuedInput(route)) return composeRoute(MediaRoute.queued, route, input)
return composeRoute(MediaRoute.inline, route, input)
}
const isStreamInput = <Request extends MediaRoute.MediaRequest, Event, Response, Frame, State, Token>(
route: MediaModel.AnyRouteInput<Request, Event, Response, Frame, State, Token>,
): route is MediaModel.RouteInput<
MediaProtocol.Addressed<Request>,
MediaProtocol.Streamed<Request, Event, Frame, State>
> => route.protocol.kind === "stream"
const isQueuedInput = <Request extends MediaRoute.MediaRequest, Event, Response, Frame, State, Token>(
route: MediaModel.AnyRouteInput<Request, Event, Response, Frame, State, Token>,
): route is MediaModel.RouteInput<Request, MediaProtocol.Queued<Request, Response, Token>> =>
route.protocol.kind === "queued"
/** Lift a synchronous Schema-class constructor into a typed `InvalidRequest` failure. */
export const tryRequest = <A>(make: () => A): Effect.Effect<A, AIError> =>
Effect.try({
+11 -5
View File
@@ -1,6 +1,6 @@
import { Effect, Layer, ManagedRuntime, Stream } from "effect"
import type { AwaitOptions, Generation, Snapshot } from "./generation.js"
import { Image, ImageModel, ImageRequest, type ImageRequestInput } from "./image.js"
import { Image, ImageModel, ImageRequest, type ImageOptions, type ImageRequestInput } from "./image.js"
import { ImageClient } from "./image-client.js"
import { LLM } from "./index.js"
import { LLMClient } from "./route/client.js"
@@ -123,10 +123,16 @@ export const make = (options: Options = {}) => {
request: Image.request,
generate: <const Model extends ImageModel>(
input: ImageRequestInput<Model> | ImageRequest,
options?: RunOptions,
) => run(Image.generate(imageRequest(input)), options),
stream: <const Model extends ImageModel>(input: ImageRequestInput<Model> | ImageRequest, options?: RunOptions) =>
iterate(Image.stream(imageRequest(input)), options),
options?: AwaitOptions & RunOptions,
) => run(Image.generate(imageRequest(input), { poll: options?.poll }), options),
stream: <const Model extends ImageModel>(
input: ImageRequestInput<Model> | ImageRequest,
options?: AwaitOptions & RunOptions,
) => iterate(Image.stream(imageRequest(input), { poll: options?.poll }), options),
start: <const Model extends ImageModel>(input: ImageRequestInput<Model> | ImageRequest, options?: RunOptions) =>
run(Image.start(imageRequest(input)), options).then(handle),
resume: <Options extends ImageOptions>(model: ImageModel<Options>, token: unknown, options?: RunOptions) =>
run(Image.resume(model, token), options).then(handle),
},
video: {
request: Video.request,
+223
View File
@@ -0,0 +1,223 @@
import { Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import type { Status } from "../generation.js"
import { ImageModel, ImageResponse, type ImageRequestFor } from "../image.js"
import { Media } from "../media.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { ProviderShared, optionalNull } from "./shared.js"
import { MediaInput } from "./utils/media-input.js"
const ADAPTER = "bfl-images"
const NAME = "Black Forest Labs"
const PROVIDER = ProviderID.make("black-forest-labs")
export const DEFAULT_BASE_URL = "https://api.bfl.ai"
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type BlackForestLabsImageOptions = {
readonly safety_tolerance?: number
readonly prompt_upsampling?: boolean
readonly disable_pup?: boolean
readonly raw?: boolean
readonly guidance?: number
readonly steps?: number
} & Record<string, unknown>
export type Request = ImageRequestFor<BlackForestLabsImageOptions>
// ---------------------------------------------------------------------------
// 2. Token and response schemas
// ---------------------------------------------------------------------------
/** Regional clusters answer on different hosts, so the returned `polling_url` is followed verbatim. */
export const Token = Schema.Struct({ id: Schema.String, pollingURL: Schema.String })
export type Token = Schema.Schema.Type<typeof Token>
const StartResponse = Schema.Struct({
id: Schema.String,
polling_url: Schema.String,
})
const Result = Schema.Struct({
id: Schema.String,
status: Schema.String,
result: optionalNull(
Schema.StructWithRest(
Schema.Struct({ sample: Schema.String, seed: optionalNull(Schema.Number), prompt: optionalNull(Schema.String) }),
[Schema.Record(Schema.String, Schema.Unknown)],
),
),
cost: optionalNull(Schema.Number),
})
const STATUS = {
Pending: "running",
Reasoning: "running",
Generating: "running",
Ready: "completed",
Error: "failed",
// Moderation is terminal; `decodeResult` reports it as a content-policy failure.
"Content Moderated": "failed",
"Request Moderated": "failed",
"Task not found": "expired",
} as const satisfies Record<string, Status>
const isModerated = (status: string) => status === "Content Moderated" || status === "Request Moderated"
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
interface Capabilities {
readonly sizing: "dimensions" | "aspectRatio" | "none"
/** `input_image` numbers extra references `input_image_2`…; the other fields take one image. */
readonly imageField: "input_image" | "image_prompt" | "image"
readonly maxImages: number
readonly mask: boolean
}
const capabilities = (model: string): Capabilities => {
if (model.startsWith("flux-pro-1.0-fill")) return { sizing: "none", imageField: "image", maxImages: 1, mask: true }
if (model.startsWith("flux-pro-1.0-expand")) return { sizing: "none", imageField: "image", maxImages: 1, mask: false }
if (model.startsWith("flux-kontext"))
return { sizing: "aspectRatio", imageField: "input_image", maxImages: 4, mask: false }
if (model.startsWith("flux-pro-1.1-ultra"))
return { sizing: "aspectRatio", imageField: "image_prompt", maxImages: 1, mask: false }
if (model.startsWith("flux-pro-1.1") || model.startsWith("flux-dev"))
return { sizing: "dimensions", imageField: "image_prompt", maxImages: 1, mask: false }
if (model.startsWith("flux-2-klein"))
return { sizing: "dimensions", imageField: "input_image", maxImages: 4, mask: false }
return { sizing: "dimensions", imageField: "input_image", maxImages: 8, mask: false }
}
const unsupported = (model: string, field: string, message: string) =>
ProviderShared.unsupportedOperation({
operation: `media.${field}`,
provider: PROVIDER,
route: ADAPTER,
message: `${model} ${message}`,
})
const validate = (request: Request, model: Capabilities) => {
const id = request.model.id
const images = request.images?.length ?? 0
if (request.n !== undefined && request.n > 1)
return Effect.fail(unsupported(id, "n", "generates one image per request; call it once per image"))
if (request.size !== undefined && model.sizing !== "dimensions")
return Effect.fail(unsupported(id, "size", "does not take size (width and height)"))
if (request.aspectRatio !== undefined && model.sizing !== "aspectRatio")
return Effect.fail(unsupported(id, "aspectRatio", "does not take aspectRatio"))
if (images > model.maxImages) return Effect.fail(unsupported(id, "images", `takes at most ${model.maxImages} images`))
if (request.mask !== undefined && !model.mask)
return Effect.fail(unsupported(id, "mask", "does not inpaint; use flux-pro-1.0-fill"))
return Effect.void
}
const imageInput = (asset: Media.Asset) => {
const value = asset.inline()?.base64 ?? ProviderShared.mediaUrl(asset)
if (value === undefined)
return Effect.fail(ProviderShared.invalidRequest(`${NAME} accepts inline images or https URLs`))
return Effect.succeed(value)
}
const fromRequest = Effect.fn("BlackForestLabsImages.fromRequest")(function* (request: Request) {
const model = capabilities(request.model.id)
yield* validate(request, model)
const images = yield* Effect.forEach(request.images ?? [], imageInput)
const fields = images.map((image, index) => [
index === 0 ? model.imageField : `${model.imageField}_${index + 1}`,
image,
])
return MediaProtocol.json(
mergeJsonRecords(
{
prompt: request.prompt,
...(request.size === undefined ? {} : MediaInput.dimensions(request.size)),
aspect_ratio: request.aspectRatio,
seed: request.seed,
output_format: request.format,
mask: request.mask === undefined ? undefined : yield* imageInput(request.mask),
...Object.fromEntries(fields),
},
request.providerOptions,
request.http?.body,
) ?? {},
)
})
// ---------------------------------------------------------------------------
// 6. Response decoding
// ---------------------------------------------------------------------------
const decodeStart = MediaProtocol.decodeStarted(ADAPTER, NAME, StartResponse, (value) => ({
token: { id: value.id, pollingURL: value.polling_url },
snapshot: { id: value.id, status: "queued" },
}))
const decodeDocument = MediaProtocol.decodeJson(ADAPTER, NAME, Result)
const decodeStatus = Effect.fn("BlackForestLabsImages.decodeStatus")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) {
const output = yield* decodeDocument(response)
return { id: context.token.id, status: yield* MediaProtocol.status(STATUS, output.value.status, output) }
})
const decodeResult = Effect.fn("BlackForestLabsImages.decodeResult")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) {
const output = yield* decodeDocument(response)
const document = output.value
const status = yield* MediaProtocol.status(STATUS, document.status, output)
if (isModerated(document.status)) return yield* output.contentPolicy(`${NAME} moderated the generation`)
if (status === "failed" || status === "expired")
return yield* output.ended(status, `${NAME} generation ${context.token.id} ended with ${document.status}`)
if (status !== "completed" || document.result === undefined || document.result === null)
return yield* output.invalid(`${NAME} generation ${context.token.id} has no result`)
const { sample, seed, prompt, ...rest } = document.result
return new ImageResponse({
// `sample` is a signed URL that expires 10 minutes after the result is ready, so it is downloaded now.
images: [yield* context.materialize(Media.url(sample))],
usage:
document.cost === undefined || document.cost === null ? undefined : { type: "credits", credits: document.cost },
providerMetadata: {
bfl: { id: context.token.id, seed: seed ?? undefined, prompt: prompt ?? undefined, ...rest },
},
})
})
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.queued<Request, ImageResponse, Token>({
id: ADAPTER,
name: NAME,
token: Token,
start: { body: { from: fromRequest }, decode: decodeStart },
status: { path: (token) => token.pollingURL, decode: decodeStatus },
result: { path: (token) => token.pollingURL, decode: decodeResult },
})
export const model = (input: MediaRoute.ModelInput) =>
ImageModel.fromRoute<BlackForestLabsImageOptions, Token>(
{
id: ADAPTER,
provider: PROVIDER,
protocol,
baseURL: DEFAULT_BASE_URL,
path: ({ request }) => `/v1/${request.model.id}`,
},
input,
)
export const BlackForestLabsImages = {
protocol,
model,
} as const
+168
View File
@@ -0,0 +1,168 @@
import { Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import { ImageModel, ImageResponse, type ImageRequestFor } from "../image.js"
import { Media } from "../media.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { ProviderShared, optionalNull } from "./shared.js"
import { FalQueue } from "./utils/fal-queue.js"
import { MediaInput } from "./utils/media-input.js"
const ADAPTER = "fal-images"
const NAME = "fal Images"
const PROVIDER = ProviderID.make("fal")
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type FalImageOptions = {
readonly image_size?:
| "square_hd"
| "square"
| "portrait_4_3"
| "portrait_16_9"
| "landscape_4_3"
| "landscape_16_9"
| (string & {})
readonly enable_safety_checker?: boolean
} & Record<string, unknown>
export type Request = ImageRequestFor<FalImageOptions>
// ---------------------------------------------------------------------------
// 2. Response schema
// ---------------------------------------------------------------------------
const QueueResult = Schema.StructWithRest(
Schema.Struct({
images: Schema.Array(
Schema.Struct({
url: Schema.String,
width: optionalNull(Schema.Number),
height: optionalNull(Schema.Number),
content_type: optionalNull(Schema.String),
}),
),
seed: optionalNull(Schema.Number),
has_nsfw_concepts: optionalNull(Schema.Array(Schema.Boolean)),
}),
[Schema.Record(Schema.String, Schema.Unknown)],
)
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
const sizing = (model: string) => {
if (/^fal-ai\/(nano-banana|flux-pro\/v1\.1-ultra)/.test(model)) return "aspect_ratio"
if (model.startsWith("fal-ai/flux")) return "image_size"
return undefined
}
const unsupported = (model: string, field: string, message: string) =>
ProviderShared.unsupportedOperation({
operation: `media.${field}`,
provider: PROVIDER,
route: ADAPTER,
message: `${model} ${message}`,
})
const validate = (request: Request) => {
const id = request.model.id
const field = sizing(id)
if (request.size !== undefined && request.aspectRatio !== undefined)
return Effect.fail(ProviderShared.invalidRequest(`${NAME} accepts either size or aspectRatio, not both`))
if (request.size !== undefined && field === "aspect_ratio")
return Effect.fail(unsupported(id, "size", "sizes by aspectRatio"))
if (request.aspectRatio !== undefined && field === "image_size")
return Effect.fail(unsupported(id, "aspectRatio", "sizes by size (image_size)"))
if ((request.images?.length ?? 0) > 1 && !isEdit(id))
return Effect.fail(unsupported(id, "images", "takes one image_url; use an /edit endpoint for several images"))
return Effect.void
}
// `/edit` endpoints take an `image_urls` list; image-to-image, fill, and Ultra take one `image_url` (beside `mask_url`).
const isEdit = (model: string) => model.endsWith("/edit")
const fromRequest = Effect.fn("FalImages.fromRequest")(function* (request: Request) {
yield* validate(request)
const images = yield* Effect.forEach(request.images ?? [], (image) => FalQueue.mediaUrl(image, NAME))
const edit = isEdit(request.model.id)
return MediaProtocol.json(
mergeJsonRecords(
{
prompt: request.prompt,
num_images: request.n,
seed: request.seed,
image_size: request.size === undefined ? undefined : MediaInput.dimensions(request.size),
aspect_ratio: request.aspectRatio,
output_format: request.format,
image_urls: edit && images.length > 0 ? images : undefined,
image_url: edit ? undefined : images[0],
mask_url: request.mask === undefined ? undefined : yield* FalQueue.mediaUrl(request.mask, NAME),
},
request.providerOptions,
request.http?.body,
) ?? {},
)
})
// ---------------------------------------------------------------------------
// 6. Response decoding
// ---------------------------------------------------------------------------
const decodeQueueResult = MediaProtocol.decodeJson(ADAPTER, NAME, QueueResult)
const decodeResult = Effect.fn("FalImages.decodeResult")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<FalQueue.Token>,
) {
const output = yield* decodeQueueResult(response)
const { images, seed, has_nsfw_concepts, ...rest } = output.value
if (images.length === 0) return yield* output.invalid(`${NAME} returned no images`)
// With the safety checker on, flagged images come back blacked out rather than omitted.
const flagged = (has_nsfw_concepts ?? []).flatMap((value, index) => (value ? [index] : []))
return new ImageResponse({
images: images.map((image) =>
Media.url(image.url, {
mediaType: image.content_type ?? undefined,
info: { width: image.width ?? undefined, height: image.height ?? undefined },
}),
),
notices:
flagged.length === 0
? undefined
: flagged.map((index) => ({ type: "moderated" as const, message: `${NAME} flagged image ${index} as NSFW` })),
providerMetadata: { fal: { requestId: context.token.requestID, seed: seed ?? undefined, ...rest } },
})
})
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = FalQueue.protocol<Request, ImageResponse>({
id: ADAPTER,
name: NAME,
from: fromRequest,
decodeResult,
})
export const model = (input: MediaRoute.ModelInput) =>
ImageModel.fromRoute<FalImageOptions, FalQueue.Token>(
{
id: ADAPTER,
provider: PROVIDER,
protocol,
baseURL: FalQueue.DEFAULT_BASE_URL,
path: ({ request }) => `/${request.model.id}`,
},
input,
)
export const FalImages = {
protocol,
model,
} as const
+11 -74
View File
@@ -1,17 +1,16 @@
import { Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import type { Status } from "../generation.js"
import { Media } from "../media.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { VideoModel, VideoResponse, type VideoRequestFor } from "../video.js"
import { ProviderShared, optionalNull } from "./shared.js"
import { FalQueue } from "./utils/fal-queue.js"
const ADAPTER = "fal-video"
const NAME = "fal Video"
const PROVIDER = ProviderID.make("fal")
export const DEFAULT_BASE_URL = "https://queue.fal.run"
// ---------------------------------------------------------------------------
// 1. Public model input
@@ -31,32 +30,9 @@ export type FalVideoOptions = {
export type Request = VideoRequestFor<FalVideoOptions>
// ---------------------------------------------------------------------------
// 2. Token and response schemas
// 2. Response schema
// ---------------------------------------------------------------------------
/** fal hands back absolute follow-up URLs on submit; they are authoritative for status, result, and cancel. */
export const Token = Schema.Struct({
requestID: Schema.String,
statusURL: Schema.String,
responseURL: Schema.String,
cancelURL: Schema.String,
})
export type Token = Schema.Schema.Type<typeof Token>
const StartResponse = Schema.Struct({
request_id: Schema.String,
status_url: Schema.String,
response_url: Schema.String,
cancel_url: Schema.String,
queue_position: optionalNull(Schema.Number),
})
const QueueStatus = Schema.Struct({
status: Schema.String,
queue_position: optionalNull(Schema.Number),
error: optionalNull(Schema.Unknown),
})
const QueueResult = Schema.StructWithRest(
Schema.Struct({
video: Schema.Struct({
@@ -70,20 +46,10 @@ const QueueResult = Schema.StructWithRest(
[Schema.Record(Schema.String, Schema.Unknown)],
)
const STATUS = {
IN_QUEUE: "queued",
IN_PROGRESS: "running",
COMPLETED: "completed",
} as const satisfies Record<string, Status>
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
// fal accepts public URLs and data URIs; there is no provider file handle to forward.
const mediaUrl = (asset: Media.Asset) =>
ProviderShared.mediaReference(asset, undefined, NAME).pipe(Effect.map((reference) => reference.value))
const fromRequest = Effect.fn("FalVideo.fromRequest")(function* (request: Request) {
if (request.frames?.last !== undefined)
return yield* ProviderShared.unsupportedOperation({
@@ -92,8 +58,9 @@ const fromRequest = Effect.fn("FalVideo.fromRequest")(function* (request: Reques
route: ADAPTER,
message: `${NAME} names the last frame per model; pass it through providerOptions (e.g. end_image_url) instead of frames.last`,
})
const imageUrl = request.frames?.first === undefined ? undefined : yield* mediaUrl(request.frames.first)
const videoUrl = request.video === undefined ? undefined : yield* mediaUrl(request.video)
const imageUrl =
request.frames?.first === undefined ? undefined : yield* FalQueue.mediaUrl(request.frames.first, NAME)
const videoUrl = request.video === undefined ? undefined : yield* FalQueue.mediaUrl(request.video, NAME)
return MediaProtocol.json(
mergeJsonRecords(
{
@@ -116,38 +83,11 @@ const fromRequest = Effect.fn("FalVideo.fromRequest")(function* (request: Reques
// 6. Response decoding
// ---------------------------------------------------------------------------
const decodeStart = MediaProtocol.decodeStarted(ADAPTER, NAME, StartResponse, (value) => ({
token: {
requestID: value.request_id,
statusURL: value.status_url,
responseURL: value.response_url,
cancelURL: value.cancel_url,
},
snapshot: { id: value.request_id, status: "queued", position: value.queue_position ?? undefined },
}))
const decodeQueueStatus = MediaProtocol.decodeJson(ADAPTER, NAME, QueueStatus)
const decodeQueueResult = MediaProtocol.decodeJson(ADAPTER, NAME, QueueResult)
const decodeStatus = Effect.fn("FalVideo.decodeStatus")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) {
const output = yield* decodeQueueStatus(response)
const decoded = output.value
const status = yield* MediaProtocol.status(STATUS, decoded.status, output)
// fal reports request failures as COMPLETED with an `error`; the response endpoint carries the details.
const failed = status === "completed" && decoded.error !== undefined && decoded.error !== null
return {
id: context.token.requestID,
status: failed ? "failed" : status,
position: status === "queued" ? (decoded.queue_position ?? undefined) : undefined,
}
})
const decodeResult = Effect.fn("FalVideo.decodeResult")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
context: MediaProtocol.PollContext<FalQueue.Token>,
) {
const output = yield* decodeQueueResult(response)
const { video, seed, ...rest } = output.value
@@ -169,24 +109,21 @@ const decodeResult = Effect.fn("FalVideo.decodeResult")(function* (
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.queued<Request, VideoResponse, Token>({
export const protocol = FalQueue.protocol<Request, VideoResponse>({
id: ADAPTER,
name: NAME,
token: Token,
unsupported: ["n", "durationSeconds", "references"],
start: { body: { from: fromRequest }, decode: decodeStart },
status: { path: (token) => token.statusURL, decode: decodeStatus },
result: { path: (token) => token.responseURL, decode: decodeResult },
cancel: { method: "PUT", path: (token) => token.cancelURL },
from: fromRequest,
decodeResult,
})
export const model = (input: MediaRoute.ModelInput) =>
VideoModel.fromRoute<FalVideoOptions, Token>(
VideoModel.fromRoute<FalVideoOptions, FalQueue.Token>(
{
id: ADAPTER,
provider: PROVIDER,
protocol,
baseURL: DEFAULT_BASE_URL,
baseURL: FalQueue.DEFAULT_BASE_URL,
path: ({ request }) => `/${request.model.id}`,
},
input,
+151 -45
View File
@@ -1,10 +1,17 @@
import { Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import { ImageModel, ImageResponse, type ImageRequestFor } from "../image.js"
import { Effect, Schema, Stream } from "effect"
import {
ImageFinishEvent,
ImageModel,
ImageOutputEvent,
ImagePartialEvent,
type ImageEvent,
type ImageRequestFor,
} from "../image.js"
import { Media } from "../media.js"
import { Framing } from "../route/framing.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { ProviderID, mergeJsonRecords, type MediaUsage } from "../schema/index.js"
import { ProviderShared } from "./shared.js"
import { MediaInput } from "./utils/media-input.js"
@@ -27,6 +34,8 @@ export type OpenAIImageOptions = {
readonly background?: OpenAIImageString<"auto" | "opaque" | "transparent">
readonly moderation?: OpenAIImageString<"auto" | "low">
readonly outputCompression?: number
/** Previews sent before the final image when streaming (default 2); ignored by `Image.generate`. */
readonly partialImages?: number
} & Record<string, unknown>
export type Request = ImageRequestFor<OpenAIImageOptions>
@@ -35,6 +44,14 @@ export type Request = ImageRequestFor<OpenAIImageOptions>
// 2. Response schema
// ---------------------------------------------------------------------------
const Usage = Schema.Struct({
input_tokens: Schema.optional(Schema.Number),
output_tokens: Schema.optional(Schema.Number),
total_tokens: Schema.optional(Schema.Number),
input_tokens_details: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
output_tokens_details: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
})
const OpenAIImageResponse = Schema.Struct({
data: Schema.Array(
Schema.Struct({
@@ -44,17 +61,44 @@ const OpenAIImageResponse = Schema.Struct({
}),
),
output_format: Schema.optional(Schema.String),
usage: Schema.optional(
Schema.Struct({
input_tokens: Schema.optional(Schema.Number),
output_tokens: Schema.optional(Schema.Number),
total_tokens: Schema.optional(Schema.Number),
input_tokens_details: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
output_tokens_details: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
}),
),
usage: Schema.optional(Usage),
})
// ---------------------------------------------------------------------------
// 3. Streaming event schema
// ---------------------------------------------------------------------------
const StreamEvent = Schema.Union([
Schema.Struct({
type: Schema.Literals(["image_generation.partial_image", "image_edit.partial_image"]),
b64_json: Schema.String,
partial_image_index: Schema.Number,
output_format: Schema.String,
}),
Schema.Struct({
type: Schema.Literals(["image_generation.completed", "image_edit.completed"]),
b64_json: Schema.String,
output_format: Schema.String,
usage: Schema.optional(Usage),
}),
])
const decodeEvent = MediaProtocol.decodeFrame(ADAPTER, NAME, StreamEvent)
const decodeDocument = Schema.decodeUnknownEffect(Schema.fromJsonString(OpenAIImageResponse))
/** `generate` reads the whole JSON response as one frame, with the requested format for responses that omit it. */
type Frame = string | { readonly document: string; readonly requested: string | undefined }
// ---------------------------------------------------------------------------
// 4. Parser state
// ---------------------------------------------------------------------------
interface State {
readonly completed: number
readonly format?: string
readonly usage?: MediaUsage
}
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
@@ -64,26 +108,49 @@ const RESERVED_FORM_FIELDS = new Set(["model", "prompt", "image", "image[]", "im
const nativeOptions = (options: OpenAIImageOptions | undefined) => {
if (!options) return undefined
const { outputCompression, ...native } = options
const { outputCompression, partialImages: _, ...native } = options
return { output_compression: outputCompression, ...native }
}
const streamOptions = (request: MediaProtocol.Addressed<Request>) => {
if (request.mode !== "stream") return Effect.succeed(undefined)
if (request.model.id.startsWith("dall-e"))
return Effect.fail(
ProviderShared.unsupportedOperation({
operation: "media.stream",
provider: PROVIDER,
route: ADAPTER,
message: `${request.model.id} does not stream; use Image.generate or a GPT image model`,
}),
)
if (request.n !== undefined && request.n > 1)
return Effect.fail(
ProviderShared.unsupportedOperation({
operation: "media.n",
provider: PROVIDER,
route: ADAPTER,
message: `${NAME} streams one image; use Image.generate for n=${request.n}`,
}),
)
return Effect.succeed({ stream: true, partial_images: request.providerOptions?.partialImages ?? 2 })
}
const isEdit = (request: Request) => (request.images?.length ?? 0) > 0
const isInline = (asset: Media.Asset) => asset.inline() !== undefined
const isInline = (asset: Media.Asset) => asset.source.type === "bytes" || asset.source.type === "base64"
const reference = (asset: Media.Asset) =>
ProviderShared.mediaReference(asset, PROVIDER, NAME).pipe(
Effect.map((item) => (item.type === "ref" ? { file_id: item.value } : { image_url: item.value })),
)
const fromRequest = Effect.fn("OpenAIImages.fromRequest")(function* (request: Request) {
const fromRequest = Effect.fn("OpenAIImages.fromRequest")(function* (request: MediaProtocol.Addressed<Request>) {
const images = request.images ?? []
const mask = request.mask
if (mask !== undefined && images.length === 0)
return yield* ProviderShared.invalidRequest("An OpenAI image mask requires at least one input image")
const fields = mergeJsonRecords(
{ n: request.n, size: request.size, output_format: request.format },
{ n: request.n, size: request.size, output_format: request.format, ...(yield* streamOptions(request)) },
nativeOptions(request.providerOptions),
request.http?.body,
)
@@ -122,7 +189,7 @@ const fromRequest = Effect.fn("OpenAIImages.fromRequest")(function* (request: Re
})
// ---------------------------------------------------------------------------
// 6. Response decoding
// 6. Stream parsing
// ---------------------------------------------------------------------------
const requestedFormat = (body: MediaProtocol.Body) => {
@@ -131,56 +198,95 @@ const requestedFormat = (body: MediaProtocol.Body) => {
return typeof value === "string" ? value : undefined
}
const decodeResponse = Effect.fn("OpenAIImages.decodeResponse")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.DecodeContext<Request>,
) {
const output = yield* MediaProtocol.decodeJson(ADAPTER, NAME, OpenAIImageResponse)(response)
const decoded = output.value
const format = decoded.output_format ?? requestedFormat(context.body) ?? "png"
const usage = (value: Schema.Schema.Type<typeof Usage> | undefined): MediaUsage | undefined =>
value === undefined
? undefined
: {
type: "tokens",
input: value.input_tokens,
output: value.output_tokens,
total: value.total_tokens,
details: { openai: value },
}
const eventImage = (frame: string, label: string, data: string, format: string) =>
MediaInput.decodedAsset(
(message, cause) => MediaProtocol.frameError(ADAPTER, message, frame, cause),
label,
data,
`image/${format}`,
{ info: { format } },
)
const onEvent = Effect.fn("OpenAIImages.onEvent")(function* (state: State, frame: string) {
const event = yield* decodeEvent(frame)
const format = event.output_format
if ("partial_image_index" in event) {
const image = yield* eventImage(frame, `${NAME} partial image`, event.b64_json, format)
return [state, [ImagePartialEvent.make({ index: event.partial_image_index, image })]] as const
}
const image = yield* eventImage(frame, `${NAME} result ${state.completed}`, event.b64_json, format)
return [
{ ...state, completed: state.completed + 1, format, usage: usage(event.usage) },
[ImageOutputEvent.make({ index: state.completed, image })],
] as const
})
const onDocument = Effect.fn("OpenAIImages.onDocument")(function* (frame: Exclude<Frame, string>) {
const invalid = (message: string, cause?: unknown) =>
MediaProtocol.frameError(ADAPTER, message, frame.document, cause)
const decoded = yield* decodeDocument(frame.document).pipe(
Effect.mapError((cause) => invalid(`${NAME} returned an invalid response`, cause)),
)
const format = decoded.output_format ?? frame.requested ?? "png"
const mediaType = `image/${format}`
const images = yield* Effect.forEach(decoded.data, (item, index) => {
const providerMetadata =
item.revised_prompt === undefined ? undefined : { openai: { revisedPrompt: item.revised_prompt } }
if (item.b64_json)
return MediaInput.decodedAsset(output.invalid, `${NAME} result ${index}`, item.b64_json, mediaType, {
return MediaInput.decodedAsset(invalid, `${NAME} result ${index}`, item.b64_json, mediaType, {
info: { format },
providerMetadata,
})
if (item.url) return Effect.succeed(Media.url(item.url, { mediaType, info: { format }, providerMetadata }))
return Effect.fail(output.invalid(`${NAME} result ${index} has neither image data nor a URL`))
})
if (images.length === 0) return yield* output.invalid(`${NAME} returned no images`)
return new ImageResponse({
images,
usage:
decoded.usage === undefined
? undefined
: {
type: "tokens",
input: decoded.usage.input_tokens,
output: decoded.usage.output_tokens,
total: decoded.usage.total_tokens,
details: { openai: decoded.usage },
},
providerMetadata: { openai: { outputFormat: format } },
return Effect.fail(invalid(`${NAME} result ${index} has neither image data nor a URL`))
})
if (images.length === 0) return yield* invalid(`${NAME} returned no images`)
const state: State = { completed: images.length, format, usage: usage(decoded.usage) }
return [state, images.map((image, index) => ImageOutputEvent.make({ index, image }))] as const
})
const step = (state: State, frame: Frame) => (typeof frame === "string" ? onEvent(state, frame) : onDocument(frame))
const finish = (state: State) => {
if (state.completed === 0) return Effect.fail(MediaProtocol.incomplete(ADAPTER))
return Effect.succeed([
ImageFinishEvent.make({ usage: state.usage, providerMetadata: { openai: { outputFormat: state.format } } }),
])
}
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.inline<Request, ImageResponse>({
export const protocol = MediaProtocol.stream<Request, ImageEvent, Frame, State>({
id: ADAPTER,
name: NAME,
unsupported: ["aspectRatio", "seed"],
body: { from: fromRequest },
response: { decode: decodeResponse },
frames: (bytes, context) =>
context.request.mode === "stream"
? Framing.sse.frame(bytes)
: Framing.document
.frame(bytes)
.pipe(Stream.map((document) => ({ document, requested: requestedFormat(context.body) }))),
initial: () => ({ completed: 0 }),
step,
finish,
})
export const model = (input: MediaRoute.ModelInput) =>
ImageModel.fromRoute<OpenAIImageOptions>(
ImageModel.fromRoute<OpenAIImageOptions, Frame, State>(
{
id: ADAPTER,
provider: PROVIDER,
@@ -0,0 +1,184 @@
import { Duration, Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import type { Status } from "../generation.js"
import { ImageModel, ImageResponse, type ImageRequestFor } from "../image.js"
import { Media } from "../media.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords, type AIError } from "../schema/index.js"
import { ProviderShared, optionalNull } from "./shared.js"
const ADAPTER = "replicate-images"
const NAME = "Replicate"
const PROVIDER = ProviderID.make("replicate")
export const DEFAULT_BASE_URL = "https://api.replicate.com"
const OUTPUT_RETENTION = Duration.hours(1)
const MAX_DATA_URL_BYTES = 256 * 1024
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
/** Model-defined `input` fields under the model's own names; `Media.Asset` values lower to URLs or data URLs. */
export type ReplicateImageOptions = Record<string, unknown>
export type Request = ImageRequestFor<ReplicateImageOptions>
// ---------------------------------------------------------------------------
// 2. Token and response schemas
// ---------------------------------------------------------------------------
export const Token = Schema.Struct({ id: Schema.String, getURL: Schema.String, cancelURL: Schema.String })
export type Token = Schema.Schema.Type<typeof Token>
const Prediction = Schema.Struct({
id: Schema.String,
status: Schema.String,
output: optionalNull(Schema.Unknown),
error: optionalNull(Schema.Unknown),
data_removed: optionalNull(Schema.Boolean),
completed_at: optionalNull(Schema.String),
metrics: optionalNull(Schema.Struct({ predict_time: optionalNull(Schema.Number) })),
urls: Schema.Struct({ get: Schema.String, cancel: Schema.String }),
})
const Output = Schema.Union([Schema.String, Schema.Array(Schema.String)])
const isOutput = Schema.is(Output)
const STATUS = {
starting: "queued",
processing: "running",
succeeded: "completed",
failed: "failed",
canceled: "cancelled",
// The prediction hit its `Cancel-After` deadline before it started running.
aborted: "expired",
} as const satisfies Record<string, Status>
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
// Official models run by `owner/name`; anything else (`owner/name:version` or a bare version id) is a pinned version.
const isOfficial = (model: string) => /^[^/:]+\/[^/:]+$/.test(model)
const inlineSize = (source: Media.Source) => {
if (source.type === "bytes") return source.data.byteLength
if (source.type === "base64") return source.data.length * 0.75
return 0
}
const fileInput = (asset: Media.Asset) => {
if (inlineSize(asset.source) > MAX_DATA_URL_BYTES)
return Effect.fail(
ProviderShared.invalidRequest(`${NAME} data URL inputs are limited to 256 KB; pass a larger file by https URL`),
)
return ProviderShared.mediaReference(asset, undefined, NAME).pipe(Effect.map((reference) => reference.value))
}
const inputValue = (value: unknown): Effect.Effect<unknown, AIError> => {
if (value instanceof Media.Asset) return fileInput(value)
if (Array.isArray(value) && value.some((item) => item instanceof Media.Asset))
return Effect.forEach(value, (item) => (item instanceof Media.Asset ? fileInput(item) : Effect.succeed(item)))
return Effect.succeed(value)
}
const fromRequest = Effect.fn("ReplicateImages.fromRequest")(function* (request: Request) {
const native = yield* Effect.forEach(Object.entries(request.providerOptions ?? {}), ([key, value]) =>
inputValue(value).pipe(Effect.map((lowered) => [key, lowered] as const)),
)
const input = mergeJsonRecords({ prompt: request.prompt }, Object.fromEntries(native))
const model = request.model.id
return MediaProtocol.json(
mergeJsonRecords(isOfficial(model) ? { input } : { version: model, input }, request.http?.body) ?? {},
)
})
// ---------------------------------------------------------------------------
// 6. Response decoding
// ---------------------------------------------------------------------------
const decodePrediction = MediaProtocol.decodeJson(ADAPTER, NAME, Prediction)
const decodeStart = Effect.fn("ReplicateImages.decodeStart")(function* (
response: HttpClientResponse.HttpClientResponse,
) {
const output = yield* decodePrediction(response)
const prediction = output.value
return {
token: { id: prediction.id, getURL: prediction.urls.get, cancelURL: prediction.urls.cancel },
// `Prefer: wait` can return an already-finished prediction, so `await` skips straight to the result.
snapshot: { id: prediction.id, status: yield* MediaProtocol.status(STATUS, prediction.status, output) },
}
})
const decodeStatus = Effect.fn("ReplicateImages.decodeStatus")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) {
const output = yield* decodePrediction(response)
return { id: context.token.id, status: yield* MediaProtocol.status(STATUS, output.value.status, output) }
})
const decodeResult = Effect.fn("ReplicateImages.decodeResult")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) {
const output = yield* decodePrediction(response)
const prediction = output.value
const status = yield* MediaProtocol.status(STATUS, prediction.status, output)
if (status === "failed" || status === "cancelled" || status === "expired")
return yield* output.ended(
status,
`${NAME} prediction ${context.token.id} ${prediction.status}${typeof prediction.error === "string" ? `: ${prediction.error}` : ""}`,
)
if (status !== "completed") return yield* output.invalid(`${NAME} prediction ${context.token.id} has not finished`)
if (prediction.data_removed === true)
return yield* output.ended("expired", `${NAME} removed the output of prediction ${context.token.id}`)
if (!isOutput(prediction.output))
return yield* output.invalid(`${NAME} prediction ${context.token.id} returned output that is not image URLs`)
const urls = typeof prediction.output === "string" ? [prediction.output] : prediction.output
if (urls.length === 0) return yield* output.invalid(`${NAME} prediction ${context.token.id} returned no images`)
const predictTime = prediction.metrics?.predict_time ?? undefined
const completedAt = prediction.completed_at ?? undefined
const expiresAt =
completedAt === undefined ? undefined : Date.parse(completedAt) + Duration.toMillis(OUTPUT_RETENTION)
return new ImageResponse({
images: urls.map((url) => Media.url(url, { expiresAt })),
usage: predictTime === undefined ? undefined : { type: "compute", seconds: predictTime },
providerMetadata: { replicate: { predictionId: prediction.id } },
})
})
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.queued<Request, ImageResponse, Token>({
id: ADAPTER,
name: NAME,
token: Token,
unsupported: ["images", "mask", "n", "size", "aspectRatio", "seed", "format"],
start: { body: { from: fromRequest }, decode: decodeStart },
status: { path: (token) => token.getURL, decode: decodeStatus },
result: { path: (token) => token.getURL, decode: decodeResult },
cancel: { method: "POST", path: (token) => token.cancelURL },
})
export const model = (input: MediaRoute.ModelInput) =>
ImageModel.fromRoute<ReplicateImageOptions, Token>(
{
id: ADAPTER,
provider: PROVIDER,
protocol,
baseURL: DEFAULT_BASE_URL,
path: ({ request }) =>
isOfficial(request.model.id) ? `/v1/models/${request.model.id}/predictions` : "/v1/predictions",
},
input,
)
export const ReplicateImages = {
protocol,
model,
} as const
@@ -0,0 +1,244 @@
import { Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import { ImageModel, ImageResponse, type ImageRequestFor } from "../image.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { ProviderShared } from "./shared.js"
import { MediaInput } from "./utils/media-input.js"
const ADAPTER = "stability-images"
const UPSCALE_ADAPTER = "stability-upscale"
const NAME = "Stability AI"
const PROVIDER = ProviderID.make("stability")
export const DEFAULT_BASE_URL = "https://api.stability.ai"
const RESULTS_PATH = "/v2beta/results"
const UPSCALE_MODEL = "creative"
/** Base64 JSON instead of raw bytes, so the seed and finish reason arrive with the image. */
const HEADERS = { accept: "application/json" }
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type StabilityStylePreset =
| "enhance"
| "anime"
| "photographic"
| "digital-art"
| "comic-book"
| "fantasy-art"
| "line-art"
| "analog-film"
| "neon-punk"
| "isometric"
| "low-poly"
| "origami"
| "modeling-compound"
| "cinematic"
| "3d-model"
| "pixel-art"
| "tile-texture"
| (string & {})
export type StabilityImageOptions = {
readonly negative_prompt?: string
readonly style_preset?: StabilityStylePreset
readonly strength?: number
readonly cfg_scale?: number
} & Record<string, unknown>
/** Creative upscale options; the one `images` source is the image to upscale. */
export type StabilityUpscaleOptions = {
readonly negative_prompt?: string
readonly style_preset?: StabilityStylePreset
readonly creativity?: number
} & Record<string, unknown>
export type Request = ImageRequestFor<StabilityImageOptions>
export type UpscaleRequest = ImageRequestFor<StabilityUpscaleOptions>
// ---------------------------------------------------------------------------
// 2. Token and response schemas
// ---------------------------------------------------------------------------
export const Token = Schema.Struct({ id: Schema.String })
export type Token = Schema.Schema.Type<typeof Token>
// Generate names the base64 image `image`; `/v2beta/results/{id}` names it `result`.
const ImageDocument = Schema.Struct({
image: Schema.optional(Schema.String),
result: Schema.optional(Schema.String),
seed: Schema.optional(Schema.Number),
finish_reason: Schema.optional(Schema.String),
})
const Started = Schema.Struct({ id: Schema.String })
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
/** `sd3.5-*` ids share the `sd3` endpoint and select the model with a form field. */
const endpoint = (model: string) => (model.startsWith("sd3") ? "sd3" : model)
const RESERVED_FORM_FIELDS = new Set(["image", "prompt", "mode", "model"])
const unsupported = (route: string, operation: string, message: string) =>
ProviderShared.unsupportedOperation({ operation, provider: PROVIDER, route, message })
const form = Effect.fn("StabilityImages.form")(function* (
route: string,
fields: Record<string, unknown>,
native: Record<string, unknown> | undefined,
source: Request["images"],
) {
if ((source?.length ?? 0) > 1) return yield* unsupported(route, "media.images", `${NAME} takes one source image`)
const body = new FormData()
const overlay = Object.entries(native ?? {}).filter(([key]) => !RESERVED_FORM_FIELDS.has(key))
Object.entries(mergeJsonRecords(fields, Object.fromEntries(overlay)) ?? {}).forEach(([key, value]) =>
body.append(key, typeof value === "string" ? value : ProviderShared.encodeJson(value)),
)
const image = source?.[0]
if (image !== undefined)
body.append("image", MediaInput.blob(yield* MediaInput.inlineBytes(route, image), image.mediaType), "image")
return MediaProtocol.multipart(body)
})
const fromRequest = Effect.fn("StabilityImages.fromRequest")(function* (request: Request) {
if (request.n !== undefined && request.n > 1)
return yield* unsupported(ADAPTER, "media.n", `${NAME} generates one image per request; call it once per image`)
const target = endpoint(request.model.id)
const edit = (request.images?.length ?? 0) > 0
if (edit && target === "core")
return yield* unsupported(ADAPTER, "media.images", `${NAME} core is text-to-image only; use ultra or sd3.5-*`)
return yield* form(
ADAPTER,
{
prompt: request.prompt,
aspect_ratio: request.aspectRatio,
seed: request.seed,
output_format: request.format,
model: target === "sd3" ? request.model.id : undefined,
mode: target === "sd3" && edit ? "image-to-image" : undefined,
},
mergeJsonRecords(request.providerOptions, request.http?.body),
request.images,
)
})
const fromUpscaleRequest = Effect.fn("StabilityImages.fromUpscaleRequest")(function* (request: UpscaleRequest) {
if ((request.images?.length ?? 0) === 0)
return yield* ProviderShared.invalidRequest(`${NAME} upscale requires the source image in images`)
return yield* form(
UPSCALE_ADAPTER,
{ prompt: request.prompt, seed: request.seed, output_format: request.format },
mergeJsonRecords(request.providerOptions, request.http?.body),
request.images,
)
})
// ---------------------------------------------------------------------------
// 6. Response decoding
// ---------------------------------------------------------------------------
const decodeImageDocument = (route: string) => {
const decode = MediaProtocol.decodeJson(route, NAME, ImageDocument)
return Effect.fn("StabilityImages.decodeImage")(function* (response: HttpClientResponse.HttpClientResponse) {
const output = yield* decode(response)
const document = output.value
const data = document.image ?? document.result
if (data === undefined) return yield* output.invalid(`${NAME} returned no image`)
const image = yield* MediaInput.decodedAsset(output.invalid, `${NAME} result`, data, undefined)
return new ImageResponse({
images: [image],
notices:
document.finish_reason === "CONTENT_FILTERED"
? [{ type: "moderated", message: `${NAME} blurred the image for violating its content policy` }]
: undefined,
providerMetadata: { stability: { seed: document.seed, finishReason: document.finish_reason } },
})
})
}
const decodeResponse = decodeImageDocument(ADAPTER)
const decodeUpscaleImage = decodeImageDocument(UPSCALE_ADAPTER)
const decodeStart = MediaProtocol.decodeStarted(UPSCALE_ADAPTER, NAME, Started, (value) => ({
token: { id: value.id },
snapshot: { id: value.id, status: "queued" },
}))
// `/v2beta/results/{id}` answers 202 while in progress and 200 with the finished image document.
const decodeStatus = (response: HttpClientResponse.HttpClientResponse, context: MediaProtocol.PollContext<Token>) =>
Effect.succeed({
id: context.token.id,
status: response.status === 202 ? ("running" as const) : ("completed" as const),
})
const decodeUpscaleResult = Effect.fn("StabilityImages.decodeUpscaleResult")(function* (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) {
if (response.status === 202) {
const output = yield* MediaProtocol.text(UPSCALE_ADAPTER, NAME, response)
return yield* output.invalid(`${NAME} upscale ${context.token.id} has not finished`)
}
return yield* decodeUpscaleImage(response)
})
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.inline<Request, ImageResponse>({
id: ADAPTER,
name: NAME,
unsupported: ["size", "mask"],
body: { from: fromRequest },
response: { decode: decodeResponse },
})
export const upscaleProtocol = MediaProtocol.queued<UpscaleRequest, ImageResponse, Token>({
id: UPSCALE_ADAPTER,
name: NAME,
token: Token,
unsupported: ["n", "size", "aspectRatio", "mask"],
start: { body: { from: fromUpscaleRequest }, decode: decodeStart },
status: { path: (token) => `${RESULTS_PATH}/${token.id}`, decode: decodeStatus },
result: { path: (token) => `${RESULTS_PATH}/${token.id}`, decode: decodeUpscaleResult },
})
export const model = (input: MediaRoute.ModelInput) =>
ImageModel.fromRoute<StabilityImageOptions>(
{
id: ADAPTER,
provider: PROVIDER,
protocol,
baseURL: DEFAULT_BASE_URL,
headers: HEADERS,
path: ({ request }) => `/v2beta/stable-image/generate/${endpoint(request.model.id)}`,
},
input,
)
export const upscaleModel = (input: Omit<MediaRoute.ModelInput, "id">) =>
ImageModel.fromRoute<StabilityUpscaleOptions, Token>(
{
id: UPSCALE_ADAPTER,
provider: PROVIDER,
protocol: upscaleProtocol,
baseURL: DEFAULT_BASE_URL,
headers: HEADERS,
// Only the creative upscaler is asynchronous; fast and conservative answer inline.
path: `/v2beta/stable-image/upscale/${UPSCALE_MODEL}`,
},
{ ...input, id: UPSCALE_MODEL },
)
export const StabilityImages = {
protocol,
upscaleProtocol,
model,
upscaleModel,
} as const
@@ -0,0 +1,92 @@
import { Effect, Schema } from "effect"
import type { HttpClientResponse } from "effect/unstable/http"
import type { Status } from "../../generation.js"
import type { Media } from "../../media.js"
import { MediaProtocol } from "../../route/media-protocol.js"
import type { AIError } from "../../schema/index.js"
import { ProviderShared, optionalNull } from "../shared.js"
export const DEFAULT_BASE_URL = "https://queue.fal.run"
/** fal hands back absolute follow-up URLs on submit; they are authoritative for status, result, and cancel. */
export const Token = Schema.Struct({
requestID: Schema.String,
statusURL: Schema.String,
responseURL: Schema.String,
cancelURL: Schema.String,
})
export type Token = Schema.Schema.Type<typeof Token>
const StartResponse = Schema.Struct({
request_id: Schema.String,
status_url: Schema.String,
response_url: Schema.String,
cancel_url: Schema.String,
queue_position: optionalNull(Schema.Number),
})
const QueueStatus = Schema.Struct({
status: Schema.String,
queue_position: optionalNull(Schema.Number),
error: optionalNull(Schema.Unknown),
})
const STATUS = {
IN_QUEUE: "queued",
IN_PROGRESS: "running",
COMPLETED: "completed",
} as const satisfies Record<string, Status>
// fal accepts public URLs and data URIs; there is no provider file handle to forward.
export const mediaUrl = (asset: Media.Asset, name: string) =>
ProviderShared.mediaReference(asset, undefined, name).pipe(Effect.map((reference) => reference.value))
export const protocol = <Request, Response>(input: {
readonly id: string
readonly name: string
readonly unsupported?: ReadonlyArray<keyof Request & string>
readonly from: (request: Request) => Effect.Effect<MediaProtocol.Body, AIError>
readonly decodeResult: (
response: HttpClientResponse.HttpClientResponse,
context: MediaProtocol.PollContext<Token>,
) => Effect.Effect<Response, AIError>
}) => {
const decodeQueueStatus = MediaProtocol.decodeJson(input.id, input.name, QueueStatus)
return MediaProtocol.queued<Request, Response, Token>({
id: input.id,
name: input.name,
token: Token,
unsupported: input.unsupported,
start: {
body: { from: input.from },
decode: MediaProtocol.decodeStarted(input.id, input.name, StartResponse, (value) => ({
token: {
requestID: value.request_id,
statusURL: value.status_url,
responseURL: value.response_url,
cancelURL: value.cancel_url,
},
snapshot: { id: value.request_id, status: "queued", position: value.queue_position ?? undefined },
})),
},
status: {
path: (token) => token.statusURL,
decode: Effect.fn("FalQueue.decodeStatus")(function* (response, context) {
const output = yield* decodeQueueStatus(response)
const decoded = output.value
const status = yield* MediaProtocol.status(STATUS, decoded.status, output)
// fal reports request failures as COMPLETED with an `error`; the response endpoint carries the details.
const failed = status === "completed" && decoded.error !== undefined && decoded.error !== null
return {
id: context.token.requestID,
status: failed ? "failed" : status,
position: status === "queued" ? (decoded.queue_position ?? undefined) : undefined,
}
}),
},
result: { path: (token) => token.responseURL, decode: input.decodeResult },
cancel: { method: "PUT", path: (token) => token.cancelURL },
})
}
export * as FalQueue from "./fal-queue.js"
@@ -34,6 +34,11 @@ export const query = (route: string, values: Record<string, unknown>): Effect.Ef
)
}
export const dimensions = (size: string) => {
const [width, height] = size.split("x").map(Number)
return { width, height }
}
/** Provider file handle when the ref belongs to this provider; refs from other providers are never forwarded. */
export const refID = (asset: Media.Asset, provider: ProviderID) =>
asset.source.type === "ref" && asset.source.provider === provider ? asset.source.id : undefined
+3
View File
@@ -86,6 +86,7 @@ const INVALID_REQUEST_CODES = new Set(["invalid_prompt", "invalid_request_error"
// emits `image_content_policy_violation` as the native code.
const CONTENT_POLICY_CODES = new Set([
"content_filter",
"content_moderation",
"responsibleaipolicyviolation",
"content_policy_violation",
"image_content_policy_violation",
@@ -204,6 +205,8 @@ function providerCodes(value: unknown) {
const exception = isRecord(decoded.exception) ? decoded.exception : undefined
return [
decoded.code,
// Stability's `{ id, name, errors }` bodies carry the code in `name`.
Array.isArray(decoded.errors) ? decoded.name : undefined,
decoded.error_type,
error?.code,
error?.type,
@@ -0,0 +1,42 @@
import { Auth } from "../route/auth.js"
import type { ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { BlackForestLabsImages, DEFAULT_BASE_URL } from "../protocols/bfl-images.js"
export type { BlackForestLabsImageOptions } from "../protocols/bfl-images.js"
export const id = ProviderID.make("black-forest-labs")
const baseURL = DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
/** `https://api.eu.bfl.ai` or `https://api.us.bfl.ai` pin inference to one region. */
readonly baseURL?: string
readonly headers?: Record<string, string>
readonly http?: HttpOptions.Input
}
const auth = (options: ProviderAuthOption<"optional">) => {
if ("auth" in options && options.auth) return options.auth
return Auth.optional("apiKey" in options ? options.apiKey : undefined, "apiKey")
.orElse(Auth.config("BFL_API_KEY"))
.pipe(Auth.header("x-key"))
}
export const configure = (input: Config = {}) => {
const image = (modelID: string | ModelID) =>
BlackForestLabsImages.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
return {
id,
image,
configure,
}
}
export const provider = configure()
export const image = provider.image
+15 -11
View File
@@ -1,12 +1,15 @@
import { Auth } from "../route/auth.js"
import type { ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { DEFAULT_BASE_URL, FalVideo } from "../protocols/fal-video.js"
import { FalImages } from "../protocols/fal-images.js"
import { FalVideo } from "../protocols/fal-video.js"
import { FalQueue } from "../protocols/utils/fal-queue.js"
export type { FalImageOptions } from "../protocols/fal-images.js"
export type { FalVideoOptions } from "../protocols/fal-video.js"
export const id = ProviderID.make("fal")
const baseURL = DEFAULT_BASE_URL
const baseURL = FalQueue.DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
readonly baseURL?: string
@@ -23,20 +26,21 @@ const auth = (options: ProviderAuthOption<"optional">) => {
}
export const configure = (input: Config = {}) => {
const video = (modelID: string | ModelID) =>
FalVideo.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
const media = (modelID: string | ModelID) => ({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
return {
id,
video,
image: (modelID: string | ModelID) => FalImages.model(media(modelID)),
video: (modelID: string | ModelID) => FalVideo.model(media(modelID)),
configure,
}
}
export const provider = configure()
export const image = provider.image
export const video = provider.video
+3
View File
@@ -6,6 +6,7 @@ export * as AmazonBedrockMantle from "./amazon-bedrock-mantle.js"
export * as AssemblyAI from "./assemblyai.js"
export * as Azure from "./azure.js"
export * as Baseten from "./baseten.js"
export * as BlackForestLabs from "./black-forest-labs.js"
export * as Cartesia from "./cartesia.js"
export * as Cerebras from "./cerebras.js"
export * as CloudflareAIGateway from "./cloudflare-ai-gateway.js"
@@ -31,7 +32,9 @@ export * as OpenAICompatible from "./openai-compatible.js"
export * as OpenAICompatibleResponses from "./openai-compatible-responses.js"
export * as OpenCodeZen from "./opencode-zen.js"
export * as OpenRouter from "./openrouter.js"
export * as Replicate from "./replicate.js"
export * as Runway from "./runway.js"
export * as Stability from "./stability.js"
export * as TogetherAI from "./togetherai.js"
export * as TypeSafeAI from "./typesafe-ai.js"
export * as VercelAIGateway from "./vercel-ai-gateway.js"
+36
View File
@@ -0,0 +1,36 @@
import { AuthOptions, type ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { DEFAULT_BASE_URL, ReplicateImages } from "../protocols/replicate-images.js"
export type { ReplicateImageOptions } from "../protocols/replicate-images.js"
export const id = ProviderID.make("replicate")
const baseURL = DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
readonly baseURL?: string
/** `{ Prefer: "wait=60" }` holds the submission open until the prediction finishes (up to 60 seconds). */
readonly headers?: Record<string, string>
readonly http?: HttpOptions.Input
}
const auth = (options: ProviderAuthOption<"optional">) => AuthOptions.bearer(options, "REPLICATE_API_TOKEN")
export const configure = (input: Config = {}) => {
const image = (modelID: string | ModelID) =>
ReplicateImages.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
return {
id,
image,
configure,
}
}
export const provider = configure()
export const image = provider.image
+35
View File
@@ -0,0 +1,35 @@
import { AuthOptions, type ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { DEFAULT_BASE_URL, StabilityImages } from "../protocols/stability-images.js"
export type { StabilityImageOptions, StabilityUpscaleOptions } from "../protocols/stability-images.js"
export const id = ProviderID.make("stability")
const baseURL = DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
readonly baseURL?: string
readonly headers?: Record<string, string>
readonly http?: HttpOptions.Input
}
const auth = (options: ProviderAuthOption<"optional">) => AuthOptions.bearer(options, "STABILITY_API_KEY")
export const configure = (input: Config = {}) => {
const deployment = {
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
}
return {
id,
image: (modelID: string | ModelID) => StabilityImages.model({ ...deployment, id: modelID }),
upscale: () => StabilityImages.upscaleModel(deployment),
configure,
}
}
export const provider = configure()
export const image = provider.image
export const upscale = provider.upscale
+3 -1
View File
@@ -88,11 +88,13 @@ export interface Started<Token> {
/**
* A follow-up call's inputs: the decoded token and the auth headers the route sent, so a protocol can attach them
* to output URLs that require the same credentials to download (Veo).
* to output URLs that require the same credentials to download (Veo). `materialize` downloads an output through the
* route's executor, for URLs that expire too soon to hand back (BFL).
*/
export interface PollContext<Token> {
readonly token: Token
readonly auth: Record<string, string>
readonly materialize: (asset: Media.Asset) => Effect.Effect<Media.Asset, AIError>
}
/**
+68 -3
View File
@@ -2,10 +2,17 @@ import { Effect, Schema, Stream } from "effect"
import { Headers, HttpClientRequest, type HttpClientResponse } from "effect/unstable/http"
import { Auth, type AuthInput } from "./auth.js"
import { Endpoint } from "./endpoint.js"
import type { Interface } from "./executor-service.js"
import { Service as RequestExecutorService, type Interface } from "./executor-service.js"
import { RequestExecutor } from "./executor.js"
import { MediaProtocol } from "./media-protocol.js"
import { Generation, type Route as GenerationRoute } from "../generation.js"
import {
Generation,
resultEvents,
type AwaitOptions,
type Observation,
type Route as GenerationRoute,
} from "../generation.js"
import type { Media } from "../media.js"
import { ProviderShared } from "../protocols/shared.js"
import {
AIError,
@@ -72,6 +79,11 @@ export interface StreamRoute<Request extends MediaRequest, Event, Response> {
readonly generate: (request: Request, execute: Execute) => Effect.Effect<Response, AIError>
}
export type AnyRoute<Request extends MediaRequest, Event, Response> =
| Route<Request, Response>
| StreamRoute<Request, Event, Response>
| QueuedRoute<Request, Response>
export interface Composition<Request extends MediaRequest> {
readonly id: string
readonly provider: string | ProviderID
@@ -136,6 +148,8 @@ export const queued = <Request extends MediaRequest, Response, Token>(
const encodeToken = Schema.encodeSync(protocol.token)
const generationRoute = (token: Token, http: HttpOptions | undefined, execute: Execute) => {
const materialize = (asset: Media.Asset) =>
asset.materialize().pipe(Effect.provideService(RequestExecutorService, { execute }))
const poll = <A>(operation: {
readonly path: (token: Token) => string
readonly decode: (
@@ -145,7 +159,7 @@ export const queued = <Request extends MediaRequest, Response, Token>(
}) =>
transport
.call("GET", operation.path(token), http, execute)
.pipe(Effect.flatMap((sent) => operation.decode(sent.response, { token, auth: sent.auth })))
.pipe(Effect.flatMap((sent) => operation.decode(sent.response, { token, auth: sent.auth, materialize })))
const cancel = protocol.cancel
const route: GenerationRoute<Response> = {
status: poll(protocol.status),
@@ -250,6 +264,57 @@ export const stream = <Request extends MediaRequest, Event, Response, Frame, Sta
}
}
export const dispatch = <Event, Response>(input: {
readonly modality: string
readonly execute: Execute
readonly responseEvents: (response: Response) => ReadonlyArray<Event>
}) => {
const notQueued = (route: { readonly provider: ProviderID; readonly id: string }, operation: string) =>
ProviderShared.unsupportedOperation({
operation: `${input.modality}.${operation}`,
provider: route.provider,
route: route.id,
message: `${route.provider}/${route.id} is not a queued route; use generate or stream`,
})
const start = <Request extends MediaRequest>(route: AnyRoute<Request, Event, Response>, request: Request) => {
if (route.kind !== "queued") return Effect.fail(notQueued(route, "start"))
return route.start(request, input.execute)
}
return {
start,
resume: <Request extends MediaRequest>(
route: AnyRoute<Request, Event, Response>,
model: MediaRequest["model"],
token: unknown,
) => {
if (route.kind !== "queued") return Effect.fail(notQueued(route, "resume"))
return route.resume(model, token, input.execute)
},
generate: <Request extends MediaRequest>(
route: AnyRoute<Request, Event, Response>,
request: Request,
options?: AwaitOptions,
) => {
if (route.kind !== "queued") return route.generate(request, input.execute)
return start(route, request).pipe(Effect.flatMap((generation) => generation.await(options)))
},
stream: <Request extends MediaRequest>(
route: AnyRoute<Request, Event, Response>,
request: Request,
options?: AwaitOptions,
): Stream.Stream<Event | Observation, AIError> => {
if (route.kind === "stream") return route.stream(request, input.execute)
if (route.kind === "queued")
return Stream.unwrap(
start(route, request).pipe(
Effect.map((generation) => resultEvents(generation, input.responseEvents, options)),
),
)
return Stream.fromIterableEffect(Effect.map(route.generate(request, input.execute), input.responseEvents))
},
}
}
// ---------------------------------------------------------------------------
// Transport plumbing shared by every kind
// ---------------------------------------------------------------------------
+11 -36
View File
@@ -1,7 +1,7 @@
import { Context, Effect, Layer, Stream } from "effect"
import { resultEvents, type AwaitOptions, type Generation } from "./generation.js"
import { ProviderShared } from "./protocols/shared.js"
import type { AwaitOptions, Generation } from "./generation.js"
import { RequestExecutor } from "./route/executor.js"
import { MediaRoute } from "./route/media.js"
import type { AIError } from "./schema/index.js"
import {
responseEvents,
@@ -10,7 +10,6 @@ import {
type TranscriptionOptions,
type TranscriptionRequestFor,
type TranscriptionResponse,
type TranscriptionRoute,
} from "./transcription.js"
export interface Interface {
@@ -70,44 +69,20 @@ export const resume = <Options extends TranscriptionOptions>(
return yield* client.resume(model, token)
})
const notQueued = <Options extends TranscriptionOptions>(route: TranscriptionRoute<Options>, operation: string) =>
ProviderShared.unsupportedOperation({
operation: `transcription.${operation}`,
provider: route.provider,
route: route.id,
message: `${route.provider}/${route.id} is not a queued route; use Transcription.generate or Transcription.stream`,
})
export const layer: Layer.Layer<Service, never, RequestExecutor.Service> = Layer.effect(
Service,
Effect.gen(function* () {
const executor = yield* RequestExecutor.Service
const start = <Options extends TranscriptionOptions>(request: TranscriptionRequestFor<Options>) => {
const route = request.model.route
if (route.kind !== "queued") return Effect.fail(notQueued(route, "start"))
return route.start(request, executor.execute)
}
const dispatch = MediaRoute.dispatch<TranscriptionEvent, TranscriptionResponse>({
modality: "transcription",
execute: executor.execute,
responseEvents,
})
return Service.of({
start,
resume: (model, token) => {
const route = model.route
if (route.kind !== "queued") return Effect.fail(notQueued(route, "resume"))
return route.resume(model, token, executor.execute)
},
generate: (request, options) => {
const route = request.model.route
if (route.kind !== "queued") return route.generate(request, executor.execute)
return start(request).pipe(Effect.flatMap((generation) => generation.await(options)))
},
stream: (request, options) => {
const route = request.model.route
if (route.kind === "stream") return route.stream(request, executor.execute)
if (route.kind === "queued")
return Stream.unwrap(
start(request).pipe(Effect.map((generation) => resultEvents(generation, responseEvents, options))),
)
return Stream.fromIterableEffect(Effect.map(route.generate(request, executor.execute), responseEvents))
},
start: (request) => dispatch.start(request.model.route, request),
resume: (model, token) => dispatch.resume(model.route, model, token),
generate: (request, options) => dispatch.generate(request.model.route, request, options),
stream: (request, options) => dispatch.stream(request.model.route, request, options),
})
}),
)
+15 -20
View File
@@ -1,7 +1,7 @@
import { Effect, Schema, Stream } from "effect"
import { Generation, ProgressEvent, QueuedEvent, type AwaitOptions } from "./generation.js"
import { Media } from "./media.js"
import { MediaModel, composeRoute, tryRequest } from "./media-model.js"
import { MediaModel, composeAnyRoute, tryRequest } from "./media-model.js"
import { MediaRoute } from "./route/media.js"
import type { MediaProtocol } from "./route/media-protocol.js"
import { AIError, HttpOptions, MediaUsage, ProviderMetadata } from "./schema/index.js"
@@ -13,10 +13,11 @@ import { TranscriptionClient, Service } from "./transcription-client.js"
export type TranscriptionOptions = Record<string, unknown>
export type TranscriptionRoute<Options extends TranscriptionOptions = TranscriptionOptions> =
| MediaRoute.Route<TranscriptionRequestFor<Options>, TranscriptionResponse>
| MediaRoute.StreamRoute<TranscriptionRequestFor<Options>, TranscriptionEvent, TranscriptionResponse>
| MediaRoute.QueuedRoute<TranscriptionRequestFor<Options>, TranscriptionResponse>
export type TranscriptionRoute<Options extends TranscriptionOptions = TranscriptionOptions> = MediaRoute.AnyRoute<
TranscriptionRequestFor<Options>,
TranscriptionEvent,
TranscriptionResponse
>
export class TranscriptionModel<Options extends TranscriptionOptions = TranscriptionOptions> extends MediaModel<
TranscriptionRoute<Options>,
@@ -47,16 +48,11 @@ export class TranscriptionModel<Options extends TranscriptionOptions = Transcrip
route: TranscriptionModel.RouteInput<Options, Frame, State, Token>,
input: MediaRoute.ModelInput,
) {
const composed: TranscriptionRoute<Options> = isStreamInput(route)
? composeRoute((composition) => MediaRoute.stream({ ...composition, collect: collectResponse }), route, input)
: isQueuedInput(route)
? composeRoute(MediaRoute.queued, route, input)
: composeRoute(MediaRoute.inline, route, input)
return new TranscriptionModel<Options>({
id: input.id,
provider: route.provider,
http: input.http,
route: composed,
route: composeAnyRoute(route, input, collectResponse),
})
}
}
@@ -89,17 +85,16 @@ export namespace TranscriptionModel {
Frame = unknown,
State = unknown,
Token = unknown,
> = InlineRouteInput<Options> | StreamRouteInput<Options, Frame, State> | QueuedRouteInput<Options, Token>
> = MediaModel.AnyRouteInput<
TranscriptionRequestFor<Options>,
TranscriptionEvent,
TranscriptionResponse,
Frame,
State,
Token
>
}
const isStreamInput = <Options extends TranscriptionOptions, Frame, State, Token>(
route: TranscriptionModel.RouteInput<Options, Frame, State, Token>,
): route is TranscriptionModel.StreamRouteInput<Options, Frame, State> => route.protocol.kind === "stream"
const isQueuedInput = <Options extends TranscriptionOptions, Frame, State, Token>(
route: TranscriptionModel.RouteInput<Options, Frame, State, Token>,
): route is TranscriptionModel.QueuedRouteInput<Options, Token> => route.protocol.kind === "queued"
export const TranscriptionModelSchema = Schema.declare(
(value): value is TranscriptionModel => value instanceof TranscriptionModel,
{ expected: "Transcription.Model" },
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -0,0 +1,65 @@
{
"version": 1,
"metadata": {
"tags": [
"prefix:replicate-images",
"provider:replicate",
"protocol:replicate-images"
],
"name": "replicate-images/waits-for-a-prediction-and-reads-its-output",
"recordedAt": "2026-09-23T18:42:06.871Z"
},
"interactions": [
{
"transport": "http",
"request": {
"method": "POST",
"url": "https://api.replicate.com/v1/models/black-forest-labs/flux-schnell/predictions",
"headers": {
"content-type": "application/json"
},
"body": "{\"input\":{\"prompt\":\"A simple flat black circle centered on a plain white background.\",\"aspect_ratio\":\"1:1\",\"output_format\":\"webp\",\"megapixels\":\"0.25\"}}"
},
"response": {
"status": 201,
"headers": {
"content-type": "application/json; charset=utf-8"
},
"body": "{\"id\":\"mde2assdssrp40d0sy9tf0p6a8\",\"model\":\"black-forest-labs/flux-schnell\",\"version\":\"hidden\",\"input\":{\"aspect_ratio\":\"1:1\",\"megapixels\":\"0.25\",\"output_format\":\"webp\",\"prompt\":\"A simple flat black circle centered on a plain white background.\"},\"logs\":\"running quantized prediction\\nUsing seed: 6062557\\n 0%| | 0/4 [00:00<?, ?it/s]\\n100%|██████████| 4/4 [00:00<00:00, 57.07it/s]\\nTotal safe images: 1 out of 1\\n\",\"output\":[\"https://replicate.delivery/xezq/r1LEsqIqPxYAH9qg7cV5mGQBxCuO98VnnE3py6f7AyTe9cQXA/out-0.webp\"],\"data_removed\":false,\"error\":null,\"source\":\"api\",\"status\":\"succeeded\",\"created_at\":\"2026-09-23T18:42:04.366Z\",\"started_at\":\"2026-09-23T18:42:04.374610941Z\",\"completed_at\":\"2026-09-23T18:42:04.540732474Z\",\"urls\":{\"cancel\":\"https://api.replicate.com/v1/predictions/mde2assdssrp40d0sy9tf0p6a8/cancel\",\"get\":\"https://api.replicate.com/v1/predictions/mde2assdssrp40d0sy9tf0p6a8\",\"stream\":\"https://stream.replicate.com/v1/files/uwhq-pppjocryviy6634d5p2ypvoge6a4sdagsyjscmwfycwk4s5mzfya\",\"web\":\"https://replicate.com/p/mde2assdssrp40d0sy9tf0p6a8\"},\"metrics\":{\"image_count\":1,\"predict_time\":0.166121534,\"total_time\":0.174732474}}"
}
},
{
"transport": "http",
"request": {
"method": "GET",
"url": "https://api.replicate.com/v1/predictions/mde2assdssrp40d0sy9tf0p6a8",
"headers": {},
"body": ""
},
"response": {
"status": 200,
"headers": {
"content-type": "application/json; charset=utf-8"
},
"body": "{\"id\":\"mde2assdssrp40d0sy9tf0p6a8\",\"model\":\"black-forest-labs/flux-schnell\",\"version\":\"hidden\",\"input\":{\"aspect_ratio\":\"1:1\",\"megapixels\":\"0.25\",\"output_format\":\"webp\",\"prompt\":\"A simple flat black circle centered on a plain white background.\"},\"logs\":\"running quantized prediction\\nUsing seed: 6062557\\n 0%| | 0/4 [00:00\\u003c?, ?it/s]\\n100%|██████████| 4/4 [00:00\\u003c00:00, 57.07it/s]\\nTotal safe images: 1 out of 1\\n\",\"output\":[\"https://replicate.delivery/xezq/r1LEsqIqPxYAH9qg7cV5mGQBxCuO98VnnE3py6f7AyTe9cQXA/out-0.webp\"],\"data_removed\":false,\"error\":null,\"source\":\"api\",\"status\":\"succeeded\",\"created_at\":\"2026-09-23T18:42:04.366Z\",\"started_at\":\"2026-09-23T18:42:04.374610941Z\",\"completed_at\":\"2026-09-23T18:42:04.540732474Z\",\"urls\":{\"cancel\":\"https://api.replicate.com/v1/predictions/mde2assdssrp40d0sy9tf0p6a8/cancel\",\"get\":\"https://api.replicate.com/v1/predictions/mde2assdssrp40d0sy9tf0p6a8\",\"stream\":\"https://stream.replicate.com/v1/files/uwhq-pppjocryviy6634d5p2ypvoge6a4sdagsyjscmwfycwk4s5mzfya\",\"web\":\"https://replicate.com/p/mde2assdssrp40d0sy9tf0p6a8\"},\"metrics\":{\"image_count\":1,\"predict_time\":0.166121534,\"total_time\":0.174732474}}\n"
}
},
{
"transport": "http",
"request": {
"method": "GET",
"url": "https://replicate.delivery/xezq/r1LEsqIqPxYAH9qg7cV5mGQBxCuO98VnnE3py6f7AyTe9cQXA/out-0.webp",
"headers": {},
"body": ""
},
"response": {
"status": 200,
"headers": {
"content-type": "image/webp"
},
"body": "UklGRqoHAABXRUJQVlA4IJ4HAAAwVwCdASoAAgACPm02m0mkIyKhIZRYEIANiWlu4XJwxZLjbB5U/WL+x9qP99sjv3F/ZcJe1t/nt8/AB9YOIDSiZAPQP+kfPL9Rewp+ufV4/bj2Qv2RBp9cdE96SBAgQIECBAgQIECBAgQIECBAgQIECBAgYGq4SZoxUi0AvBTNzvUYC1IL8Tony3/cMBakF+J0T3pIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgPjkCconhIWPUL5Xo7aIszZOJuVIL8TonvSQIEB8IfkD7YkBBYPELVVnMEQwMwgLd5Es/h/cMBakF+J0T2aLO3m0GdsX4aL0CN5CKd/I3ojE96SBAgQIEBIdQDRI7TUSgOzphkwsYG5xOie9JAgQHw2bVdGWeQ6S3XI0lHX1OYf3DAWpBfic90Li0k5gaWqyuqN7OOKc5IN1wwFqQX4jo1ih59LIM7stRlYNAowFqQX4nROmRC4X6AMR8j39bYuvr+4YC1IL8TedTyvtc6+YLQxAUXWlfYT/cMBakF+I6fhv4OT5r0mTXCaIrg9QSfnRPekgQICRNQ1eBEz/cLigA0zw+cU8TonvSQIEB8OHaIgz6O7cpPqpaBGdm2s5myW64YC1IL8To5Ud8hgA5Wyl91NPzFoGwnJr6QK/pSyXzlOXLly5cuXK3owgv6gBckgZgIkkQYtGlHyryK1ny1422ie9JAgQIECAoRmPiQoiJ70kCBAgQIEB8fu813inhgLUgvxOie9JAgWgKEYC1IL8TonvSQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAgQIECBAG1Cn+4YC1IL8TonvSQID4AAP7/MKAAAN0g+U+ImcfgCz2aKy4tNvylZV3Cv/wwUni2BNEBjnTHPGqscmeShAAAAAAAAAAAADYj+y+z0NN7R6a8RbDtfusZMx4etW8gtyViZe34vq3W3NjOiOV0qRDklCsSVvOrcF5V9WC0L4zbxWLXVhIbqnl80h/HDhw74+xFxxmtsYX5cC3w4m79qsrVjZ37egibdk2Ge6Qu68EslcdUf6rog27AFCm99fBcOsq4fOSy0iGr5r5f8I63WS67KRWyrZ27vV1uEXysWJdWkEhXemFO6vNoGF9/9zunp+A0MWM3weOtuGaFDpkxdlJOR7uIigJtU1cJkAA9R8aToMqOX4rQIdqjDnETQBX33kvzRVW1csWjjxBU9Yf6rE07KNQOy04qetnkZ3CdSXTxmpaOn9T8PlOD1L9vLrjbmnf07z60lNFprW3AfLBLgnrOsZ8VoOveR1HliiAEo4QJg33k1KYSbPXNyN+wyZBsuddhix7aZYt2Isz2l0Z6DnQIRUaqFR3Yu7Bc9xyFNLbUifeFjXdzhboteWMO/UTEeQWtcLTyLyaUOlopJYUU7Zzz/bdgx/ad4cKWPkRd1JP6UAzr+/JAy4dhTw4DhduXAc/NUTd7USztOlHPlq/aGQVT3A/6v597iQf0PlaSdvE1BTU2Z05kiWAka8N6SKr+Aw+x8QzxYOoEHA+CiDWwtW54ScLGC+cXIexAb65A3XyQ1JTmgTmXVBa5FFTvCI98pm//ueu+4OPajrgFaZb+Od7g/sCw5KiGnFlI5gu3KQzAkDJlevAInyOP/WJ4koNusg4FuFuknXI/Enhju9Vk2hPmfzLzeXNP7E1RRXMIQJqjDMHovfeyNtF/q+c8wGOYJ3yOmxg7HUoHIoztGO5UDtwM81B7muLfhrJ0KTwCuLyQ1r+v6CikUIK7ZyiXNBhwkz44EYO5uKk63W1TqIsYeRgkPPZjsqRbafuddHvf3MjxeZ/RICg3eswECK6A7S1lWwKTcSiitUOWhyC6QW7jsDIQT49rrnT+K+EoryGdhsN51UAmBQ+uYHhVrBvVQX9GneRRAMVHBCleZ3Iynm+KABdh15vySt7UNzltzo1iySExQgU10YykYowoG3oE4125RiGiOakea8W1W+giqu+AOebuB7UvEvFJcEEIPY3AAX49W/Vrlhsbltrh8DN5p+E8h7CS6DlkQTnWF/4iFndcv5LDOTT+lmqlPRdzxiA4N0wL4px6MG5f8qvgdfp3gyg0+mYJLN+FQ/lhnvb19guhXb+fOcCDzwIjViWtoO38UrltJEsaT4w1TGNCEgsKAtuPD5U47DU3NDVqRxAOqIgsFNEJsmT+U9gxZWKZ09pku5T5EBrcRciU4M2LH1uyE/IXEoKM4aCfZhwSnaKE9+lFcvdqpbJn16TA5n+v9HhBvwfjGV2S8qNnLx7ek57DEEYceTzp2rIoFbj68/EJ1Ji4j34Aq97vcKQGiuaioY1sfP0+xxZxvusGf3nz4ETKoYlRyhgTebEiHBGo/eBNUdsGgWE/a7WklCM5ouTocXNl/VJ9XpVUNMq+mIW1UOjiEHyX8Tp1EKQl/Z/1LsOsCku3gZHOqsHU6Se3AeghZ4DaSxFbgs8I9dG0FONqEH/d45QKPO8WFAAAAAAAAAAAAAA=",
"bodyEncoding": "base64"
}
}
]
}
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+87 -4
View File
@@ -1,10 +1,13 @@
import { describe, expect } from "bun:test"
import { Effect, Layer } from "effect"
import { Effect, Layer, Stream } from "effect"
import { HttpClientRequest } from "effect/unstable/http"
import { Image, ImageClient, Media } from "../src/index.js"
import { Google, OpenAI, XAI, ZAI } from "../src/providers.js"
import { BlackForestLabs, Fal, Google, OpenAI, Replicate, Stability, XAI, ZAI } from "../src/providers.js"
import { it } from "./lib/effect.js"
import { dynamicResponse } from "./lib/http.js"
import { dynamicResponse, json } from "./lib/http.js"
const layer = (handler: Parameters<typeof dynamicResponse>[0]) =>
ImageClient.layer.pipe(Layer.provideMerge(dynamicResponse(handler)))
describe("Image", () => {
for (const provider of [OpenAI, Google, XAI, ZAI]) {
@@ -69,7 +72,11 @@ describe("Image", () => {
expect(response.images).toHaveLength(2)
expect(response.image.mediaType).toBe("image/webp")
expect(response.image.source).toEqual({ type: "bytes", data: Uint8Array.from([1, 2, 3]), mediaType: "image/webp" })
expect(response.image.source).toEqual({
type: "bytes",
data: Uint8Array.from([1, 2, 3]),
mediaType: "image/webp",
})
expect(yield* response.image.bytes()).toEqual(Uint8Array.from([1, 2, 3]))
expect(response.image.providerMetadata).toEqual({ openai: { revisedPrompt: "A precise robot" } })
expect(response.usage).toMatchObject({ type: "tokens", total: 12 })
@@ -692,4 +699,80 @@ describe("Image", () => {
),
),
)
it.effect("rejects what a route cannot honor before sending anything", () =>
Effect.gen(function* () {
const openai = OpenAI.configure({ apiKey: "test" })
const replicate = Replicate.configure({ apiKey: "test" }).image("black-forest-labs/flux-schnell")
const prompt = "A lighthouse"
const errors = yield* Effect.all(
[
Image.start({ model: Google.configure({ apiKey: "test" }).image("gemini-3.1-flash-image"), prompt }),
Image.start({
model: BlackForestLabs.configure({ apiKey: "test" }).image("flux-2-pro"),
prompt,
aspectRatio: "16:9",
}),
Image.start({
model: Fal.configure({ apiKey: "test" }).image("fal-ai/nano-banana-2"),
prompt,
size: "512x512",
}),
Stream.runCollect(Image.stream({ model: openai.image("dall-e-3"), prompt })),
Stream.runCollect(Image.stream({ model: openai.image("gpt-image-2"), prompt, n: 2 })),
Image.start({ model: replicate, prompt, seed: 7 }),
Image.start({
model: replicate,
prompt,
providerOptions: { image: Media.bytes(new Uint8Array(300 * 1024), "image/png") },
}),
Image.start({ model: Stability.configure({ apiKey: "test" }).upscale(), prompt }),
].map((effect) => Effect.flip(effect)),
)
expect(errors.map((error) => [error.reason._tag, "operation" in error.reason && error.reason.operation])).toEqual(
[
["UnsupportedOperation", "image.start"],
["UnsupportedOperation", "media.aspectRatio"],
["UnsupportedOperation", "media.size"],
["UnsupportedOperation", "media.stream"],
["UnsupportedOperation", "media.n"],
["UnsupportedOperation", "media.seed"],
["InvalidRequest", false],
["InvalidRequest", false],
],
)
}).pipe(Effect.provide(layer(() => Effect.die("an unsupported request reached the network")))),
)
const moderated = { id: "req_1", status: "Content Moderated" }
const prediction = {
id: "p_1",
status: "succeeded",
output: { text: "not an image" },
urls: { get: "https://replicate.test/p_1", cancel: "https://replicate.test/p_1/cancel" },
}
it.effect("classifies terminal outcomes the recordings never saw", () =>
Effect.gen(function* () {
const bfl = yield* Image.resume(BlackForestLabs.configure({ apiKey: "test" }).image("flux-2-pro"), {
id: "req_1",
pollingURL: "https://bfl.test/v1/get_result?id=req_1",
}).pipe(
Effect.flatMap((generation) => generation.await()),
Effect.flip,
)
const replicate = yield* Image.generate({
model: Replicate.configure({ apiKey: "test", baseURL: "https://replicate.test" }).image("owner/model"),
prompt: "A lighthouse",
}).pipe(Effect.flip)
expect(bfl.reason).toMatchObject({ _tag: "ContentPolicy", body: JSON.stringify(moderated) })
expect(replicate.reason).toMatchObject({ _tag: "InvalidProviderOutput", body: JSON.stringify(prediction) })
}).pipe(
Effect.provide(
layer((input) =>
Effect.succeed(json(input, input.request.url.startsWith("https://bfl.test") ? moderated : prediction)),
),
),
),
)
})
+9 -1
View File
@@ -10,7 +10,7 @@ import {
type ImageRoute,
} from "../src/index.js"
import type { Service } from "../src/image-client.js"
import { Anthropic, Google, OpenAI, XAI, ZAI } from "../src/providers.js"
import { Anthropic, BlackForestLabs, Google, OpenAI, Stability, XAI, ZAI } from "../src/providers.js"
type Requirements<T> = T extends Effect.Effect<infer _A, infer _E, infer R> ? R : never
type Equal<A, B> = [A, B] extends [B, A] ? true : false
@@ -95,6 +95,14 @@ Image.generate({ model: openai, prompt: "A lighthouse", providerOptions: { nativ
Image.generate({ model: openai, prompt: "A lighthouse", providerOptions: { quality: 1 } })
// @ts-expect-error Known OpenAI numeric options retain their value kind.
Image.generate({ model: openai, prompt: "A lighthouse", providerOptions: { outputCompression: "80" } })
// @ts-expect-error Partial image counts are numeric.
Image.stream({ model: openai, prompt: "A lighthouse", providerOptions: { partialImages: "1" } })
const bfl = BlackForestLabs.configure({ apiKey: "test" }).image("flux-2-pro")
// @ts-expect-error Known BFL numeric options retain their value kind.
Image.start({ model: bfl, prompt: "A lighthouse", providerOptions: { safety_tolerance: "2" } })
const stability = Stability.configure({ apiKey: "test" })
// @ts-expect-error Only the creative upscaler is queued, so the selector takes no model id.
stability.upscale("fast")
OpenAI.imageGeneration({ action: "future-action", quality: "future-quality", size: "2048x2048" })
// @ts-expect-error Hosted image generation numeric options retain their value kind.
OpenAI.imageGeneration({ partialImages: "2" })
+29 -1
View File
@@ -1,4 +1,5 @@
import { Effect, Layer, Ref } from "effect"
import { Effect, Fiber, Layer, Ref } from "effect"
import * as TestClock from "effect/testing/TestClock"
import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { LLMClient, RequestExecutor } from "../../src/route.js"
import type { Service as LLMClientService } from "../../src/route/client.js"
@@ -17,6 +18,33 @@ export type Handler = (
input: HandlerInput,
) => Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError.HttpClientError>
export interface Call {
readonly method: string
readonly url: string
readonly headers: Headers
readonly body: string
}
/** Record every request and tell the handler how many times this exact method+URL has been seen (1-based). */
export const observe = (calls: Array<Call>, input: HandlerInput) =>
Effect.gen(function* () {
const web = yield* HttpClientRequest.toWeb(input.request).pipe(Effect.orDie)
const call = { method: web.method, url: web.url, headers: web.headers, body: input.text }
calls.push(call)
return { call, nth: calls.filter((seen) => seen.method === call.method && seen.url === call.url).length }
})
export const json = (input: HandlerInput, value: unknown, init?: ResponseInit) =>
input.respond(JSON.stringify(value), { ...init, headers: { "content-type": "application/json", ...init?.headers } })
/** Fork the polling program, let the test clock cover `seconds` of polling, and join. */
export const settle = <A, E, R>(program: Effect.Effect<A, E, R>, seconds: number) =>
Effect.gen(function* () {
const fiber = yield* Effect.forkChild(program)
yield* TestClock.adjust(`${seconds} seconds`)
return yield* Fiber.join(fiber)
})
export const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
Layer.succeed(
HttpClient.HttpClient,
+32 -6
View File
@@ -4,8 +4,8 @@ import { HttpClientRequest } from "effect/unstable/http"
import { AIError, LLMEvent, Media, SpeechEvent, TranscriptionEvent } from "../src/index.js"
import { RequestExecutor } from "../src/route.js"
import { AI } from "../src/promise.js"
import { AssemblyAI, OpenAI, Runway } from "../src/providers.js"
import { handlerLayer } from "./lib/http.js"
import { AssemblyAI, OpenAI, Replicate, Runway } from "../src/providers.js"
import { handlerLayer, json } from "./lib/http.js"
import { sseEvents } from "./lib/sse.js"
const openai = OpenAI.configure({ apiKey: "test", baseURL: "https://openai.test/v1" })
@@ -29,8 +29,28 @@ const executor = (seen: Array<string>) =>
const web = yield* HttpClientRequest.toWeb(input.request).pipe(Effect.orDie)
seen.push(web.url)
if (web.url.endsWith("/images/generations"))
return input.respond(JSON.stringify({ data: [{ b64_json: "AQID" }], output_format: "png" }), {
headers: { "content-type": "application/json" },
return JSON.parse(input.text).stream === true
? input.respond(
sseEvents(
{
type: "image_generation.partial_image",
b64_json: "AQ==",
partial_image_index: 0,
output_format: "png",
},
{ type: "image_generation.completed", b64_json: "AQID", output_format: "png" },
),
{ headers: { "content-type": "text/event-stream" } },
)
: input.respond(JSON.stringify({ data: [{ b64_json: "AQID" }], output_format: "png" }), {
headers: { "content-type": "application/json" },
})
if (web.url.startsWith("https://replicate.test"))
return json(input, {
id: "p_1",
status: "succeeded",
output: "https://replicate.test/a.webp",
urls: { get: "https://replicate.test/p_1", cancel: "https://replicate.test/p_1/cancel" },
})
if (web.url.endsWith("/chat/completions"))
return input.respond(chatBody, { headers: { "content-type": "text/event-stream" } })
@@ -89,7 +109,7 @@ const executor = (seen: Array<string>) =>
)
describe("AI promise client", () => {
test("generates text, images, and streams over one managed runtime", async () => {
test("generates text, images, and streams over one managed runtime, and queues images", async () => {
const seen: Array<string> = []
const ai = AI.make({ layer: executor(seen) })
@@ -111,7 +131,7 @@ describe("AI promise client", () => {
for await (const event of ai.image.stream({ model: openai.image("gpt-image-2"), prompt: "A lighthouse" })) {
imageEvents.push(event.type)
}
expect(imageEvents).toEqual(["image", "finish"])
expect(imageEvents).toEqual(["image-partial", "image", "finish"])
expect(seen).toEqual([
"https://openai.test/v1/chat/completions",
@@ -119,6 +139,12 @@ describe("AI promise client", () => {
"https://openai.test/v1/chat/completions",
"https://openai.test/v1/images/generations",
])
const generation = await ai.image.start({
model: Replicate.configure({ apiKey: "test", baseURL: "https://replicate.test" }).image("owner/model"),
prompt: "A lighthouse",
})
expect((await generation.await()).image.source).toMatchObject({ url: "https://replicate.test/a.webp" })
await ai.dispose()
})
+4
View File
@@ -196,6 +196,10 @@ describe("provider error classification", () => {
classifyProviderFailure({ message: messages.error.message, rawBody: JSON.stringify(messages) }),
].map((failure) => failure._tag),
).toEqual(["ContentPolicy", "ContentPolicy", "ContentPolicy"])
const stability = { id: "a1b2", name: "content_moderation", errors: ["Your request was flagged"] }
expect(
classifyProviderFailure({ message: "Forbidden", status: 403, rawBody: JSON.stringify(stability) })._tag,
).toBe("ContentPolicy")
})
test("recovers policy codes that OpenCode Zen preserves only in its message label", () => {
@@ -0,0 +1,39 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Image } from "../../src/index.js"
import { BlackForestLabs } from "../../src/providers.js"
import { dimensions } from "../lib/image.js"
import { recordedTests } from "../recorded-test.js"
import { queuedPoll } from "./queued-recording.js"
const model = BlackForestLabs.configure({
apiKey: process.env.BFL_API_KEY ?? "fixture",
}).image("flux-2-klein-4b")
const recorded = recordedTests({
prefix: "bfl-images",
provider: "black-forest-labs",
protocol: "bfl-images",
requires: ["BFL_API_KEY"],
})
describe("Black Forest Labs Images recorded", () => {
recorded.effect(
"submits, resumes, and polls an image",
() =>
Effect.gen(function* () {
const started = yield* Image.start({
model,
prompt: "A simple flat black circle centered on a plain white background.",
size: "512x512",
format: "jpeg",
})
const generation = yield* Image.resume(model, JSON.parse(JSON.stringify(started.token)))
const response = yield* generation.await({ poll: queuedPoll })
expect(response.image.source.type).toBe("bytes")
expect(dimensions(yield* response.image.bytes())).toEqual({ width: 512, height: 512 })
}),
{ timeout: 15 * 60 * 1000 },
)
})
@@ -0,0 +1,41 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Image } from "../../src/index.js"
import { Fal } from "../../src/providers.js"
import { dimensions } from "../lib/image.js"
import { recordedTests } from "../recorded-test.js"
import { queuedPoll } from "./queued-recording.js"
const model = Fal.configure({
apiKey: process.env.FAL_KEY ?? "fixture",
}).image("fal-ai/flux/schnell")
const recorded = recordedTests({
prefix: "fal-images",
provider: "fal",
protocol: "fal-images",
requires: ["FAL_KEY"],
})
describe("fal Images recorded", () => {
recorded.effect(
"generates an image",
() =>
Effect.gen(function* () {
const response = yield* Image.generate(
{
model,
prompt: "A simple flat black circle centered on a plain white background.",
size: "512x512",
format: "jpeg",
},
{ poll: queuedPoll },
)
expect(response.images).toHaveLength(1)
expect(response.image.info).toEqual({ width: 512, height: 512 })
expect(dimensions(yield* response.image.bytes())).toEqual({ width: 512, height: 512 })
}),
{ timeout: 15 * 60 * 1000 },
)
})
@@ -1,6 +1,6 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Image, Media } from "../../src/index.js"
import { Effect, Stream } from "effect"
import { Image, ImageEvent, Media } from "../../src/index.js"
import { OpenAI } from "../../src/providers.js"
import { dimensions } from "../lib/image.js"
import { recordedTests } from "../recorded-test.js"
@@ -60,4 +60,24 @@ describe("OpenAI Images recorded", () => {
expect(dimensions(yield* response.image.bytes())).toEqual({ width: 1024, height: 1024 })
}),
)
recorded.effect("streams a partial image before the final image", () =>
Effect.gen(function* () {
const events = Array.from(
yield* Image.stream({
model,
prompt: "A simple flat black circle centered on a plain white background.",
size: "1024x1024",
format: "jpeg",
providerOptions: { quality: "low", outputCompression: 10, partialImages: 1 },
}).pipe(Stream.runCollect),
)
expect(events.map((event) => event.type)).toEqual(["image-partial", "image", "finish"])
const image = events.find(ImageEvent.is.image)
expect(image?.image.mediaType).toBe("image/jpeg")
expect(dimensions(yield* image!.image.bytes())).toEqual({ width: 1024, height: 1024 })
expect(events.find(ImageEvent.is.finish)?.usage).toMatchObject({ type: "tokens" })
}),
)
})
@@ -0,0 +1,41 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Image } from "../../src/index.js"
import { Replicate } from "../../src/providers.js"
import { recordedTests } from "../recorded-test.js"
import { queuedPoll } from "./queued-recording.js"
const model = Replicate.configure({
apiKey: process.env.REPLICATE_API_TOKEN ?? "fixture",
}).image("black-forest-labs/flux-schnell")
const recorded = recordedTests({
prefix: "replicate-images",
provider: "replicate",
protocol: "replicate-images",
requires: ["REPLICATE_API_TOKEN"],
})
describe("Replicate Images recorded", () => {
recorded.effect(
"waits for a prediction and reads its output",
() =>
Effect.gen(function* () {
const response = yield* Image.generate(
{
model,
prompt: "A simple flat black circle centered on a plain white background.",
providerOptions: { aspect_ratio: "1:1", output_format: "webp", megapixels: "0.25" },
http: { headers: { Prefer: "wait=60" } },
},
{ poll: queuedPoll },
)
expect(response.images).toHaveLength(1)
expect(response.image.source.type).toBe("url")
expect(response.usage).toMatchObject({ type: "compute" })
expect((yield* response.image.bytes()).length).toBeGreaterThan(0)
}),
{ timeout: 15 * 60 * 1000 },
)
})
@@ -0,0 +1,58 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Image, Media } from "../../src/index.js"
import { Stability } from "../../src/providers.js"
import { dimensions } from "../lib/image.js"
import { recordedTests } from "../recorded-test.js"
import { queuedPoll } from "./queued-recording.js"
const stability = Stability.configure({ apiKey: process.env.STABILITY_API_KEY ?? "fixture" })
const recorded = recordedTests({
prefix: "stability-images",
provider: "stability",
protocol: "stability-images",
requires: ["STABILITY_API_KEY"],
// Multipart boundaries change on every request, so bodies cannot be matched.
options: { match: (incoming, recorded) => incoming.method === recorded.method && incoming.url === recorded.url },
})
describe("Stability Images recorded", () => {
recorded.effect("generates an image inline", () =>
Effect.gen(function* () {
const response = yield* Image.generate({
model: stability.image("core"),
prompt: "A simple flat black circle centered on a plain white background.",
aspectRatio: "1:1",
format: "jpeg",
})
expect(response.image.mediaType).toBe("image/jpeg")
expect(response.providerMetadata).toMatchObject({ stability: { finishReason: "SUCCESS" } })
expect(dimensions(yield* response.image.bytes()).width).toBeGreaterThan(0)
}),
)
recorded.effect(
"upscales an image through the results queue",
() =>
Effect.gen(function* () {
const generation = yield* Image.start({
model: stability.upscale(),
prompt: "A simple flat black circle on a white background.",
images: [
Media.bytes(
yield* Effect.promise(() => Bun.file("test/fixtures/images/edit-source.jpg").bytes()),
"image/jpeg",
),
],
format: "jpeg",
})
const response = yield* generation.await({ poll: queuedPoll })
expect(response.image.mediaType).toBe("image/jpeg")
expect(dimensions(yield* response.image.bytes()).width).toBeGreaterThan(1024)
}),
{ timeout: 15 * 60 * 1000 },
)
})
+2 -31
View File
@@ -1,42 +1,13 @@
import { describe, expect } from "bun:test"
import { Effect, Fiber, Layer, Stream } from "effect"
import * as TestClock from "effect/testing/TestClock"
import { HttpClientRequest } from "effect/unstable/http"
import { Effect, Layer, Stream } from "effect"
import { Media, Video, VideoClient, type GenerationEvent } from "../src/index.js"
import { Fal, Google, Runway, XAI } from "../src/providers.js"
import { it } from "./lib/effect.js"
import { dynamicResponse, type HandlerInput } from "./lib/http.js"
interface Call {
readonly method: string
readonly url: string
readonly headers: Headers
readonly body: string
}
/** Record every request and tell the handler how many times this exact method+URL has been seen (1-based). */
const observe = (calls: Array<Call>, input: HandlerInput) =>
Effect.gen(function* () {
const web = yield* HttpClientRequest.toWeb(input.request).pipe(Effect.orDie)
const call = { method: web.method, url: web.url, headers: web.headers, body: input.text }
calls.push(call)
return { call, nth: calls.filter((seen) => seen.method === call.method && seen.url === call.url).length }
})
const json = (input: HandlerInput, value: unknown, init?: ResponseInit) =>
input.respond(JSON.stringify(value), { ...init, headers: { "content-type": "application/json", ...init?.headers } })
import { dynamicResponse, json, observe, settle, type Call } from "./lib/http.js"
const layer = (handler: Parameters<typeof dynamicResponse>[0]) =>
VideoClient.layer.pipe(Layer.provideMerge(dynamicResponse(handler)))
/** Fork the polling program, let the test clock cover `seconds` of polling, and join. */
const settle = <A, E, R>(program: Effect.Effect<A, E, R>, seconds: number) =>
Effect.gen(function* () {
const fiber = yield* Effect.forkChild(program)
yield* TestClock.adjust(`${seconds} seconds`)
return yield* Fiber.join(fiber)
})
const DAY = 24 * 60 * 60 * 1000
// ---------------------------------------------------------------------------