Streaming

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',
  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), message, error (carries retryable), and done. The stream NEVER fails — an upstream error arrives as a terminal error event.

#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), or pair it with the resumable-stream store's markDone so other nodes stop tailing. Single-node dev/self-host needs nothing extra.

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 (streamEventsTable + streamStateTable, register them in your database/index.ts) 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.

#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:

1. A send action appends the user message.
2. The action calls runAssistant(...).
3. runAssistant inserts one agent_messages row with streaming: true.
4. Token/tool deltas patch that row.
5. 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.