feat (ai/core): add terminateOnError option to readUIMessageStream (#7503)

## Background

Generator use of read ui message stream requires early termination in
case of errors.

## Summary

Add `onError` and `terminateOnError` options to `readUIMesageStream`

## Related Issues

Fixes #7420
Continues #7500

---------

Co-authored-by: remorses <beats.by.morse@gmail.com>

Author: lgrammel

.changeset/nasty-cows-care.md

@@ -0,0 +1,5 @@
+---
+'ai': patch
+---
+
+feat (ai/core): add terminateOnError option to readUIMessageStream

content/docs/07-reference/02-ai-sdk-ui/43-read-ui-message-stream.mdx

@@ -21,17 +21,31 @@ import { readUIMessageStream } from 'ai';
 
 <PropertiesTable
   content={[
+    {
+      name: 'message',
+      type: 'UIMessage',
+      isOptional: true,
+      description:
+        'The last assistant message to use as a starting point when the conversation is resumed. Otherwise undefined.',
+    },
     {
       name: 'stream',
       type: 'ReadableStream<UIMessageChunk>',
       description: 'The stream of UIMessageChunk objects to read.',
     },
     {
-      name: 'message',
-      type: 'UIMessage',
+      name: 'onError',
+      type: '(error: unknown) => void',
       isOptional: true,
       description:
-        'The last assistant message to use as a starting point when the conversation is resumed. Otherwise undefined.',
+        'A function that is called when an error occurs during stream processing.',
+    },
+    {
+      name: 'terminateOnError',
+      type: 'boolean',
+      isOptional: true,
+      description:
+        'Whether to terminate the stream if an error occurs. Defaults to false.',
     },
   ]}
 />

packages/ai/src/ui-message-stream/read-ui-message-stream.test.ts

@@ -100,4 +100,22 @@ describe('readUIMessageStream', () => {
         ]
       `);
   });
+
+  it('should throw an error when encountering an error UI stream part', async () => {
+    const stream = createUIMessageStream([
+      { type: 'start', messageId: 'msg-123' },
+      { type: 'text-start', id: 'text-1' },
+      { type: 'text-delta', id: 'text-1', delta: 'Hello' },
+      { type: 'error', errorText: 'Test error message' },
+    ]);
+
+    const uiMessages = readUIMessageStream({
+      stream,
+      terminateOnError: true,
+    });
+
+    await expect(convertAsyncIterableToArray(uiMessages)).rejects.toThrow(
+      'Test error message',
+    );
+  });
 });

packages/ai/src/ui-message-stream/read-ui-message-stream.ts

@@ -16,18 +16,25 @@ import { consumeStream } from '../util/consume-stream';
  *
  * @param options.message - The last assistant message to use as a starting point when the conversation is resumed. Otherwise undefined.
  * @param options.stream - The stream of `UIMessageChunk`s to read.
+ * @param options.terminateOnError - Whether to terminate the stream if an error occurs.
+ * @param options.onError - A function that is called when an error occurs.
  *
  * @returns An `AsyncIterableStream` of `UIMessage`s. Each stream part is a different state of the same message
  * as it is being completed.
  */
 export function readUIMessageStream<UI_MESSAGE extends UIMessage>({
   message,
   stream,
+  onError,
+  terminateOnError = false,
 }: {
   message?: UI_MESSAGE;
   stream: ReadableStream<UIMessageChunk>;
+  onError?: (error: unknown) => void;
+  terminateOnError?: boolean;
 }): AsyncIterableStream<UI_MESSAGE> {
   let controller: ReadableStreamDefaultController<UI_MESSAGE> | undefined;
+  let hasErrored = false;
 
   const outputStream = new ReadableStream<UI_MESSAGE>({
     start(controllerParam) {
@@ -40,6 +47,15 @@ export function readUIMessageStream<UI_MESSAGE extends UIMessage>({
     lastMessage: message,
   });
 
+  const handleError = (error: unknown) => {
+    onError?.(error);
+
+    if (!hasErrored && terminateOnError) {
+      hasErrored = true;
+      controller?.error(error);
+    }
+  };
+
   consumeStream({
     stream: processUIMessageStream({
       stream,
@@ -56,12 +72,15 @@ export function readUIMessageStream<UI_MESSAGE extends UIMessage>({
           },
         });
       },
-      onError: error => {
-        throw error;
-      },
+      onError: handleError,
     }),
+    onError: handleError,
   }).finally(() => {
-    controller?.close();
+    // Only close if no error occurred. Calling close() on an errored controller
+    // throws "Invalid state: Controller is already closed" TypeError.
+    if (!hasErrored) {
+      controller?.close();
+    }
   });
 
   return createAsyncIterableStream(outputStream);

packages/ai/src/ui/process-ui-message-stream.ts

@@ -537,7 +537,3 @@ export function processUIMessageStream<UI_MESSAGE extends UIMessage>({
     }),
   );
 }
-
-function isObject(value: unknown): value is object {
-  return typeof value === 'object' && value !== null;
-}