fix(ai): surface fal moderation and compute, xAI cost, and upload query

This commit is contained in:
Shoubhit Dash
2026-09-27 18:04:18 +05:30
parent 01eb18144b
commit 6760cba80d
15 changed files with 150 additions and 24 deletions
+3 -2
View File
@@ -754,8 +754,8 @@ Status polls, result fetches, cancels, and asset downloads all run through the s
auth. `Generation.await` and `Generation.events` fail with a
`Timeout` reason when `poll.timeout` (default 10 minutes) elapses. Failed,
cancelled, and expired generations fail typed with the provider's terminal document on `reason.body`; moderation
outcomes (Veo `raiMediaFilteredReasons`, xAI `respect_moderation`, Runway `SAFETY.*` codes) surface as `notices` when
a video is still returned and as a `ContentPolicy` reason when nothing is.
outcomes (Veo `raiMediaFilteredReasons`, xAI `respect_moderation`, fal `has_nsfw_concepts`, Runway `SAFETY.*`
codes) surface as `notices` when a video is still returned and as a `ContentPolicy` reason when nothing is.
Provider notes:
@@ -921,6 +921,7 @@ Provider notes:
- **OpenAI** takes inline audio only; `diarize` needs `gpt-4o-transcribe-diarize`, timestamps need `whisper-1`, and `whisper-1` does not stream.
- **Gemini** needs a transcribe model (`gemini-3.5-transcribe`); `prompt` and `speakers` fail typed.
- **Deepgram** detects the language unless `language` is set; vocabulary goes in `providerOptions.keyterm`.
`providerOptions.multichannel: true` fails typed, because the response holds one transcript.
- **AssemblyAI** uploads inline audio before submitting and is the only route that accepts `speakers`.
The promise client mirrors the Effect API:
+2 -2
View File
@@ -168,8 +168,8 @@ Effect.gen(function* () {
// Simple: wait for it.
const response = yield* Video.generate(request, { poll: { interval: "10 seconds", timeout: "10 minutes" } })
response.video // Media.Asset: url (expiresAt on Veo and Runway; transient `headers` for Veo downloads)
response.usage // credits on Runway; the other three report none (xAI's usage.cost_in_usd_ticks is not decoded)
response.notices // Veo raiMediaFilteredReasons → filtered, xAI respect_moderation → moderated
response.usage // credits on Runway; the other three report none (xAI's USD cost is providerMetadata.xai.costInUsdTicks)
response.notices // Veo raiMediaFilteredReasons → filtered, xAI respect_moderation and fal has_nsfw_concepts → moderated
yield* response.video.materialize() // pull bytes before the URL expires
// Explicit generation control.
@@ -94,6 +94,12 @@ const query = (request: Request) =>
)
const fromRequest = Effect.fn("DeepgramTranscription.fromRequest")(function* (request: Request) {
// The response carries one transcript per channel, but the common response holds a single transcript.
if (request.providerOptions?.multichannel === true)
return yield* route.unsupported(
"transcription.multichannel",
`${route.name} multichannel returns a transcript per channel; transcribe each channel separately`,
)
const url = ProviderShared.mediaUrl(request.audio)
if (url !== undefined)
return MediaProtocol.json(mergeJsonRecords({ url }, request.http?.body) ?? {}, yield* query(request))
+4 -1
View File
@@ -40,6 +40,7 @@ const QueueResult = Schema.StructWithRest(
),
seed: optionalNull(Schema.Number),
has_nsfw_concepts: optionalNull(Schema.Array(Schema.Boolean)),
timings: optionalNull(Schema.Struct({ inference: optionalNull(Schema.Number) })),
}),
[Schema.Record(Schema.String, Schema.Unknown)],
)
@@ -111,10 +112,11 @@ const decodeResult = Effect.fn("FalImages.decodeResult")(function* (
context: MediaProtocol.PollContext<FalQueue.Token>,
) {
const output = yield* decodeQueueResult(response)
const { images, seed, has_nsfw_concepts, ...rest } = output.value
const { images, seed, has_nsfw_concepts, timings, ...rest } = output.value
if (images.length === 0) return yield* output.invalid(`${route.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] : []))
const inference = timings?.inference ?? undefined
return new ImageResponse({
images: images.map((image) => {
const info = { width: image.width ?? undefined, height: image.height ?? undefined }
@@ -131,6 +133,7 @@ const decodeResult = Effect.fn("FalImages.decodeResult")(function* (
type: "moderated" as const,
message: `${route.name} flagged image ${index} as NSFW`,
})),
usage: inference === undefined ? undefined : { type: "compute", seconds: inference },
providerMetadata: { fal: { requestId: context.token.requestID, seed: seed ?? undefined, ...rest } },
})
})
+5 -1
View File
@@ -38,6 +38,7 @@ const QueueResult = Schema.StructWithRest(
file_size: optionalNull(Schema.Number),
}),
seed: optionalNull(Schema.Number),
has_nsfw_concepts: optionalNull(Schema.Array(Schema.Boolean)),
}),
[Schema.Record(Schema.String, Schema.Unknown)],
)
@@ -84,9 +85,12 @@ const decodeResult = Effect.fn("FalVideo.decodeResult")(function* (
context: MediaProtocol.PollContext<FalQueue.Token>,
) {
const output = yield* decodeQueueResult(response)
const { video, seed, ...rest } = output.value
const { video, seed, has_nsfw_concepts, ...rest } = output.value
return new VideoResponse({
videos: [Media.url(video.url, { mediaType: video.content_type ?? "video/mp4" })],
notices: has_nsfw_concepts?.includes(true)
? [{ type: "moderated", message: `${route.name} flagged the generated video as NSFW` }]
: undefined,
providerMetadata: {
fal: {
requestId: context.token.requestID,
+4 -5
View File
@@ -39,7 +39,7 @@ const XAIImageResponse = Schema.Struct({
mime_type: optionalNull(Schema.String),
}),
),
usage: Schema.optional(Schema.Unknown),
usage: optionalNull(Schema.Struct({ cost_in_usd_ticks: optionalNull(Schema.Number) })),
})
// ---------------------------------------------------------------------------
@@ -100,12 +100,11 @@ const decodeResponse = Effect.fn("XAIImages.decodeResponse")(function* (
}),
)
if (images.length === 0) return yield* output.invalid(`${route.name} returned no images`)
const usage = ProviderShared.isRecord(decoded.usage) ? decoded.usage : undefined
// xAI reports a USD cost (`cost_in_usd_ticks`) rather than tokens, seconds, or credits; the raw record stays in
// provider metadata.
// xAI reports a USD cost (1 tick = 1e-10 USD) rather than tokens, seconds, or credits, so it stays in metadata.
const costInUsdTicks = decoded.usage?.cost_in_usd_ticks ?? undefined
return new ImageResponse({
images,
providerMetadata: usage === undefined ? undefined : { xai: { usage } },
providerMetadata: costInUsdTicks === undefined ? undefined : { xai: { costInUsdTicks } },
})
})
+9 -1
View File
@@ -56,6 +56,7 @@ const VideoStatus = Schema.Struct({
}),
),
model: optionalNull(Schema.String),
usage: optionalNull(Schema.Struct({ cost_in_usd_ticks: optionalNull(Schema.Number) })),
})
const STATUS = {
@@ -165,7 +166,14 @@ const decodeResult = Effect.fn("XAIVideo.decodeResult")(function* (
}),
],
notices,
providerMetadata: { xai: { requestId: context.token.requestID, model: decoded.model ?? undefined } },
// xAI reports a USD cost (1 tick = 1e-10 USD) rather than seconds or credits, so it stays in metadata.
providerMetadata: {
xai: {
requestId: context.token.requestID,
model: decoded.model ?? undefined,
costInUsdTicks: decoded.usage?.cost_in_usd_ticks ?? undefined,
},
},
})
})
+15 -8
View File
@@ -281,7 +281,18 @@ const makeTransport = <Request extends MediaRequest>(
) => {
const routeHttp = input.headers === undefined ? undefined : new HttpOptions({ headers: input.headers })
const authorize = Auth.toEffect(input.auth)
const baseURL = (path: string) => new URL(`${Endpoint.trimBaseUrl(input.endpoint.baseURL ?? "")}${path}`)
/** URL for a call beside the submit (upload, status, result, cancel), carrying the request and endpoint query. */
const callURL = (path: string, http: HttpOptions | undefined) => {
// Provider-issued absolute URLs (fal `status_url`) are used as-is; everything else resolves against the base.
const url = withQuery(
/^https?:\/\//.test(path)
? new URL(path)
: new URL(`${Endpoint.trimBaseUrl(input.endpoint.baseURL ?? "")}${path}`),
http?.query,
)
for (const [key, value] of Object.entries(input.endpoint.query ?? {})) url.searchParams.set(key, value)
return url
}
/** `auth` is only what `Auth` added or changed, never untouched deployment headers. */
const send = Effect.fn("MediaRoute.send")(function* (
call: {
@@ -330,7 +341,7 @@ const makeTransport = <Request extends MediaRequest>(
protocol.prepare === undefined
? request
: yield* protocol.prepare(request, (path, body) =>
send({ method: "POST", url: baseURL(path), headers, request, body }, execute).pipe(
send({ method: "POST", url: callURL(path, http), headers, request, body }, execute).pipe(
Effect.map((sent) => sent.response),
),
)
@@ -348,12 +359,8 @@ const makeTransport = <Request extends MediaRequest>(
return { response: sent.response, context: { request: resolved, body } }
}),
/** Bodiless follow-up call (status, result, cancel) with the same auth and headers as `submit`. */
call: (method: AuthInput["method"], path: string, http: HttpOptions | undefined, execute: Execute) => {
// Provider-issued absolute URLs (fal `status_url`) are used as-is; everything else resolves against the base.
const url = withQuery(/^https?:\/\//.test(path) ? new URL(path) : baseURL(path), http?.query)
for (const [key, value] of Object.entries(input.endpoint.query ?? {})) url.searchParams.set(key, value)
return send({ method, url, headers: Headers.fromInput(http?.headers), request: { http } }, execute)
},
call: (method: AuthInput["method"], path: string, http: HttpOptions | undefined, execute: Execute) =>
send({ method, url: callURL(path, http), headers: Headers.fromInput(http?.headers), request: { http } }, execute),
}
}
+27
View File
@@ -866,6 +866,33 @@ describe("Image", () => {
),
)
it.effect("reports fal inference time as compute usage", () =>
Effect.gen(function* () {
const generation = yield* Image.resume(Fal.configure({ apiKey: "test" }).image("fal-ai/flux/schnell"), falToken)
const response = yield* generation.await()
expect(response.usage).toEqual({ type: "compute", seconds: 0.67 })
expect(response.notices).toEqual([{ type: "moderated", message: "fal Images flagged image 0 as NSFW" }])
expect(response.providerMetadata).toEqual({ fal: { requestId: "r1", seed: 3, prompt: "A lighthouse" } })
}).pipe(
Effect.provide(
layer((input) =>
Effect.succeed(
input.request.url === falToken.statusURL
? json(input, { status: "COMPLETED" })
: json(input, {
images: [{ url: "https://v3.fal.media/out.jpg", content_type: "image/jpeg" }],
timings: { inference: 0.67 },
seed: 3,
has_nsfw_concepts: [true],
prompt: "A lighthouse",
}),
),
),
),
),
)
const moderated = { id: "req_1", status: "Content Moderated" }
const prediction = {
id: "p_1",
@@ -35,6 +35,7 @@ describe("fal Images recorded", () => {
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 })
expect(response.usage).toMatchObject({ type: "compute" })
}),
{ timeout: 15 * 60 * 1000 },
)
@@ -29,6 +29,7 @@ describe("xAI Images recorded", () => {
expect(response.images).toHaveLength(1)
expect(response.image.mediaType.startsWith("image/")).toBe(true)
expect((yield* response.image.bytes()).length).toBeGreaterThan(0)
expect(response.providerMetadata?.xai?.costInUsdTicks).toBeGreaterThan(0)
}),
)
+2 -2
View File
@@ -39,7 +39,7 @@ describe("xAI Images", () => {
expect(response.images[1].mediaType).toBe("application/octet-stream")
expect(response.images[1].source).toEqual({ type: "url", url: "https://api.xai.test/image.jpg" })
expect(response.usage).toBeUndefined()
expect(response.providerMetadata).toEqual({ xai: { usage: { num_images: 2 } } })
expect(response.providerMetadata).toEqual({ xai: { costInUsdTicks: 200000000 } })
}).pipe(
Effect.provide(
ImageClient.layer.pipe(
@@ -67,7 +67,7 @@ describe("xAI Images", () => {
{ b64_json: "AQID", url: null, mime_type: "image/jpeg" },
{ b64_json: null, url: "https://api.xai.test/image.jpg", mime_type: null },
],
usage: { num_images: 2 },
usage: { cost_in_usd_ticks: 200000000 },
}),
{ headers: { "content-type": "application/json" } },
)
@@ -42,6 +42,7 @@ describe("xAI Video recorded", () => {
expect(response.video.source.type).toBe("url")
expect(response.video.info?.durationSeconds).toBe(2)
expect((yield* response.video.bytes()).length).toBeGreaterThan(0)
expect(response.providerMetadata?.xai?.costInUsdTicks).toBeGreaterThan(0)
}),
{ timeout: 15 * 60 * 1000 },
)
+33
View File
@@ -47,6 +47,7 @@ describe("Transcription", () => {
Transcription.generate({ model: deepgram, audio, prompt: "OpenCode" }),
Transcription.generate({ model: google, audio, speakers: 2 }),
Transcription.start({ model: deepgram, audio }),
Transcription.generate({ model: deepgram, audio, providerOptions: { multichannel: true } }),
Transcription.generate({ model: deepgram, audio, http: { body: { callback: "https://hook.test" } } }),
Transcription.generate({
model: openai.transcription("gpt-transcribe"),
@@ -74,6 +75,7 @@ describe("Transcription", () => {
["UnsupportedOperation", "media.prompt"],
["UnsupportedOperation", "media.speakers"],
["UnsupportedOperation", "transcription.start"],
["UnsupportedOperation", "transcription.multichannel"],
["InvalidRequest", false],
["InvalidRequest", false],
["InvalidRequest", false],
@@ -439,6 +441,37 @@ describe("Transcription", () => {
}),
)
it.effect("sends the configured and request query on the AssemblyAI upload like the submit", () =>
Effect.gen(function* () {
const calls: Array<Call> = []
yield* Transcription.start({
model: AssemblyAI.configure({
apiKey: "aai-key",
baseURL: "https://assemblyai.test",
http: { query: { tenant: "t1" } },
}).transcription("universal-3-5-pro"),
audio,
http: { query: { trace: "1" } },
}).pipe(
Effect.provide(
layer((input) =>
observe(calls, input).pipe(
Effect.map(({ call }) =>
call.url.includes("/v2/upload")
? json(input, { upload_url: "https://cdn.assemblyai.test/upload/1" })
: json(input, { id: "tr_1", status: "queued" }),
),
),
),
),
)
expect(calls.map((call) => call.url)).toEqual([
"https://assemblyai.test/v2/upload?tenant=t1&trace=1",
"https://assemblyai.test/v2/transcript?tenant=t1&trace=1",
])
}),
)
it.effect("enables AssemblyAI speaker labels when only an expected speaker count is given", () =>
Effect.gen(function* () {
const calls: Array<Call> = []
+37 -2
View File
@@ -290,6 +290,7 @@ describe("Video / xAI", () => {
status: "done",
video: { url: "https://vidgen.x.ai/out.mp4", duration: 10, respect_moderation: true },
model: "grok-imagine-video-1.5",
usage: { cost_in_usd_ticks: 1600000000 },
})
}),
),
@@ -302,7 +303,10 @@ describe("Video / xAI", () => {
expect(response.video.source).toEqual({ type: "url", url: "https://vidgen.x.ai/out.mp4", mediaType: "video/mp4" })
expect(response.video.info).toEqual({ durationSeconds: 10 })
expect(response.notices).toBeUndefined()
expect(response.providerMetadata).toEqual({ xai: { requestId: "req_1", model: "grok-imagine-video-1.5" } })
expect(response.usage).toBeUndefined()
expect(response.providerMetadata).toEqual({
xai: { requestId: "req_1", model: "grok-imagine-video-1.5", costInUsdTicks: 1600000000 },
})
}),
)
@@ -509,8 +513,9 @@ describe("Video / fal", () => {
url: "https://v3.fal.media/out.mp4",
mediaType: "video/mp4",
})
expect(response.notices).toBeUndefined()
expect(response.providerMetadata).toEqual({
fal: { requestId: "r1", seed: 7, fileName: "out.mp4", fileSize: 10, has_nsfw_concepts: [false] },
fal: { requestId: "r1", seed: 7, fileName: "out.mp4", fileSize: 10 },
})
expect(calls.map((call) => `${call.method} ${call.url}`)).toEqual([
"POST https://queue.fal.test/fal-ai/veo3.1",
@@ -524,6 +529,36 @@ describe("Video / fal", () => {
}),
)
it.effect("flags NSFW results as a moderated notice", () =>
Effect.gen(function* () {
const generation = yield* Video.resume(model, {
requestID: "r1",
statusURL: urls.status,
responseURL: urls.response,
cancelURL: urls.cancel,
})
const response = yield* generation.result()
expect(response.video.source).toEqual({
type: "url",
url: "https://v3.fal.media/out.mp4",
mediaType: "video/mp4",
})
expect(response.notices).toEqual([
{ type: "moderated", message: "fal Video flagged the generated video as NSFW" },
])
}).pipe(
Effect.provide(
layer((input) =>
Effect.succeed(
input.request.url === urls.status
? json(input, { status: "COMPLETED" })
: json(input, { video: { url: "https://v3.fal.media/out.mp4" }, has_nsfw_concepts: [false, true] }),
),
),
),
),
)
it.effect("treats a COMPLETED status carrying an error as failed", () =>
Effect.gen(function* () {
const generation = yield* Video.start({ model, prompt: "x" })