Function runProcessorWithChunks

packages/ai/tests/helpers/processor-harness.ts:63–103

(
  chunks: Array<StreamChunk>,
)

 * @returns The last assistant UIMessage from the processor after the stream ends.
 */
export async function runProcessorWithChunks(
  chunks: Array<StreamChunk>,
): Promise<UIMessage> {
  const processor = new StreamProcessor()

  const envelopeChunks: Array<StreamChunk> = [
    runStartedChunk(),
    textMessageStartChunk(),
    ...chunks,
    {
      type: EventType.TEXT_MESSAGE_END,
      timestamp: Date.now(),
      messageId: 'msg-1',
    },
    {
      type: EventType.RUN_FINISHED,
      timestamp: Date.now(),
      runId: 'run-1',
      threadId: 'thread-1',
      finishReason: 'stop',
    },
  ]

  async function* streamOf(cs: Array<StreamChunk>): AsyncIterable<StreamChunk> {
    for (const c of cs) {
      yield c
    }
  }

  await processor.process(streamOf(envelopeChunks))

  const messages = processor.getMessages()
  const assistant = messages.findLast((m) => m.role === 'assistant')
  if (!assistant) {
    throw new Error(
      'runProcessorWithChunks: no assistant message produced. Messages: ' +
        JSON.stringify(messages),
    )
  }
  return assistant
}