Use a readable stream directly

Author: ericallam

packages/core/src/v3/realtimeStreams/manager.ts

@@ -1,5 +1,5 @@
 import { ApiClient } from "../apiClient/index.js";
-import { ensureAsyncIterable } from "../streams/asyncIterableStream.js";
+import { ensureAsyncIterable, ensureReadableStream } from "../streams/asyncIterableStream.js";
 import { taskContext } from "../task-context-api.js";
 import { StreamInstance } from "./streamInstance.js";
 import {
@@ -30,7 +30,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
     options?: RealtimeStreamOperationOptions
   ): RealtimeStreamInstance<T> {
     // Normalize ReadableStream to AsyncIterable
-    const asyncIterableSource = ensureAsyncIterable(source);
+    const readableStreamSource = ensureReadableStream(source);
 
     const runId = getRunIdForOptions(options);
 
@@ -52,7 +52,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
       baseUrl: this.baseUrl,
       runId,
       key,
-      source: asyncIterableSource,
+      source: readableStreamSource,
       signal: combinedSignal,
       requestOptions: options?.requestOptions,
       target: options?.target,

packages/core/src/v3/realtimeStreams/streamInstance.ts

@@ -10,7 +10,7 @@ export type StreamInstanceOptions<T> = {
   baseUrl: string;
   runId: string;
   key: string;
-  source: AsyncIterable<T>;
+  source: ReadableStream<T>;
   signal?: AbortSignal;
   requestOptions?: AnyZodFetchOptions;
   target?: "self" | "parent" | "root" | string;

packages/core/src/v3/realtimeStreams/streamsWriterV1.ts

@@ -8,7 +8,7 @@ export type StreamsWriterV1Options<T> = {
   baseUrl: string;
   runId: string;
   key: string;
-  source: AsyncIterable<T>;
+  source: ReadableStream<T>;
   headers?: Record<string, string>;
   signal?: AbortSignal;
   version?: string;
@@ -43,7 +43,7 @@ export class StreamsWriterV1<T> implements StreamsWriter {
   private streamComplete = false;
 
   constructor(private options: StreamsWriterV1Options<T>) {
-    const [serverStream, consumerStream] = this.createTeeStreams();
+    const [serverStream, consumerStream] = this.options.source.tee();
     this.serverStream = serverStream;
     this.consumerStream = consumerStream;
     this.maxRetries = options.maxRetries ?? 10;
@@ -60,23 +60,6 @@ export class StreamsWriterV1<T> implements StreamsWriter {
     return randomBytes(4).toString("hex");
   }
 
-  private createTeeStreams() {
-    const readableSource = new ReadableStream<T>({
-      start: async (controller) => {
-        try {
-          for await (const value of this.options.source) {
-            controller.enqueue(value);
-          }
-          controller.close();
-        } catch (error) {
-          controller.error(error);
-        }
-      },
-    });
-
-    return readableSource.tee();
-  }
-
   private startBuffering(): void {
     this.streamReader = this.serverStream.getReader();
 
@@ -131,10 +114,10 @@ export class StreamsWriterV1<T> implements StreamsWriter {
         if (this.isRetryableError(error)) {
           if (this.retryCount < this.maxRetries) {
             this.retryCount++;
-            
+
             // Clean up the current request to avoid socket leaks
             req.destroy();
-            
+
             const delayMs = this.calculateBackoffDelay();
 
             await this.delay(delayMs);
@@ -157,10 +140,10 @@ export class StreamsWriterV1<T> implements StreamsWriter {
         // Timeout is retryable
         if (this.retryCount < this.maxRetries) {
           this.retryCount++;
-          
+
           // Clean up the current request to avoid socket leaks
           req.destroy();
-          
+
           const delayMs = this.calculateBackoffDelay();
 
           await this.delay(delayMs);
@@ -182,13 +165,13 @@ export class StreamsWriterV1<T> implements StreamsWriter {
         if (res.statusCode && this.isRetryableStatusCode(res.statusCode)) {
           if (this.retryCount < this.maxRetries) {
             this.retryCount++;
-            
+
             // Drain and destroy the response and request to avoid socket leaks
             // We need to consume the response before destroying it
             res.resume(); // Start draining the response
             res.destroy(); // Destroy the response to free the socket
             req.destroy(); // Destroy the request as well
-            
+
             const delayMs = this.calculateBackoffDelay();
 
             await this.delay(delayMs);
@@ -391,7 +374,7 @@ export class StreamsWriterV1<T> implements StreamsWriter {
         if (this.isRetryableError(error) && attempt < maxHeadRetries) {
           // Clean up the current request to avoid socket leaks
           req.destroy();
-          
+
           await this.delay(1000 * (attempt + 1)); // Simple linear backoff
           const result = await this.queryServerLastChunkIndex(attempt + 1);
           resolve(result);
@@ -424,7 +407,7 @@ export class StreamsWriterV1<T> implements StreamsWriter {
             res.resume();
             res.destroy();
             req.destroy();
-            
+
             await this.delay(1000 * (attempt + 1));
             const result = await this.queryServerLastChunkIndex(attempt + 1);
             resolve(result);

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -5,7 +5,7 @@ export type StreamsWriterV2Options<T = any> = {
   basin: string;
   stream: string;
   accessToken: string;
-  source: AsyncIterable<T>;
+  source: ReadableStream<T>;
   signal?: AbortSignal;
   flushIntervalMs?: number; // Used as lingerDuration for BatchTransform (default 200ms)
   maxRetries?: number; // Not used with appendSession, kept for compatibility
@@ -81,7 +81,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       });
     }
 
-    const [serverStream, consumerStream] = this.createTeeStreams();
+    const [serverStream, consumerStream] = this.options.source.tee();
     this.serverStream = serverStream;
     this.consumerStream = consumerStream;
 
@@ -111,31 +111,6 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
     this.log("[S2MetadataStream] Abort cleanup complete");
   }
 
-  private createTeeStreams() {
-    const readableSource = new ReadableStream<T>({
-      start: async (controller) => {
-        try {
-          let count = 0;
-
-          for await (const value of this.options.source) {
-            if (this.aborted) {
-              controller.error(new Error("Stream aborted"));
-              return;
-            }
-            controller.enqueue(value);
-            count++;
-          }
-
-          controller.close();
-        } catch (error) {
-          controller.error(error);
-        }
-      },
-    });
-
-    return readableSource.tee();
-  }
-
   private async initializeServerStream(): Promise<void> {
     try {
       if (this.aborted) {
@@ -171,7 +146,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
         )
         .pipeThrough(
           new BatchTransform({
-            lingerDuration: this.flushIntervalMs,
+            lingerDurationMillis: this.flushIntervalMs,
           })
         )
         .pipeTo(session.writable);

packages/core/src/v3/streams/asyncIterableStream.ts

@@ -133,3 +133,26 @@ export function ensureAsyncIterable<T>(
     },
   };
 }
+
+export function ensureReadableStream<T>(
+  input: AsyncIterable<T> | ReadableStream<T>
+): ReadableStream<T> {
+  if ("getReader" in input) {
+    return input as ReadableStream<T>;
+  }
+
+  return new ReadableStream<T>({
+    async start(controller) {
+      const iterator = input[Symbol.asyncIterator]();
+
+      while (true) {
+        const { done, value } = await iterator.next();
+        if (done) {
+          break;
+        }
+        controller.enqueue(value);
+      }
+      controller.close();
+    },
+  });
+}

pnpm-lock.yaml

@@ -18,7 +18,8 @@
     "react": "19.1.0",
     "react-dom": "19.1.0",
     "shiki": "^3.13.0",
-    "streamdown": "^1.4.0"
+    "streamdown": "^1.4.0",
+    "zod": "3.25.76"
   },
   "devDependencies": {
     "@tailwindcss/postcss": "^4",

references/realtime-streams/package.json

@@ -1,7 +1,8 @@
 import { aiStream } from "@/app/streams";
 import { openai } from "@ai-sdk/openai";
 import { logger, streams, task } from "@trigger.dev/sdk";
-import { convertToModelMessages, streamText, UIMessage } from "ai";
+import { convertToModelMessages, readUIMessageStream, streamText, tool, UIMessage } from "ai";
+import { z } from "zod/v4";
 
 export type AIChatPayload = {
   messages: UIMessage[];
@@ -19,13 +20,43 @@ export const aiChatTask = task({
       model: openai("gpt-4o"),
       system: "You are a helpful assistant.",
       messages: convertToModelMessages(payload.messages),
+      tools: {
+        getCommonUseCases: tool({
+          description: "Get common use cases",
+          inputSchema: z.object({
+            useCase: z.string().describe("The use case to get common use cases for"),
+          }),
+          execute: async ({ useCase }) => {
+            return {
+              useCase,
+              commonUseCases: [
+                "Streaming data to a client",
+                "Streaming data to a server",
+                "Streaming data to a database",
+                "Streaming data to a file",
+                "Streaming data to a socket",
+                "Streaming data to a queue",
+                "Streaming data to a message broker",
+                "Streaming data to a message queue",
+                "Streaming data to a message broker",
+              ],
+            };
+          },
+        }),
+      },
     });
 
     // Get the UI message stream
     const uiMessageStream = result.toUIMessageStream();
 
     // Append the stream to metadata
-    const { waitUntilComplete } = aiStream.pipe(uiMessageStream);
+    const { waitUntilComplete, stream } = aiStream.pipe(uiMessageStream);
+
+    for await (const uiMessage of readUIMessageStream({
+      stream: stream,
+    })) {
+      logger.log("Current message state", { uiMessage });
+    }
 
     // Wait for the stream to complete
     await waitUntilComplete();