Chat - Core streaming

Understand the stream chunk protocol that turns adapter responses into normalized message state, including text, reasoning, tool, source, file, and data chunks.

The adapter's sendMessage() method returns a ReadableStream<ChatMessageChunk | ChatStreamEnvelope>. The runtime reads this stream, processes each chunk, and updates the normalized store. Your UI components see the updates through hooks and selectors.

The following demo visualizes the streaming lifecycle:

Chunk categories

Lifecycle chunks

These chunks mark the boundaries of a response:

Every stream must begin with start and end with either finish or abort.

Text chunks

Text chunks create a ChatTextMessagePart in the message. Multiple text-delta chunks are batched according to streamFlushInterval before being applied to the store.

Reasoning chunks

Reasoning chunks create a ChatReasoningMessagePart — useful for chain-of-thought or thinking trace displays.

Tool chunks

Tool chunks create ChatToolMessagePart or ChatDynamicToolMessagePart entries. The toolInvocation.state tracks the tool lifecycle: 'input-streaming' → 'input-available' → 'approval-requested' → 'output-available' (or 'output-error' / 'output-denied').

Source chunks

These create ChatSourceUrlMessagePart or ChatSourceDocumentMessagePart entries.

File chunks

Creates a ChatFileMessagePart.

Data chunks

Data chunks use a type that matches the data-* pattern (for example, data-citations or data-summary). They create ChatDataMessagePart entries and also fire the onData callback. If transient is true, the part is not persisted in the message.

Step boundary chunks

Step chunks create ChatStepStartMessagePart entries, useful for showing multi-step processing in the UI.

Metadata chunks

Stream envelopes

For deduplication and ordering, wrap chunks in a ChatStreamEnvelope:

interface ChatStreamEnvelope {
  eventId?: string; // unique event identifier for deduplication
  sequence?: number; // monotonic ordering number
  chunk: ChatMessageChunk;
}

The runtime accepts both raw ChatMessageChunk objects and enveloped chunks in the same stream. Envelopes are useful for SSE-based transports where chunks might arrive out of order or be replayed.

streamFlushInterval

Rapid text and reasoning deltas are batched before being applied to the store. The streamFlushInterval prop on ChatProvider controls the batching window (default: 16ms).

<ChatProvider adapter={adapter} streamFlushInterval={32}>
  <MyChat />
</ChatProvider>

How chunks become parts

The stream processor maps chunks to ChatMessagePart entries:

1. text-start creates a new ChatTextMessagePart with state: 'streaming'.
2. text-delta appends to the existing text part.
3. text-end sets state: 'done'.
4. Tool chunks follow a similar lifecycle, updating toolInvocation.state through each phase.
5. Source, file, and data chunks each create their corresponding part type immediately.
6. start-step creates a ChatStepStartMessagePart separator.

The message's status field also updates through the stream:

• 'sending' → set when the user message is optimistically added
• 'streaming' → set when start arrives
• 'sent' → set when finish arrives
• 'cancelled' → set when abort arrives
• 'error' → set when the stream fails

Error and disconnect handling

If the stream closes without a terminal chunk (finish or abort), the runtime:

1. Records a recoverable stream error.
2. Sets the message status to 'error'.
3. Calls onError and onFinish with isDisconnect: true.
4. If reconnectToStream() is implemented, attempts to resume.

If the adapter's sendMessage() throws, the runtime records a send error and surfaces it through the error model.

Building a stream

When writing an adapter, you can construct a ReadableStream from an array of chunks:

function createStream(
  chunks: ChatMessageChunk[],
  delayMs = 0,
): ReadableStream<ChatMessageChunk> {
  return new ReadableStream({
    async start(controller) {
      for (const chunk of chunks) {
        if (delayMs > 0) {
          await new Promise((r) => setTimeout(r, delayMs));
        }
        controller.enqueue(chunk);
      }
      controller.close();
    },
  });
}

Or convert a server-sent event stream into chunks:

async function fromSSE(
  url: string,
  signal: AbortSignal,
): Promise<ReadableStream<ChatMessageChunk>> {
  const response = await fetch(url, { signal });
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();

  return new ReadableStream({
    async pull(controller) {
      const { done, value } = await reader.read();
      if (done) {
        controller.close();
        return;
      }
      const chunk = JSON.parse(decoder.decode(value));
      controller.enqueue(chunk);
    },
  });
}

See also

• Adapters for the adapter interface that produces streams.
• State and store for streamFlushInterval and error model.
• Streaming lifecycle for send, stop, retry, and callbacks in action.
• Message parts for rendering the parts that chunks produce.

API

• ChatRoot