Media modalities

Speech, transcription, images and video — one provider config, one error type, one cost ledger. Plus the seam for a speech provider the gateway does not carry.

The gateway catalog names eight modalities. Four of them are media, and each has a primitive with the same shape as the rest of @voltro/ai: a ProviderConfig, an AiError failure channel, and usage that reaches the same cost ledger.

Modality	Primitive
`image`	`generateImage`
`video`	`generateVideo`
`speech`	`generateSpeech`
`transcription`	`transcribe`

Streaming raw audio with `streamTranscribe`

streamTranscribe returns an Effect Stream<TranscriptionStreamEvent, TranscriptionStreamError>. It keeps provider credentials server-side and emits partial text before the final result. It does not decode uploaded MP3, WAV, WebM or other containers. Supply a single-use ReadableStream<Uint8Array> of already-decoded raw audio and its actual format: audio/pcm with a positive integer sample rate, or audio/pcmu / audio/pcma. The selected provider/model must support that format and streaming transcription; an unsupported adapter fails explicitly.

import { streamTranscribe } from '@voltro/ai'

// rawAudio is supplied by your server's authorized audio source/decoder.
const events = streamTranscribe({
  audio: rawAudio,
  inputAudioFormat: { type: 'audio/pcm', rate: 24000 },
  provider: { name: 'gateway', model: 'openai/gpt-realtime-whisper' },
  signal: abortController.signal,
})

The event type describes how to update a transcript:

transcript-delta: append delta.
transcript-partial: replace the provisional text for that provider id/channel; it is not another committed sentence.
transcript-final: the provider finalized that segment, not the entire call.
finish: the complete text, timestamped segments, and optional language/durationSec. Only this event means successful completion.

The stream fails with TranscriptionStreamError.reason of input, unsupported, provider or cancelled; it never fabricates finish after failure. Error text and events exclude provider-private metadata, raw chunks and response headers. The original failure remains a non-enumerable server-side cause. Import TranscriptionStreamEvent and TranscriptionStreamError from @voltro/ai/events for browser-safe defineStream descriptor element and error schemas; the executor imports streamTranscribe from the server-only root. Guard and rate-limit the declared stream before permitting paid work.

External abort, scope interruption and early consumption termination cancel the provider and audio source. Backpressure when sending audio to the provider is preserved; Voltro does not first accumulate the entire audio input. There is no automatic retry, audio replay, durable checkpoint or billing estimate. Live microphone upload still needs an authorized duplex transport; a server-held raw recording can send partial results over the existing server-to-client stream. useTranscription remains the upload-plus-one-shot-action helper below; it does not consume these partial events or decode uploaded files into raw audio.

Through a declared RPC stream

Only a recording identifier crosses JSON RPC. This runnable example assumes a trusted, server-owned directory of single-channel, 24 kHz signed 16-bit PCM files named <recordingId>.pcm. The scope grants access to that whole directory: for a multi-tenant or per-record policy, resolve and authorize the recording inside the server handler before opening it. Do not use this directory layout as a substitute for that authorization, and do not put arbitrary uploads there.

// apps/api/streams/recordings.transcribeLive.stream.ts
import { Schema } from 'effect'
import { defineStream } from '@voltro/protocol'
import { TranscriptionStreamEvent, TranscriptionStreamError } from '@voltro/ai/events'

export default defineStream({
  name: 'recordings.transcribeLive',
  guards: [{ scope: 'recordings:transcribe' }],
  input: Schema.Struct({ recordingId: Schema.String.pipe(Schema.pattern(/^[a-zA-Z0-9_-]+$/)) }),
  element: TranscriptionStreamEvent,
  error: TranscriptionStreamError,
})

// apps/api/streams/recordings.transcribeLive.stream.server.ts
import { createReadStream } from 'node:fs'
import { Readable } from 'node:stream'
import { join } from 'node:path'
import { streamTranscribe } from '@voltro/ai'

export default (input: { recordingId: string }) => streamTranscribe({
  audio: Readable.toWeb(createReadStream(
    join('/var/lib/recordings', `${input.recordingId}.pcm`),
    { highWaterMark: 64 * 1024 },
  )) as ReadableStream<Uint8Array>,
  inputAudioFormat: { type: 'audio/pcm', rate: 24000 },
  provider: { name: 'gateway', model: 'openai/gpt-realtime-whisper' },
})

The ordinary client stream hook already carries these events. This single-channel view replaces provisional hypotheses, accumulates finalized segments, and treats only finish as complete. For interleaved utterances/channels, maintain separate buffers keyed by the provider's id and channelIndex instead.

import { useAgentStream } from '@voltro/client'
import type { TranscriptionStreamEvent } from '@voltro/ai/events'

export function RecordingTranscript({ recordingId }: { recordingId: string }) {
  const run = useAgentStream<TranscriptionStreamEvent>('app', 'recordings.transcribeLive')
  let committed = ''
  let provisional = ''
  let complete = false
  for (const event of run.events) {
    switch (event.type) {
      case 'transcript-partial': provisional = event.text; break
      case 'transcript-delta': provisional += event.delta; break
      case 'transcript-final':
        committed = [committed, event.text].filter(Boolean).join(' ')
        provisional = ''
        break
      case 'finish':
        committed = event.text
        provisional = ''
        complete = true
    }
  }
  return <section>
    <button onClick={() => run.start({ recordingId })}>Transcribe</button>
    <button onClick={run.cancel} disabled={run.status !== 'streaming'}>Stop</button>
    <p>{committed} <em>{provisional}</em></p>
    <p>{complete ? 'Complete' : run.status === 'error' ? 'Transcription failed' : 'Not complete'}</p>
  </section>
}

cancel() interrupts this RPC stream on its owning connection. That closes the server's audio source and provider socket; it is not confirmation that the provider refunded or stopped billing. These are transient events, not a durable transcript: persist the final result explicitly when the application needs it.

`transcribe` — audio to text

import { transcribe } from '@voltro/ai'

const { text, segments } = yield* transcribe({
  source: { storageRef: recording.ref },
  resolveRef: (ref) => ctx.storage.read(ref),
  timestamps: 'segment',
  language: 'de',
})

The source is a reference, not bytes

Audio is large and usually already somewhere. Pushing a ninety-minute recording through an rpc frame to hand it back to a provider is the wrong direction, so source takes three shapes and they are genuinely different:

{ storageRef } — the case to reach for. The app resolves it, because the app's storage is what knows about tenancy, signing and lifetime. The resolver is injected (resolveRef) rather than imported, so @voltro/ai never learns about buckets.
{ url } — fetched by transcribe. Convenient, but the bytes do pass through the process: the SDK takes bytes or base64 and never a location.
{ bytes, mediaType } — already in hand. The one that costs memory by construction.

Timestamps and speakers

timestamps: 'none' | 'segment' | 'word' is a request, not a default — providers charge for the detail. diarize: true asks for speaker separation; a provider that cannot do it returns segments without speaker rather than failing.

segments is absent when the provider returned none, never []. An empty array is a claim ("this audio has no speech"); absent is the truth ("the provider did not say").

`generateSpeech` — text to audio

const { audio } = yield* generateSpeech({
  text: 'Willkommen.',
  provider: { name: 'gateway', model: 'openai/tts-1-hd' },
  voice: 'alloy',
  outputFormat: 'mp3',
})

A provider the gateway does not carry

The gateway serves nine speech models. A real voice catalog runs to hundreds behind vendors it does not carry — Cartesia, ElevenLabs, Google Cloud TTS — and an app with its own adapter for those had nowhere to plug it in, so it called its own synthesizer and skipped the primitive entirely, losing the shared error type and cost path with it.

import { registerSpeechProvider, generateSpeech } from '@voltro/ai'

registerSpeechProvider('elevenlabs', {
  listVoices: () => …,
  synthesize: (request) => …,
})

yield* generateSpeech({ text, provider: { name: 'elevenlabs', model: 'eleven_turbo_v3' }, voice })

This ships no adapters, on purpose: bundling an ElevenLabs client would put the framework in the business of tracking somebody else's REST API, and the app already has the adapter. What the app deletes is its dispatch, not its adapters.

A registered provider receives every option the primitive already had — voice, outputFormat, speed, language, instructions, providerOptions — or the seam would be a narrower primitive wearing the same name.

Taking over a built-in name

mock, openai, gateway and anthropic resolve without a registration, and shadowing one by accident would make the same config mean different things depending on which module loaded first. So it takes a flag:

registerSpeechProvider('openai', myOpenAiTts, { replaceBuiltin: true })

The flag exists because a provider name is not always an alias you choose. openai is the value the gateway puts in its model ids, so it is what a stored catalog's rows carry and what a picker submits — an app told to pick another name cannot, without rewriting persisted rows and live sessions. Refusing outright did not protect that app; it made it keep its own dispatch for every provider, not just the colliding one.

What the refusal was protecting is kept: the intent is now in the call, so an override is a decision someone wrote down rather than an accident of load order. Registered providers are resolved before built-ins — otherwise the flag would let you register a name and never reach it.

Cost — speech is not billed in tokens

generateSpeech never records a ledger row itself; it returns usage and you pass it to recordAiUsage, the same as generateText. The difference is the unit. Every speech vendor meters characters of input or seconds of audio, so a usage shape that only knew tokens could not hold the call at all — speech spend was not un-priced, it was absent.

A registered provider returns a MeteredUsage:

synthesize: (request) => Effect.succeed({
  audio,
  usage: { unit: 'characters', input: request.text.length },
})

yield* recordAiUsage(store, {
  provider: 'elevenlabs', model: 'eleven_turbo_v3', operation: 'generateSpeech',
  usage: { unit: 'characters', input: 4200 },
  price: { inputPer1M: 30, outputPer1M: 0 },   // your rate, in your unit
})

The ledger row carries unit beside the two counts. A non-token unit is never priced by the token table: without a price or an actualCostUsd the row is written costSource: 'unpriced' with the real count and a zero cost, because a per-1M-token rate applied to a character count produces a number indistinguishable from a real one. Sum the count columns grouped by unit, never across it.

Image references and grounding

const result = yield* generateImage({
  prompt: 'Keep the subject; change the background to a garden.',
  inputs: { images: ['https://example.com/reference.png'] },
  provider: { name: 'openai', model: 'gpt-image-1' },
})
for (const call of result.calls) {
  const grounding = call.providerMetadata
  const images = call.images
}

inputs.images and optional inputs.mask accept URL strings, data URLs, base64, or Uint8Array. They reach the SDK's image-editing prompt, not the text. Provider support for editing and masks varies.

calls keeps each underlying call's images, usage and provider metadata together. Read grounding there when calls must stay apart: the top-level providerMetadata is rebuilt from the calls (a provider's fields from the first call, images concatenated), so a grounding block survives at the top too — but combining calls loses their scope. A gateway answer whose provider metadata carries no images list (a Gemini grounding answer) is accepted; it used to make the aggregation throw after the call had been charged. maxRetries is passed through to the SDK (default 2): a retried 500 is a paid retry, so set it to 0 where the caller retries on its own terms. Each image also carries its own providerMetadata. No HTTP response headers or bodies are copied into these records. These are returned values, not a durable ledger or browser attribution events.

Video references with roles

const result = yield* generateVideo({
  prompt: 'Move from the first frame towards the last.',
  provider: { name: 'gateway', model: 'google/veo-3.1-generate-001' },
  params: { generateAudio: false },
  inputs: {
    frameImages: [
      { frameType: 'first_frame', image: 'https://example.com/start.png' },
      { frameType: 'last_frame', image: { type: 'url', url: 'https://example.com/end.png', mediaType: 'image/png' } },
    ],
    references: [{ type: 'url', url: 'https://example.com/motion.mp4', mediaType: 'video/mp4' }],
  },
})

image, references and frameImages[].image accept URL shorthand or typed { type: 'url', url, mediaType? } / { type: 'file', data, mediaType } inputs. Roles, known media types and explicit generateAudio: false reach the provider. Support for any particular input combination depends on the selected model.

Media seeds

Set generateImage({ prompt, seed: 0, ... }) or generateVideo({ prompt, params: { seed: 0 }, ... }) to request the numeric seed zero. Zero is not an omitted seed: Gateway image/video HTTP requests retain it, as do asynchronous video operation starts through the resolved model. Nonzero seeds stay unchanged and an omitted seed stays omitted. Each call owns its seed, including simultaneous calls using the same model. This is included in @voltro/ai; no application-side dependency patch is needed. Seed support and reproducibility remain properties of the selected provider/model — transmission does not guarantee identical generated media.

Video calls also preserve numeric zero for durationSeconds and fps, and support aspectRatio: 'adaptive'. Explicit empty dimension strings received from JavaScript or unvalidated configuration are sent unchanged, not silently replaced by omitted parameters. This applies to image size/aspect ratio and video aspect ratio/resolution, including asynchronous video operation starts. It does not make empty dimensions valid: TypeScript excludes them, and the selected provider may reject them. Validate untrusted configuration before a paid call. Omitted parameters remain omitted; concurrent calls do not share them.

Cancellation

generateImage, generateSpeech and generateVideo accept abortSignal. Aborting it interrupts the Effect and signals pending provider I/O. Interrupting the calling fiber does the same. A signal aborted before execution prevents the provider call. Cancellation is interruption, not a generation failure to retry.

Registered speech adapters receive request.abortSignal; pass it to their HTTP client. Their Effects are interrupted too, so finalizers run. An adapter or remote service ignoring cancellation can still finish or charge a request: local interruption is not confirmation of remote job cancellation.

Inline video limits

generateVideo({ prompt, maxInlineBytes: 64 * 1024 * 1024 }) limits a provider's inline video result to 64 MiB by default. Base64 must be canonical, padded standard base64: malformed characters, whitespace, missing padding and non-zero padding bits are refused as AiError. The decoded length is checked before allocating the decoded byte buffer; binary results are checked against the same limit. Configure a non-negative safe integer byte limit, including zero to accept only empty inline results or URLs. URL results are returned without downloading them.

This limit bounds the inline result and its decoding allocation, not an SDK's already-buffered HTTP response or a later download by your application. It does not turn a media request into a durable, resumable provider job.

The gateway's response budget

On a gateway provider the HTTP response itself is bounded too, for images, video, speech and transcription alike. A declared Content-Length over the budget is refused before a byte is read; a response without one is read up to the budget and cut off there. The refusal is an AiError that names the budget. The budget is maxResponseBytes on the provider config, else VOLTRO_AI_MEDIA_MAX_RESPONSE_BYTES, else 128 MiB:

import { generateImage } from '@voltro/ai'

export const thumbnail = (prompt: string) =>
  generateImage({ prompt, provider: { name: 'gateway', model: 'openai/gpt-image-1-mini', maxResponseBytes: 8 * 1024 * 1024 } })

As a durable step — `transcribeStep`

A ninety-minute recording is not a request. aiStep and aiObjectStep have had offload: true since they shipped, and transcription — the modality that runs longest — is where it matters most:

const { text, language } = yield* transcribeStep({
  name: 'transcribe-recording',
  source: { storageRef: recording.ref },
  resolveRef: (ref) => ctx.storage.read(ref),
  offload: true,
})

Offloaded, the run parks on a durable signal and frees its worker; the dispatcher performs the call and resumes it. The queue row carries the source, never the bytes — putting the audio in the row would reintroduce exactly the cost offloading exists to remove. A ref source therefore needs resolveTranscriptionRef wired on the dispatcher too, since that is the process doing the read.

The journal keeps text, language and durationSec and not segments: a journal is read back on every replay, and segments for a long recording would make each replay pay for the transcript again. The segments belong wherever the app puts the transcript.

A media job as a durable step — `mediaStep`

Image, video and speech generation are external jobs: minutes long, paid on submission, owned by whichever process submitted them. mediaStep makes one a durable step that is always offloaded — there is no inline form on purpose.

mediaStep can run inside another step, including with a finite offloadTimeoutMs (one hour by default). It enqueues once, suspends while the result is outstanding, and reuses that inference identity when resumed. Its wait uses the shared suspending-signal boundary; do not replace the timeout with Infinity merely to make nested suspension work. A workflow suspension can run interruption finalizers and does not itself mean the media job failed.

Only code that the engine replays can park. Where a host runs work that is NOT replayed — an AI-Flows agentic planner runs its whole tool loop inside one step, whose re-run would plan differently — it declares so with SuspensionUnavailable (@voltro/workflow), and mediaStep, every offloaded call and awaitSignalSuspending refuse there with WorkflowCannotPark before anything is submitted. Without that, the job was submitted and paid for, and nothing read its result.

const { artifacts } = yield* mediaStep({
  name: 'render-poster',
  modality: 'image',
  prompt: `A poster for ${title}`,
  params: { n: 2, size: '1024x1024' },
  store: ctx.store,
  tenantId: ctx.subject.tenantId,
})
// artifacts: [{ url, refId, mediaType }] — hosted through the storage plugin

The queue row is the job's identity, allocated before the first paid call, so a replay finds the row instead of queueing a second job. The dispatcher performs the job, persists every artifact through the storage plugin's mediaPersist capability under <row>#<attempt>#<index> — a store with resumable puts keeps one object per attempt — and resumes the run with URLs and storage ref ids. Bytes never enter the queue row or the resume payload; inputs are URLs for the same reason. A provider video URL is re-hosted when the store can ingest it and handed on as-is when it cannot.

persist sets who the stored artifacts belong to, how they are served, and the tags they are found by — handed to the storage plugin as-is, so deletion, restore and listing by owner or tag work on generated media like on any upload. Without it an artifact is stored owner-less, private and untagged.

const { artifacts } = yield* mediaStep({
  name: 'render-poster',
  modality: 'image',
  prompt: `A poster for ${title}`,
  store: ctx.store,
  tenantId: ctx.subject.tenantId,
  persist: { ownerId: ctx.subject.id, visibility: 'private', tags: ['ai:origin:poster'] },
})

Changing prompt from '' to the real prompt in a workflow body keeps an already journaled enqueue when that same execution replays: activity identity uses the step name and attempt, not the recorded input or its prompt digest. The existing queue row keeps its original prompt and parameters; replay does not update them or submit another job. New executions use the corrected prompt. Keep both the workflow execution identity and the media step name unchanged. Changing either, or explicitly starting a new run, creates a new job identity.

A job whose worker was lost — the lease expired with the row still running — is not submitted again: the provider may have produced and charged for the artifact, and a built-in generator cannot be asked. The run is resumed with AiOffloadError whose outcome is 'unknown', and the row stays for an operator. Retrying is the app's explicit decision. Without the storage plugin a media job fails with the reason; it never resumes a run with bytes.

A provider with a task handle — `registerMediaTaskProvider`

A hosted media queue does not answer a submission with the artifact. It answers with a task handle and is asked for the result later. mediaStep reaches such a provider through a registration — the framework ships the contract, not an adapter for any one vendor:

import { Effect } from 'effect'
import { registerMediaTaskProvider, aiGenerationError } from '@voltro/ai'

registerMediaTaskProvider('hosted-queue', {
  submit: (request) =>
    Effect.tryPromise({
      try: () => queue.submit({ model: request.model, input: { prompt: request.prompt, ...request.params }, idempotencyKey: request.idempotencyKey }),
      catch: (cause) => aiGenerationError(cause, { prefix: 'hosted-queue' }),
    }).pipe(Effect.map((accepted) => ({ taskId: accepted.id }))),
  poll: (handle) =>
    Effect.tryPromise({
      try: () => queue.status(String(handle['taskId'])),
      catch: (cause) => aiGenerationError(cause, { prefix: 'hosted-queue' }),
    }).pipe(Effect.map((task) =>
      task.state === 'done' ? { status: 'succeeded', artifacts: [{ url: task.outputUrl, mediaType: 'video/mp4' }], usage: { unit: 'seconds', output: task.seconds } }
      : task.state === 'error' ? { status: 'failed', message: task.error }
      : { status: 'pending' })),
})

// Then, in a workflow body:
const { artifacts } = yield* mediaStep({
  name: 'render-clip',
  modality: 'video',
  prompt: `A clip for ${title}`,
  provider: { name: 'hosted-queue', model: 'clip-v2' },
  store: ctx.store,
})

submit returns the provider's handle as soon as it has one. The dispatcher writes the handle onto the queue row before anything else, then asks poll once and releases the row — the lease is never held across a job that may run for minutes. Every later claim, on any replica, finds the handle and polls again (every taskPollMs, default five seconds, or the provider's own retryAfterMs when longer) until the provider reports succeeded or failed. Nothing has to write to the queue for that next poll to happen: when the row comes due, the dispatcher is already armed for it — including on a deployment where the dispatcher otherwise stops ticking between arrivals (Postgres change data capture, or a broadcast broker). The same holds for a retry waiting out its backoff and for a row whose worker was lost, which is reclaimed the moment its lease runs out. Artifacts go through the same mediaPersist path under the same <row>#<attempt>#<index> identity; usage is recorded in the provider's unit under the same row and attempt.

submit receives an idempotencyKey derived from the queue row and the attempt. Forward it to a provider that de-duplicates on one, and say so with idempotentSubmit: true on the registration.

The lost-worker case changes shape for a task provider. A row reclaimed with a handle is polled — never submitted again and never reported unknown; the run resumes with the result of the task the lost worker had submitted. What remains is the window between the provider accepting the submission and the handle reaching the row: a row reclaimed with no handle still resumes its run with outcome: 'unknown', exactly as a built-in generator does. A registration with idempotentSubmit: true closes that window too — the reclaim submits again under the same key and receives the same task.

A failed status is terminal by default: the provider ran the job and failed it, and a fresh submission is a new charge. The run resumes with AiOffloadError carrying the provider's message. retryable: true on the status re-submits under the next attempt with the dispatcher's backoff, under a fresh key. A task still pending after taskTimeoutMs (default one hour, the same hour a parked run waits by default) is failed with the reason.

Recovery does not authorize another paid task

A transient polling or artifact-persistence error retains the confirmed task handle, attempt number and artifact operation IDs. The dispatcher retries that same task; only an explicit terminal provider status with retryable: true permits a new submission attempt. A failed submission without a durably stored handle has an unknown outcome, not permission to submit again. Providers that guarantee idempotentSubmit can recover using the same submission key; the task deadline also bounds that recovery.

Both successful results and terminal errors are journaled before the workflow is resumed. If delivery fails, a later dispatcher sends the saved signal without calling the provider or persisting the artifacts again. The inference stays running, with completedAt: null, until delivery succeeds; it is not eligible for completed-inference retention. A crash after delivery can replay the signal, so the workflow resume operation must remain idempotent.

This does not make provider acceptance and database persistence atomic. If the provider accepted a paid job but no handle reached durable storage, recovery still reports unknown unless the provider guarantees idempotent submission. Do not manually reset that row to pending without reconciling the provider job.

On the client — `useTranscription`

The entrance is a file, so the two halves are one hook:

const t = useTranscription('app', { action: 'media.transcribe' })

await t.transcribe(file)
t.phase       // 'idle' | 'uploading' | 'transcribing' | 'done' | 'error'
t.progress    // 0..1 while uploading, undefined once the model is running
t.result      // { text, language, durationSec, segments? }

progress is scoped to the phase on purpose. A single number across both reaches 1 when the upload finishes, while the model has not started — a bar that says finished is worse than no bar, so the transcribing phase reports undefined and an indeterminate spinner is the honest control.

The hook opens no transport: the upload is useUpload's and the transcription is an action you declare, guard and point at a model. It owns the reduction, which is the part every app was rewriting.

Realtime voice and text

defineRealtimeSession now relays a server-owned voice/text session through an authenticated gateway route. resolveRealtimeModel supports Gateway and OpenAI provider configuration. canExecute({ modality: 'realtime' }) is true. This is a dedicated gateway socket, not the existing RPC subscription connection; session credentials stay server-side. Tool execution, budget reservation and automatic replay are not provided.

Media modalities

#Streaming raw audio with streamTranscribe

#Through a declared RPC stream

#transcribe — audio to text

#The source is a reference, not bytes

#Timestamps and speakers

#generateSpeech — text to audio

#A provider the gateway does not carry

#Taking over a built-in name

#Cost — speech is not billed in tokens

#Image references and grounding

#Video references with roles

#Media seeds

#Cancellation

#Inline video limits

#The gateway's response budget

#As a durable step — transcribeStep

#A media job as a durable step — mediaStep

#A provider with a task handle — registerMediaTaskProvider

#Recovery does not authorize another paid task

#On the client — useTranscription

#Realtime voice and text