Defined in: activities/chat/stream/processor.ts:114

StreamProcessor - State machine for processing AI response streams

Manages the full UIMessage[] conversation and emits events on changes.

State tracking:

• Full message array
• Current assistant message being streamed
• Text content accumulation
• Multiple parallel tool calls
• Tool call completion detection

Tool call completion is detected when:

1. A new tool call starts at a different index
2. Text content arrives
3. Stream ends

new StreamProcessor(options): StreamProcessor;

Defined in: activities/chat/stream/processor.ts:142

StreamProcessorOptions = {}

StreamProcessor

addToolApprovalResponse(approvalId, approved): void;

Defined in: activities/chat/stream/processor.ts:255

Add an approval response (called by client after handling onApprovalRequest)

string

boolean

void

addToolResult(
   toolCallId, 
   output, 
   error?): void;

Defined in: activities/chat/stream/processor.ts:211

Add a tool result (called by client after handling onToolCall)

string

any

string

void

addUserMessage(content): UIMessage;

Defined in: activities/chat/stream/processor.ts:169

Add a user message to the conversation

string

UIMessage

areAllToolsComplete(): boolean;

Defined in: activities/chat/stream/processor.ts:286

Check if all tool calls in the last assistant message are complete Useful for auto-continue logic

boolean

clearMessages(): void;

Defined in: activities/chat/stream/processor.ts:318

Clear all messages

void

finalizeStream(): void;

Defined in: activities/chat/stream/processor.ts:814

Finalize the stream - complete all pending operations

void

getMessages(): UIMessage[];

Defined in: activities/chat/stream/processor.ts:278

Get current messages

UIMessage[]

getRecording(): ChunkRecording | null;

Defined in: activities/chat/stream/processor.ts:893

Get the current recording

ChunkRecording | null

getState(): ProcessorState;

Defined in: activities/chat/stream/processor.ts:866

Get current processor state

ProcessorState

process(stream): Promise<ProcessorResult>;

Defined in: activities/chat/stream/processor.ts:331

Process a stream and emit events through handlers

AsyncIterable<any>

Promise<ProcessorResult>

processChunk(chunk): void;

Defined in: activities/chat/stream/processor.ts:359

Process a single chunk from the stream

AGUIEvent

void

removeMessagesAfter(index): void;

Defined in: activities/chat/stream/processor.ts:310

Remove messages after a certain index (for reload/retry)

number

void

reset(): void;

Defined in: activities/chat/stream/processor.ts:916

Full reset (including messages)

void

setMessages(messages): void;

Defined in: activities/chat/stream/processor.ts:161

Set the messages array (e.g., from persisted state)

UIMessage[]

void

startAssistantMessage(): string;

Defined in: activities/chat/stream/processor.ts:187

Start streaming a new assistant message Returns the message ID

string

startRecording(): void;

Defined in: activities/chat/stream/processor.ts:880

Start recording chunks

void

toModelMessages(): ModelMessage<
  | string
  | ContentPart<unknown, unknown, unknown, unknown, unknown>[]
  | null>[];

Defined in: activities/chat/stream/processor.ts:267

Get the conversation as ModelMessages (for sending to LLM)

ModelMessage< | string | ContentPart<unknown, unknown, unknown, unknown, unknown>[] | null>[]

static replay(recording, options?): Promise<ProcessorResult>;

Defined in: activities/chat/stream/processor.ts:925

Replay a recording through the processor

ChunkRecording

StreamProcessorOptions

Promise<ProcessorResult>