Streaming

AI token streams with defineStream, useAgentStream, and streamText — plus cancel, retry, and resumable streams.

AI streaming in Voltro uses the same stream primitive as any other one-shot server-to-client feed:

*.stream.ts declares the defineStream descriptor.
*.stream.server.ts returns an Effect Stream.
useAgentStream consumes the stream on the client.

For durable chat that survives reloads, do not stream raw tokens to React state. Persist token deltas into agent_messages and expose them through a reactive query.

Transient Stream

Descriptor:

// apps/api/streams/support.run.stream.ts
import { AgentEvent } from '@voltro/ai/events'   // browser-safe Schema entry — NOT '@voltro/ai' (server-only)
import { defineStream } from '@voltro/protocol'
import { Schema } from 'effect'

export const supportRun = defineStream({
  name:    'support.run',
  // Every call spends provider tokens, so this is metered work — guard it
  // rather than declaring it open, and rate-limit it before it faces the
  // internet (`@voltro/plugin-ratelimit`).
  guards:  [{ scope: 'support:chat' }],
  input:   Schema.Struct({ message: Schema.String }),
  element: AgentEvent,
})

Server executor:

// apps/api/streams/support.run.stream.server.ts
import { streamText } from '@voltro/ai'

export default (input: { message: string }) =>
  streamText({
    system: 'You are concise and helpful.',
    prompt: input.message,
  })

streamText emits AgentEvent elements. The _tags are: token (text delta), reasoning (the model's thinking, separate from the answer), toolCall, toolResult, source (a cited RAG/web source — url or document variant), file (an inline file the model produced, usually an image: mediaType + base64 data), stepMetadata (provider metadata for one completed model step), message, error (carries retryable), and done. The stream NEVER fails — an upstream error arrives as a terminal error event.

Grounding and attribution

Grounding stays with the provider response that supplied it. source and file events retain their own optional providerMetadata. A stepMetadata event carries { step, providerMetadata }: step is zero-based, counts all model steps (including those without metadata), and is local to this stream. The event is emitted when that step finishes, before the terminal done. Tool loops retain each step separately rather than exposing only the last one. No metadata event is invented for a step whose provider returned none.

For example, Google grounding may appear under providerMetadata.google.groundingMetadata, with source chunks and segment attribution. Voltro preserves the payload, not a vendor-independent claim that the answer is correct. Configure the provider's grounding/search capability separately. Validate provider-specific structures before using them; segment offsets belong to that provider step, not automatically to concatenated text from the whole run. Treat URLs and HTML as untrusted; do not insert provider HTML directly. The framework does not copy raw HTTP headers or request/response bodies into these events.

Use useAgentStream<AgentEvent>(...).events or useResumableAgentStream<AgentEvent>(...).events for the complete event list:

import type { AgentEvent } from '@voltro/ai/events'

export const attributionOf = (events: ReadonlyArray<AgentEvent>) =>
  events.filter(event =>
    event._tag === 'stepMetadata' ||
    ((event._tag === 'source' || event._tag === 'file') &&
      event.providerMetadata !== undefined),
  )

Resumable streams journal these events with their ordinary seq cursor. For durable agent chats, runAssistant stores the same attribution in messages.parts (type: 'stepMetadata', 'source', or 'file'); it is not added to content or to the text sent back as conversation history. The text-only useAgent helper does not expose attribution. Update exhaustive event/part switches for stepMetadata; voltro update flags affected consumers.

Smooth Streaming

Provider chunks are transport artifacts: one provider may send a word at a time while another releases a whole paragraph in a burst. Opt into server-side semantic smoothing when the UI should receive predictable word or line chunks:

streamText({
  prompt: input.message,
  smooth: {
    chunking: 'word', // default; alternatively 'line'
    delayMs: 10,      // default; null rechunks without adding delay
  },
})

smooth is framework-owned and deliberately narrower than the AI SDK's experimental transform surface. It smooths both text and reasoning, flushes pending content before tool/source/file/error/done events, and shares the stream's abort signal. Omit it or set smooth: false to preserve provider-native chunks and latency.

With resumableStreamText, put smooth inside options. Smoothing happens before AgentEvent conversion and before the append-only journal, so a reconnect replays the exact same SeqEvent boundaries seen live:

resumableStreamText({
  streamId: input.streamId,
  store,
  options: {
    prompt: input.message,
    smooth: { chunking: 'word', delayMs: 10 },
  },
  fromSeq: input.fromSeq,
})

Smoothing changes only text/reasoning boundaries and timing. Content, usage, finish reason, tool ids, and cost accounting stay unchanged. The client should still batch rendering (for example, at most once per animation frame); server smoothing normalizes semantics, it does not replace UI backpressure.

Cancel, retry

A "Stop" button needs to end an in-flight run out of band. Pass a streamId and call cancelStream(streamId) from a separate action/mutation — it aborts the provider call (token spend stops) and the stream ends on a terminal, non-retryable cancelled error.

import { streamText, cancelStream } from '@voltro/ai'

// executor — register the run under a client-chosen id
streamText({ prompt: input.message, streamId: input.streamId })

// a separate stop.action.server.ts
export default (input: { streamId: string }) => ({ cancelled: cancelStream(input.streamId) })

The registry is in-process: cancelStream only aborts a stream running on the same node and returns false when the id isn't known locally. On a multi-replica deployment, route the cancel call to the node that owns the stream (sticky by streamId). Do not use the resumable store's markDone as remote cancellation: it neither interrupts the producer nor persists a cancellation event. An abort request also does not prove that a remote provider stopped billing.

streamText also accepts an external signal (AbortSignal) and SDK-level maxRetries (retries the provider HTTP call before any bytes stream). For recovering from an immediate provider error, streamTextWithRetry(options, { maxAttempts, backoffMs }) re-runs the stream — but ONLY while nothing has streamed yet, so it never duplicates tokens; once content flows, a later error is surfaced as-is.

Resumable Streams

A resumable stream survives a client disconnect: close the tab mid-answer, reopen, and the assistant keeps streaming from where it left off. resumableStreamText runs the model ONCE (the producer, as a daemon that outlives the connection) and persists every event to a store; every consumer replays past its cursor then tails to the end.

// support.run.stream.server.ts
import { resumableStreamText, memoryResumableStreamStore } from '@voltro/ai'

const store = memoryResumableStreamStore() // single node — see below for multi-node

export default (input: { message: string; streamId: string; fromSeq?: number }) =>
  resumableStreamText({
    streamId: input.streamId,
    store,
    options: { prompt: input.message },
    fromSeq: input.fromSeq, // a reconnect passes the last `seq` it rendered
  })

It returns a Stream<SeqEvent> — each element is { seq, event }. The descriptor declares element: SeqEvent (import the Schema from @voltro/ai/events, the browser-safe entry — the @voltro/ai root is server-only) and an optional fromSeq on its input.

On the client, useResumableAgentStream('app', 'support.run') does the rest: it unwraps each SeqEvent (so .events are the plain inner events), tracks the highest seq, and on a transport drop BEFORE the run's terminal event it auto-reconnects with fromSeq = the last seq it rendered (exponential backoff; the no-progress cap resets whenever a reconnect delivers a new event, so a long flaky stream survives any number of well-spaced drops). The server replays past the cursor, then continues — one seamless stream.

The hook's own surface is small:

import { useResumableAgentStream } from '@voltro/client'
import type { AgentEvent } from '@voltro/ai/events'

const run = useResumableAgentStream<AgentEvent>('app', 'support.run')

// `input` MUST carry the resume key — `fromSeq` is injected by the hook.
run.start({ streamId, message: prompt })
run.cancel()

Field	Meaning
`events`	The unwrapped inner events so far, in order and deduped by `seq`.
`status`	`'idle'`, `'streaming'`, `'reconnecting'`, `'done'`, or `'error'`.
`reconnects`	How many times the transport dropped and auto-reconnected this run.
`error`	Set when `status === 'error'`.
`start(input?)`	Begins a run, clearing previous events.
`cancel()`	Stops the run and any pending reconnect.

A third options argument tunes the reconnect policy: maxReconnects (default 6), backoffMs (400), maxBackoffMs (8000), and isTerminal — which defaults to treating an AgentEvent-shaped { _tag: 'done' | 'error' } as the end of the run. Override isTerminal when your element type signals completion some other way, or the hook will keep trying to resume a finished stream.

For multi-node deployments use dataStoreResumableStreamStore(ctx.store) — it persists to the framework's own database (_voltro_stream_events + _voltro_stream_state, created for you: the table assembly adds both for any app that depends on @voltro/ai, so voltro dev / voltro db apply / voltro migrate all plan them) and elects exactly ONE producer per streamId via an atomic claim, so only one node runs the model while every node's consumers tail the shared log. Sweep finished streams with gcResumableStreams(store, { olderThan }).

For the fastest path, redisResumableStreamStore(redis, { ttlSeconds }) backs the log with a Redis LIST (RPUSH/LRANGE) plus a SET … NX producer claim — TTL evicts finished/abandoned streams without a sweep. @voltro/ai takes no Redis dependency; you inject a tiny ResumableRedis client (five methods: setNx / rpush / lrange / set / exists) adapting ioredis / node-redis. All three backends satisfy the same ResumableStreamStore interface, so they swap without touching the producer/consumer code.

Producer failures and persistence

A resumable producer persists exactly one terminal done or error before marking its log complete. A thrown source factory, a producer defect or an EOF without a terminal produces error with code: 'internal'; interrupting the producer itself produces code: 'cancelled'. These generated errors carry generic messages, never the exception, provider response or headers, and are not automatically retryable. Disconnecting a consumer does not interrupt the producer. A reconnect replays the same terminal event from the log.

Even on the successful path, receiving the terminal event does not imply that store.isDone(streamId) is already true. Appending the event precedes the separate markDone write, so a consumer may finish in between. Use the terminal event as its completion signal; code that owns startResumableProducer may await that Effect when it also requires the done-flag write. A custom store must preserve this ordering: marking done first could let a reader stop before seeing the final event. Neither signal proves external workflow or application persistence finished.

Consumers also finish on a persisted terminal if the subsequent done-flag write failed. If the log cannot store the terminal at all, the producer cannot promise its delivery: startResumableProducer fails, and the detached wrapper logs ai: resumable stream finalization could not be persisted. It does not mark an unwritten terminal as complete or rerun a potentially billable model call. After an ambiguous append acknowledgement, sequence numbers can have gaps; resume cursors remain monotonic. If the unacknowledged append was a terminal, the producer reads that exact position back before writing another, so a stored success is not followed by an invented failure solely because its acknowledgement was lost.

A process kill cannot execute cleanup. Use idleTimeoutMs on the resumable consumer to bound waiting for an abandoned producer, and inspect the server log when a run never supplies a terminal. That timeout is not a success receipt or automatic takeover of the producer claim; durable retry orchestration remains a workflow concern.

Client

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

const support = useAgentStream<AgentEvent>('app', 'support.run')
const text = support.events
  .filter((event) => event._tag === 'token')
  .map((event) => event.text)
  .join('')

return (
  <>
    <button
      disabled={support.status === 'streaming'}
      onClick={() => support.start({ message: prompt })}
    >
      Send
    </button>
    <button onClick={support.cancel}>Cancel</button>
    <pre>{text}</pre>
  </>
)

useAgent is a convenience wrapper over useAgentStream for transient chat UIs. It derives tokens and history so you do not have to filter raw events yourself.

Durable Chat

A plain transient stream is request-scoped: a reload loses the in-flight text (use a resumable stream, above, if you only need reconnect-resume). For product chat you usually want the full DURABLE history too — persist the assistant turn and stream the persisted rows through a query:

A send action appends the user message.
The action calls runAssistant(...).
runAssistant inserts one agent_messages row with streaming: true.
Token/tool deltas patch that row.
A query with source: 'agent_messages' re-runs and updates the UI.

// actions/chat.send.action.server.ts
import { appendMessage, getMessages, runAssistant } from '@voltro/ai'
import { Effect } from 'effect'

export default (
  input: { threadId: string; text: string },
  ctx,
) =>
  Effect.gen(function* () {
    const existing = yield* getMessages(ctx.store, input.threadId)
    const order = existing.length

    yield* appendMessage(ctx.store, {
      threadId: input.threadId,
      role: 'user',
      content: input.text,
      order,
    })

    yield* runAssistant(ctx.store, {
      threadId: input.threadId,
      prompt: input.text,
      order: order + 1,
    })

    return { ok: true }
  })

// queries/chat.messages.query.ts declares source: 'agent_messages'
// queries/chat.messages.query.server.ts
import { getMessages } from '@voltro/ai'

export default (input: { threadId: string }, ctx) =>
  getMessages(ctx.store, input.threadId)

const { data: messages } = useSubscription('app', 'chat.messages', { threadId })
const send = useAction('app', 'chat.send')

await send.run({ threadId, text })

This pattern survives reloads, works across tabs, and can be driven by workflows.

Stream vs Persisted Query

Need	Use
Playground token stream	`defineStream` + `useAgentStream`
Cancelable one-shot generation	`streamText({ streamId })` + `cancelStream`
Reconnect-resume an in-flight stream	`resumableStreamText` + a stream store
Chat history after reload	Action + `agent_messages` query
Cross-tab live conversation	Action + `agent_messages` query
Crash/retry semantics	Workflow patches persisted rows

Anti-Patterns

Saving only the final text for chat. Persist deltas so the UI can show live progress and survive reloads.
Using useAgentStream for blocking calls. Use an action with generateText or generateObject.
Using streams as subscriptions. If the data is durable state, expose it through a query.

Streaming

#Transient Stream

#Grounding and attribution

#Smooth Streaming

#Cancel, retry

#Resumable Streams

#Producer failures and persistence

#Client

#Durable Chat

#Stream vs Persisted Query

#Anti-Patterns