VoltroDocumentation

Streams

`*.stream.ts` + `*.stream.server.ts` pairs — one-shot server-to-client element streams with defineStream and useAgentStream.

A stream is a one-shot server-to-client feed. It emits plain elements, in order, and then finishes. It is not reactive, does not have snapshots or patches, and does not participate in auto-optimistic cache updates.

Use streams for transient work: LLM token events, progress updates, import logs, provider events, or any server-side operation where the client should see elements as they arrive.

Live — click Start and { n } elements arrive one at a time (150ms apart), then the stream finishes (one-shot push, no snapshot/delta):

const s = useAgentStream('app', 'progress.count')
s.start({ to: 20 })   // s.events appends as elements arrive

Minimal stream pair

Descriptor:

// apps/api/streams/ticker.stream.ts
import { defineStream } from '@voltro/protocol'
import { Schema } from 'effect'

export const ticker = defineStream({
  name:    'ticker.watch',
  input:   Schema.Struct({ symbol: Schema.String }),
  element: Schema.Struct({
    price: Schema.Number,
    at:    Schema.Date,
  }),
})

Server executor:

// apps/api/streams/ticker.stream.server.ts
import { Stream } from 'effect'

export default (input: { symbol: string }) =>
  Stream.fromAsyncIterable(
    externalTickerStream(input.symbol),
    (error) => error,
  )

The executor can return Stream, Effect<Stream>, or a Promise<Stream>. The runtime binds it with bindStream and interrupts it when the client cancels or disconnects.

Client hook

useAgentStream consumes any defineStream RPC, not only AI agents:

import { useAgentStream } from '@voltro/client'

const ticker = useAgentStream<{ price: number; at: Date }>('app', 'ticker.watch')

return (
  <>
    <button onClick={() => ticker.start({ symbol: 'BTC' })}>Start</button>
    <button onClick={ticker.cancel}>Cancel</button>
    <ul>
      {ticker.events.map((event, index) => (
        <li key={index}>{event.price}</li>
      ))}
    </ul>
  </>
)

State shape:

Field Meaning
events Elements received so far, in order.
status 'idle', 'streaming', 'done', or 'error'.
error Last stream failure when status === 'error'.
start(input?) Starts a new stream and clears previous events.
cancel() Interrupts the in-flight stream.

Stream vs query

Query Stream
Files *.query.ts + *.query.server.ts *.stream.ts + *.stream.server.ts
Descriptor field output element
Client hook useSubscription useAgentStream
Wire element Subscription event envelope Plain element
Reactivity Re-runs on matching writes No automatic reactivity
Persistence Usually backed by database state Transient unless you persist yourself
Best for Live data views Token/progress/event feeds

If users should be able to reload the page and still see the result, persist rows and expose them through a query. If the stream is just live progress for the current run, keep it as a stream.

AI streams

Transient agent runs are just streams whose elements are token/tool events. useAgent is an ergonomic wrapper over useAgentStream; durable chat uses a different pattern: an action writes/patches agent_messages, and a query streams those persisted rows.

See AI streaming and Agents for those patterns.

Anti-patterns

  • Using streams as a database subscription replacement. Queries already do this and handle reconnect snapshots.
  • Streaming data that must be durable. Put it in a table and subscribe to a query.
  • Forgetting cancellation. Long streams should release upstream resources when interrupted.

See also