feat: add stream helper (#98)

.changeset/wicked-ladybugs-stare.md

@@ -0,0 +1,40 @@
+---
+"@llama-flow/docs": patch
+"@llama-flow/core": minor
+---
+
+feat: add stream helper
+
+In this release, we built-in some stream helper (inspired from (TC39 Async Iterator Helpers)[https://github.com/tc39/proposal-async-iterator-helpers])
+
+- move `@llama-flow/core/stream/until` into `stream.until`
+- move `@llama-flow/core/stream/filter` into `stream.filter`
+- move `@llama-flow/core/stream/consumer` into `stream.toArray()`
+- add `stream.take(limit)
+- add `stream.toArray()`
+
+```diff
+- import { collect } from "@llama-flow/core/stream/consumer";
+- import { until } from "@llama-flow/core/stream/until";
+- import { filter } from "@llama-flow/core/stream/filter";
+
+-  const results = await collect(
+-    until(
+-      filter(
+-        stream,
+-        (ev) =>
+-          processedValidEvent.include(ev) || processedInvalidEvent.include(ev),
+-      ),
+-      () => {
+-        return results.length >= totalItems;
+-      },
+-    ),
+-  );
++  const results = await stream
++    .filter(
++      (ev) =>
++        processedValidEvent.include(ev) || processedInvalidEvent.include(ev),
++    )
++    .take(totalItems)
++    .toArray();
+```

README.md

@@ -67,9 +67,7 @@ const result = await pipeline(stream, async function (source) {
 });
 console.log(result); // stop received!
 // or
-import { until } from "@llama-flow/core/stream/until";
-import { collect } from "@llama-flow/core/stream/consumer";
-const allEvents = await collect(until(stream, stopEvent));
+const allEvents = await stream.until(stopEvent).toArray();
 ```
 
 ### Helper Functions for Common Tasks
@@ -102,10 +100,6 @@ By default, we provide a simple fan-out utility to run multiple workflows in par
 - `getContext().stream` will return a stream of events emitted by the sub-workflow
 
 ```ts
-import { until } from "@llama-flow/core/stream/until";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { filter } from "@llama-flow/core/stream/filter";
-
 let condition = false;
 workflow.handle([startEvent], async (start) => {
   const { sendEvent, stream } = getContext();
@@ -113,12 +107,10 @@ workflow.handle([startEvent], async (start) => {
     sendEvent(convertEvent.with(i));
   }
   // You define the condition to stop the workflow
-  const results = await collect(
-    filter(
-      until(stream, () => condition),
-      (ev) => convertStopEvent.includes(ev),
-    ),
-  );
+  const results = await stream
+    .until(() => condition)
+    .filter(convertStopEvent)
+    .toArray();
   console.log(results.length); // 10
   return stopEvent.with();
 });

demo/node/basic.ts

@@ -1,8 +1,6 @@
 import { createWorkflow, workflowEvent, getContext } from "@llama-flow/core";
 import { pipeline } from "node:stream/promises";
 import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
-import { filter } from "@llama-flow/core/stream/filter";
 
 //#region define workflow events
 const startEvent = workflowEvent<string>();
@@ -23,16 +21,7 @@ workflow.handle([startEvent], async () => {
   sendEvent(branchBEvent.with("Branch B"));
   sendEvent(branchCEvent.with("Branch C"));
 
-  let condition = 0;
-  const results = await collect(
-    until(
-      filter(stream, (ev) => branchCompleteEvent.include(ev)),
-      () => {
-        condition++;
-        return condition === 3;
-      },
-    ),
-  );
+  const results = await stream.filter(branchCompleteEvent).take(3).toArray();
 
   return allCompleteEvent.with(results.map((e) => e.data).join(", "));
 });

demo/node/file-parse-promise.ts

@@ -3,8 +3,6 @@ import {
   startEvent,
   stopEvent,
 } from "../workflows/file-parse-agent.js";
-import { until } from "@llama-flow/core/stream/until";
-import { nothing } from "@llama-flow/core/stream/consumer";
 
 const directory = "..";
 
@@ -12,6 +10,6 @@ const { state, sendEvent, stream } = fileParseWorkflow.createContext();
 
 sendEvent(startEvent.with(directory));
 
-await nothing(until(stream, stopEvent));
+await stream.until(stopEvent).toArray();
 
-console.log("r", state.output);
+console.log(state.output);

demo/node/mcp-file-parse-tool.ts

@@ -3,8 +3,6 @@ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
 import { z } from "zod";
 import { fileParseWorkflow } from "../workflows/file-parse-agent.js";
 import { createWorkflow, workflowEvent } from "@llama-flow/core";
-import { until } from "@llama-flow/core/stream/until";
-import { nothing } from "@llama-flow/core/stream/consumer";
 import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
 
 const server = new McpServer({
@@ -24,7 +22,7 @@ const wrappedWorkflow = createWorkflow();
 wrappedWorkflow.handle([startEvent], async ({ data: { filePath } }) => {
   const { stream, sendEvent, state } = fileParseWorkflow.createContext();
   sendEvent(startEvent.with({ filePath }));
-  await nothing(until(stream, stopEvent));
+  await stream.until(stopEvent).toArray();
   return stopEvent.with({
     content: [
       {

demo/workflows/file-parse-agent.ts

@@ -2,7 +2,6 @@ import { createWorkflow, workflowEvent } from "@llama-flow/core";
 import { readdir, stat } from "node:fs/promises";
 import { resolve } from "node:path";
 import { AsyncLocalStorage } from "node:async_hooks";
-import { until } from "@llama-flow/core/stream/until";
 import { createStatefulMiddleware } from "@llama-flow/core/middleware/state";
 
 export const messageEvent = workflowEvent<string>({
@@ -39,7 +38,9 @@ const locks: {
 fileParseWorkflow.handle([startEvent], async ({ data: dir }) => {
   const { stream, sendEvent } = getContext();
   sendEvent(readDirEvent.with([dir, 0]));
-  await until(stream, () => locks.length > 0 && locks.every((l) => l.finish));
+  await stream
+    .until(() => locks.length > 0 && locks.every((l) => l.finish))
+    .toArray();
   return stopEvent.with();
 });
 

demo/workflows/llama-parse-workflow.ts

@@ -1,6 +1,4 @@
 import { workflowEvent, createWorkflow } from "@llama-flow/core";
-import { until } from "@llama-flow/core/stream/until";
-import { nothing } from "@llama-flow/core/stream/consumer";
 import { z } from "zod";
 import { zodEvent } from "@llama-flow/core/util/zod";
 import { createStatefulMiddleware } from "@llama-flow/core/middleware/state";
@@ -49,7 +47,7 @@ llamaParseWorkflow.handle(
       },
     ).then((res) => res.json());
     sendEvent(checkStatusEvent.with(id));
-    await nothing(until(stream, checkStatusSuccessEvent));
+    await stream.until(checkStatusSuccessEvent).toArray();
     return fetch(
       `https://api.cloud.llamaindex.ai/api/v1/parsing/job/${id}/result/markdown`,
       {

demo/workflows/tool-call-agent.ts

@@ -4,8 +4,6 @@ import type {
   ChatCompletionMessageToolCall,
   ChatCompletionTool,
 } from "openai/resources/chat/completions/completions";
-import { until } from "@llama-flow/core/stream/until";
-import { collect } from "@llama-flow/core/stream/consumer";
 
 const llm = new OpenAI();
 const tools = [
@@ -69,7 +67,7 @@ toolCallWorkflow.handle([chatEvent], async ({ data }) => {
       await Promise.all(
         choices[0].message.tool_calls.map(async (tool_call) => {
           sendEvent(toolCallEvent.with(tool_call));
-          return collect(until(stream, toolCallResultEvent));
+          return stream.until(toolCallResultEvent).toArray();
         }),
       )
     )

docs/llamaflow/advanced-events.mdx

@@ -65,9 +65,6 @@ You can create complex event flows with branching and merging patterns:
 
 ```ts
 import { createWorkflow, workflowEvent, getContext } from "@llama-flow/core";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
-import { filter } from "@llama-flow/core/stream/filter";
 import { pipeline } from "node:stream/promises";
 
 // Define events for a data processing pipeline with branching and merging
@@ -130,19 +127,13 @@ workflow.handle([startEvent], async (event) => {
   console.log(`Setting up merger to collect ${totalItems} processed results`);
 
   // Merge: collect results from both processing paths
-  const results = await collect(
-    until(
-      filter(
-        stream,
-        (ev) =>
-          processedValidEvent.include(ev) || processedInvalidEvent.include(ev),
-      ),
-      () => {
-        // Stop when we've collected the expected number of results
-        return results.length >= totalItems;
-      },
-    ),
-  );
+  const results = await stream
+    .filter(
+      (ev) =>
+        processedValidEvent.include(ev) || processedInvalidEvent.include(ev),
+    )
+    .take(totalItems)
+    .toArray();
 
   // Extract and sort the data from collected events
   const processedResults = results.map((event) => event.data);
@@ -200,9 +191,6 @@ You can filter and transform events to build sophisticated data processing pipel
 
 ```ts
 import { createWorkflow, workflowEvent, getContext } from "@llama-flow/core";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
-import { filter } from "@llama-flow/core/stream/filter";
 
 // Define events
 const dataEvent = workflowEvent<number>();
@@ -245,15 +233,13 @@ workflow.handle([initEvent], async (event) => {
 
   // Collect transformed events
   let numResults = 0;
-  const results = await collect(
-    until(
-      filter(stream, (ev) => transformedEvent.include(ev)),
-      () => {
-        numResults++;
-        return numResults >= event.data;
-      },
-    ),
-  );
+  const results = await stream
+    .filter(transformedEvent)
+    .until((event) => {
+      numResults++;
+      return numResults >= event.data;
+    })
+    .toArray();
 
   return resultEvent.with(results.map((r) => r.data));
 });
@@ -473,7 +459,6 @@ LlamaIndex workflows excel at managing complex asynchronous patterns:
 
 ```ts
 import { createWorkflow, workflowEvent, getContext } from "@llama-flow/core";
-import { until } from "@llama-flow/core/stream/until";
 
 // Events for an orchestration workflow
 const orchestrateEvent = workflowEvent<string[]>();
@@ -498,7 +483,7 @@ workflow.handle([orchestrateEvent], async (event) => {
   const results: Record<string, string> = {};
 
   // Process task completion and progress events
-  for await (const event of until(stream, () => completed === tasks.length)) {
+  for await (const event of stream.until(() => completed === tasks.length)) {
     if (progressEvent.include(event)) {
       console.log(`Task ${event.data.task}: ${event.data.progress}%`);
     } else if (taskCompleteEvent.include(event)) {

docs/llamaflow/basic-workflow.mdx

@@ -11,9 +11,6 @@ One of the most powerful features of workflows is the ability to run tasks in pa
 
 ```ts
 import { createWorkflow, workflowEvent, getContext } from "@llama-flow/core";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
-import { filter } from "@llama-flow/core/stream/filter";
 
 // Define events
 const startEvent = workflowEvent<string>();
@@ -29,7 +26,7 @@ let itemsToProcess = 10; // Total number of items
 let itemsProcessed = 0;
 
 // Process start event: fan out to multiple processItemEvent events
-workflow.handle([startEvent], (start) => {
+workflow.handle([startEvent], async (start) => {
   const { sendEvent, stream } = getContext();
   itemsProcessed = 0; // Reset counter for this execution context
 
@@ -39,17 +36,17 @@ workflow.handle([startEvent], (start) => {
   }
 
   // Use an async IIFE to collect results and emit completeEvent
-  (async () => {
-    const results = await collect(
-      // Filter for resultEvent and stop when all items are processed
-      until(
-        filter(stream, (event) => resultEvent.include(event)), // Only consider resultEvent
-        () => itemsProcessed === itemsToProcess, // Stop condition check
-      ),
-    );
+  try {
+    const results = await stream
+      .filter(resultEvent)
+      .until(() => itemsProcessed === itemsToProcess)
+      .toArray();
     // Send the final aggregated result
     sendEvent(completeEvent.with(results.map((event) => event.data)));
-  })().catch(console.error); // Handle potential errors during collection
+  } catch (err) {
+    console.error("Error processing items:", err);
+    // Handle error if needed
+  }
 
   // Note: This handler finishes *before* the collection completes.
   // Returning nothing or a specific "processing started" event might be appropriate.
@@ -100,8 +97,6 @@ You can implement conditional logic in your workflows:
 
 ```ts
 import { createWorkflow, workflowEvent } from "@llama-flow/core";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
 
 const inputEvent = workflowEvent<number>();
 const evenNumberEvent = workflowEvent<string>();
@@ -136,7 +131,7 @@ async function run(input_number: number) {
   sendEvent(inputEvent.with(input_number));
 
   // Collect all events until we get a stopEvent
-  const allEvents = await collect(until(stream, resultEvent));
+  const allEvents = await stream.until(resultEvent).toArray();
 
   // The last event will be the stopEvent that was requested
   const finalEvent = allEvents[allEvents.length - 1];

docs/llamaflow/index.mdx

@@ -111,15 +111,10 @@ workflow.handle([startEvent], async (event) => {
       );
     });
     let counter = 0;
-    const results = collect(
-      filter(
-        until(
-          stream,
-          () => counter++ === response.choices[0].message.tool_calls.length,
-        ),
-        toolCallResultEvent,
-      ),
-    );
+    const results = stream
+      .until(() => counter++ === response.choices[0].message.tool_calls.length)
+      .filter(toolCallResultEvent)
+      .toArray();
     return sendEvent(
       startEvent.with([...event.data, ...results.map((r) => r.data)]),
     );

docs/llamaflow/streaming.mdx

@@ -57,8 +57,6 @@ Workflows provide utility functions to make working with streams easier:
 
 ```ts
 import { createWorkflow, workflowEvent, getContext } from "@llama-flow/core";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
 
 const startEvent = workflowEvent<void>();
 const progressEvent = workflowEvent<number>();
@@ -82,10 +80,10 @@ const { stream, sendEvent } = workflow.createContext();
 sendEvent(startEvent.with());
 
 // Collect all events until resultEvent is encountered
-const events = await collect(until(stream, resultEvent));
-
-// Filter only progress events
-const progressEvents = events.filter((event) => progressEvent.include(event));
+const progressEvents = await stream
+  .until(resultEvent)
+  .filter(progressEvent)
+  .toArray();
 console.log(`Received ${progressEvents.length} progress updates`);
 ```
 

packages/core/src/core/stream.ts

@@ -187,4 +187,88 @@ export class WorkflowStream<R = any>
   ): ReadableStreamAsyncIterator<WorkflowEventData<any>> {
     return this.#stream.values(options);
   }
+
+  take(limit: number): WorkflowStream<WorkflowEventData<any>> {
+    let count = 0;
+    return WorkflowStream.fromReadableStream(
+      this.#stream.pipeThrough(
+        new TransformStream({
+          transform: (ev, controller) => {
+            if (count < limit) {
+              controller.enqueue(ev);
+              count++;
+            }
+            if (count >= limit) {
+              controller.terminate();
+            }
+          },
+        }),
+      ),
+    );
+  }
+
+  filter(predicate: WorkflowEvent<any>): WorkflowStream<R>;
+  filter(
+    predicate: (event: WorkflowEventData<any>) => boolean,
+  ): WorkflowStream<R>;
+  filter(
+    predicate:
+      | ((event: WorkflowEventData<any>) => boolean)
+      | WorkflowEvent<any>,
+  ): WorkflowStream<R> {
+    return WorkflowStream.fromReadableStream(
+      this.#stream.pipeThrough(
+        new TransformStream({
+          transform: (ev, controller) => {
+            if (
+              typeof predicate === "function"
+                ? predicate(ev)
+                : predicate.include(ev)
+            ) {
+              controller.enqueue(ev);
+            }
+          },
+        }),
+      ),
+    );
+  }
+
+  until(
+    predicate: (event: WorkflowEventData<any>) => boolean,
+  ): WorkflowStream<R>;
+  until(event: WorkflowEvent<any>): WorkflowStream<R>;
+  until(
+    predicate:
+      | WorkflowEvent<any>
+      | ((event: WorkflowEventData<any>) => boolean),
+  ): WorkflowStream<R> {
+    return WorkflowStream.fromReadableStream(
+      this.#stream.pipeThrough(
+        new TransformStream({
+          transform: (ev, controller) => {
+            controller.enqueue(ev);
+            if (
+              typeof predicate === "function"
+                ? predicate(ev)
+                : predicate.include(ev)
+            ) {
+              controller.terminate();
+            }
+          },
+        }),
+      ),
+    );
+  }
+
+  async toArray(): Promise<WorkflowEventData<any>[]> {
+    const events: WorkflowEventData<any>[] = [];
+    await this.#stream.pipeTo(
+      new WritableStream({
+        write: (event: WorkflowEventData<any>) => {
+          events.push(event);
+        },
+      }),
+    );
+    return events;
+  }
 }

packages/core/src/middleware/trace-events.ts

@@ -4,6 +4,7 @@ import {
   type WorkflowContext,
   type WorkflowEvent,
   type WorkflowEventData,
+  WorkflowStream,
 } from "@llama-flow/core";
 import { isPromiseLike } from "../core/utils";
 import {
@@ -91,15 +92,15 @@ export function withTraceEvents<
   ): HandlerRef<AcceptEvents, Result, Fn>;
   substream<T extends WorkflowEventData<any>>(
     eventData: WorkflowEventData<any>,
-    stream: ReadableStream<T>,
-  ): ReadableStream<T>;
+    stream: WorkflowStream<T>,
+  ): WorkflowStream<T>;
 } {
   return {
     ...workflow,
     substream: <T extends WorkflowEventData<any>>(
       eventData: WorkflowEventData<any>,
-      stream: ReadableStream<T>,
-    ): ReadableStream<T> => {
+      stream: WorkflowStream<T>,
+    ): WorkflowStream<T> => {
       const rootContext = eventToHandlerContextWeakMap.get(eventData);
       return stream.pipeThrough(
         new TransformStream({

packages/core/src/stream/consumer.ts

@@ -1,25 +1,23 @@
-import {
-  type WorkflowEvent,
-  type WorkflowEventData,
-  WorkflowStream,
-} from "@llama-flow/core";
+import { type WorkflowEventData, WorkflowStream } from "@llama-flow/core";
+
+const noopStream = new WritableStream({
+  write: () => {
+    // no-op
+  },
+});
 
 /**
  * A no-op function that consumes a stream of events and does nothing with them.
  *
  * Do not collect the raw stream from `workflow.createContext()`
  * or `getContext()`, it's infinite and will never finish
+ *
+ * @deprecated uss `await stream.toArray()` instead
  */
 export const nothing = async (
   stream: ReadableStream | WorkflowStream,
 ): Promise<void> => {
-  await stream.pipeTo(
-    new WritableStream<unknown>({
-      write: () => {
-        // no-op
-      },
-    }),
-  );
+  await stream.pipeTo(noopStream);
 };
 
 /**
@@ -27,6 +25,8 @@ export const nothing = async (
  *
  * Do not collect the raw stream from `workflow.createContext()`
  * or getContext()`, it's infinite and will never finish.
+ *
+ * @deprecated uss `await stream.toArray()` instead
  */
 export const collect = async <T extends WorkflowEventData<any>>(
   stream: ReadableStream<T> | WorkflowStream,

packages/core/src/stream/filter.ts

@@ -1,23 +0,0 @@
-import type {
-  WorkflowEvent,
-  WorkflowEventData,
-  WorkflowStream,
-} from "@llama-flow/core";
-
-export function filter<
-  Event extends WorkflowEventData<any>,
-  Final extends Event,
->(
-  stream: ReadableStream<Event> | WorkflowStream<Event>,
-  cond: (event: Event) => event is Final,
-): ReadableStream<Final> | WorkflowStream<Final> {
-  return stream.pipeThrough(
-    new TransformStream<Event, Final>({
-      transform(event, controller) {
-        if (cond(event)) {
-          controller.enqueue(event);
-        }
-      },
-    }),
-  );
-}

packages/core/src/stream/run.ts

@@ -1,14 +1,30 @@
 import type {
+  Workflow,
   WorkflowEvent,
   WorkflowEventData,
-  Workflow,
 } from "@llama-flow/core";
-import { collect } from "./consumer";
-import { until } from "./until";
+
+/**
+ * Runs a workflow with the provided events and returns the resulting stream.
+ *
+ * ```ts
+ * const events = await run(workflow, startEvent.with("42")).toArray();
+ * ```
+ *
+ */
+export function run(
+  workflow: Workflow,
+  events: WorkflowEventData<any> | WorkflowEventData<any>[],
+) {
+  const { stream, sendEvent } = workflow.createContext();
+  sendEvent(...(Array.isArray(events) ? events : [events]));
+  return stream;
+}
 
 /**
  * Runs a workflow with a specified input event and returns the first matching event of the specified output type.
  *
+ * @deprecated Use `stream.until().toArray()` for a more idiomatic approach.
  * @example
  * ```ts
  * const result = await runWorkflow(workflow, startEvent.with("42"), stopEvent);
@@ -26,22 +42,18 @@ export async function runWorkflow<Input, Output>(
   sendEvent(inputEvent);
 
   // Create a stream until we get the output event
-  const untilStream = until(stream, outputEvent);
-
-  // Find the first matching event
-  for await (const event of untilStream) {
-    if (outputEvent.include(event)) {
-      return event as WorkflowEventData<Output>;
-    }
+  const result = (await stream.until(outputEvent).toArray()).at(-1);
+  if (!result) {
+    throw new Error("No output event received");
   }
-
-  throw new Error(`No matching ${outputEvent.toString()} event found`);
+  return result as WorkflowEventData<Output>;
 }
 
 /**
  * Runs a workflow with a specified input event and collects all events until a specified output event is encountered.
  * Returns an array containing all events including the final output event.
  *
+ * @deprecated Use `stream.until().toArray()` for a more idiomatic approach.
  * @example
  * ```ts
  * const allEvents = await runAndCollect(workflow, startEvent.with("42"), stopEvent);
@@ -60,7 +72,7 @@ export async function runAndCollect<Input, Output>(
   sendEvent(inputEvent);
 
   // Collect all events until the output event
-  return await collect(until(stream, outputEvent));
+  return await stream.until(outputEvent).toArray();
 }
 
 /**
@@ -69,6 +81,7 @@ export async function runAndCollect<Input, Output>(
  *
  * This allows processing events one by one without collecting them all upfront.
  *
+ * @deprecated Use `stream.until().toArray()` for a more idiomatic approach.
  * @example
  * ```ts
  * const eventStream = runStream(workflow, startEvent.with("42"), stopEvent);
@@ -89,5 +102,5 @@ export function runStream<Input, Output>(
   sendEvent(inputEvent);
 
   // Return the stream that runs until the output event is encountered
-  return until(stream, outputEvent);
+  return stream.until(outputEvent).values();
 }

packages/core/src/stream/until.ts

@@ -1,54 +0,0 @@
-import {
-  type WorkflowEvent,
-  type WorkflowEventData,
-  WorkflowStream,
-} from "@llama-flow/core";
-
-const isWorkflowEvent = (value: unknown): value is WorkflowEvent<any> =>
-  value != null &&
-  typeof value === "object" &&
-  "with" in value &&
-  "include" in value;
-
-export function until(
-  stream: WorkflowStream | ReadableStream<WorkflowEventData<any>>,
-  cond: (event: WorkflowEventData<any>) => boolean | Promise<boolean>,
-): WorkflowStream;
-export function until<Stop>(
-  stream: WorkflowStream | ReadableStream<WorkflowEventData<any>>,
-  cond: WorkflowEvent<Stop>,
-): WorkflowStream;
-export function until(
-  stream: WorkflowStream | ReadableStream<WorkflowEventData<any>>,
-  cond:
-    | ((event: WorkflowEventData<any>) => boolean | Promise<boolean>)
-    | WorkflowEvent<any>,
-): WorkflowStream<WorkflowEventData<any>> {
-  let reader: ReadableStreamDefaultReader<WorkflowEventData<any>> | null = null;
-  return WorkflowStream.fromReadableStream(
-    new ReadableStream<WorkflowEventData<any>>({
-      start: () => {
-        reader = stream.getReader();
-      },
-      pull: async (controller) => {
-        const { done, value } = await reader!.read();
-        if (value) {
-          controller.enqueue(value);
-        }
-        if (done) {
-          reader!.releaseLock();
-          reader = null;
-          controller.close();
-        } else {
-          if (isWorkflowEvent(cond) && cond.include(value)) {
-            reader!.releaseLock();
-            controller.close();
-          } else if (typeof cond === "function" && (await cond(value))) {
-            reader!.releaseLock();
-            controller.close();
-          }
-        }
-      },
-    }),
-  );
-}

packages/core/tests/core/context-api.spec.ts

@@ -6,8 +6,6 @@ import {
   workflowEvent,
   type WorkflowEventData,
 } from "@llama-flow/core";
-import { until } from "@llama-flow/core/stream/until";
-import { collect, nothing } from "@llama-flow/core/stream/consumer";
 
 describe("workflow context api", () => {
   const startEvent = workflowEvent({
@@ -35,9 +33,9 @@ describe("workflow context api", () => {
       const { sendEvent, stream } = getContext();
       const ev = parseEvent.with(2);
       sendEvent(ev);
-      await nothing(
-        until(stream, (e) => parseResultEvent.include(e) && e.data === 0),
-      );
+      await stream
+        .until((e) => parseResultEvent.include(e) && e.data === 0)
+        .toArray();
       return stopEvent.with(1);
     });
     workflow.handle([parseEvent], async ({ data }) => {
@@ -50,9 +48,9 @@ describe("workflow context api", () => {
     });
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events.length).toBe(6);
     expect(events.at(-1)!.data).toBe(1);
     expect(events.map((e) => eventSource(e))).toEqual([
@@ -83,13 +81,13 @@ describe("workflow context api", () => {
       const { sendEvent, stream } = getContext();
       const ev = parseEvent.with(2);
       sendEvent(ev);
-      await nothing(
-        until(stream, (e) => parseResultEvent.include(e) && e.data === 0),
-      );
+      await stream
+        .until((e) => parseResultEvent.include(e) && e.data === 0)
+        .toArray();
       sendEvent(ev);
-      await nothing(
-        until(stream, (e) => parseResultEvent.include(e) && e.data === 0),
-      );
+      await stream
+        .until((e) => parseResultEvent.include(e) && e.data === 0)
+        .toArray();
       return stopEvent.with(1);
     });
     workflow.handle([parseEvent], async ({ data }) => {
@@ -104,9 +102,9 @@ describe("workflow context api", () => {
 
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events.length).toBe(10);
     expect(events.at(-1)!.data).toBe(1);
     expect(events.map((e) => eventSource(e))).toEqual([
@@ -134,9 +132,9 @@ describe("workflow context api", () => {
     workflow.handle([startEvent], fn);
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with());
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(fn).toBeCalledTimes(1);
     expect(events).toHaveLength(2);
   });
@@ -161,9 +159,9 @@ describe("workflow context api", () => {
     workflow.handle([aEvent], fn2);
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with());
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(fn).toBeCalledTimes(1);
     expect(events.map((e) => eventSource(e))).toEqual([
       startEvent,

packages/core/tests/core/readable-stream.spec.ts

@@ -6,8 +6,6 @@ import {
   workflowEvent,
   type WorkflowEventData,
 } from "@llama-flow/core";
-import { until } from "@llama-flow/core/stream/until";
-import { collect, nothing } from "@llama-flow/core/stream/consumer";
 
 describe("workflow basic", () => {
   const startEvent = workflowEvent<string>({
@@ -52,9 +50,9 @@ describe("workflow basic", () => {
 
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(3);
     expect(events.at(-1)!.data).toBe(1);
   });
@@ -69,9 +67,9 @@ describe("workflow basic", () => {
 
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(3);
     expect(events.at(-1)!.data).toBe(1);
   });
@@ -86,9 +84,9 @@ describe("workflow basic", () => {
 
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(3);
     expect(events.at(-1)!.data).toBe(1);
   });
@@ -104,9 +102,9 @@ describe("workflow basic", () => {
 
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(3);
     expect(events.at(-1)!.data).toBe(1);
   });
@@ -134,8 +132,8 @@ describe("workflow basic", () => {
     sendEvent(startEvent.with("100"));
     const [l, r] = newStream.tee();
     expect(newStream.locked).toBe(true);
-    await nothing(until(l, stopEvent));
-    await nothing(until(r, stopEvent));
+    await l.until(stopEvent).toArray();
+    await r.until(stopEvent).toArray();
   });
 });
 
@@ -152,9 +150,9 @@ describe("workflow simple logic", () => {
     workflow.handle([startEvent], f1);
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with());
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(f1).toBeCalledTimes(1);
     expect(events).toHaveLength(2);
   });
@@ -169,9 +167,9 @@ describe("workflow simple logic", () => {
     workflow.handle([event], f2);
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with());
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(f1).toBeCalledTimes(1);
     expect(f2).toBeCalledTimes(1);
     expect(events).toHaveLength(3);
@@ -189,9 +187,9 @@ describe("workflow simple logic", () => {
     {
       const { stream, sendEvent } = workflow.createContext();
       sendEvent(startEvent.with(1));
-      const events: WorkflowEventData<any>[] = await collect(
-        until(stream, stopEvent),
-      );
+      const events: WorkflowEventData<any>[] = await stream
+        .until(stopEvent)
+        .toArray();
       expect(f1).toBeCalledTimes(2);
       expect(events).toHaveLength(3);
     }
@@ -199,9 +197,9 @@ describe("workflow simple logic", () => {
     {
       const { stream, sendEvent } = workflow.createContext();
       sendEvent(startEvent.with(-1));
-      const events: WorkflowEventData<any>[] = await collect(
-        until(stream, stopEvent),
-      );
+      const events: WorkflowEventData<any>[] = await stream
+        .until(stopEvent)
+        .toArray();
       expect(f1).toBeCalledTimes(1);
       expect(events).toHaveLength(2);
     }
@@ -225,9 +223,9 @@ describe("workflow simple logic", () => {
     workflow.handle([event2], f3);
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with());
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(f1).toBeCalledTimes(1);
     expect(f2).toBeCalledTimes(2);
     expect(f3).toBeCalledTimes(2);
@@ -248,9 +246,9 @@ describe("workflow simple logic", () => {
     {
       const { stream, sendEvent } = workflow.createContext();
       sendEvent(startEvent.with("100"));
-      const events: WorkflowEventData<any>[] = await collect(
-        until(stream, stopEvent),
-      );
+      const events: WorkflowEventData<any>[] = await stream
+        .until(stopEvent)
+        .toArray();
       expect(events).toHaveLength(2);
       expect(events.at(-1)!.data).toBe(1);
     }
@@ -258,9 +256,9 @@ describe("workflow simple logic", () => {
     {
       const { stream, sendEvent } = workflow.createContext();
       sendEvent(startEvent.with("200"));
-      const events: WorkflowEventData<any>[] = await collect(
-        until(stream, stopEvent),
-      );
+      const events: WorkflowEventData<any>[] = await stream
+        .until(stopEvent)
+        .toArray();
       expect(events).toHaveLength(2);
       expect(events.at(-1)!.data).toBe(-1);
     }
@@ -305,9 +303,9 @@ describe("workflow simple logic", () => {
 
     const { stream, sendEvent } = jokeFlow.createContext();
     sendEvent(startEvent.with("pirates"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(5);
     expect(events.at(-1)!.data).toBe("critique analysis");
     expect(events.map((e) => eventSource(e))).toEqual([
@@ -343,9 +341,9 @@ describe("workflow simple logic", () => {
 
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(4);
     expect(events.at(-1)!.data).toBe(1);
     expect(events.map((e) => eventSource(e))).toEqual([
@@ -384,9 +382,9 @@ describe("workflow simple logic", () => {
     );
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(102);
     expect(events.at(-1)!.data).toBe(1);
     expect(events.map((e) => eventSource(e))).toEqual([
@@ -422,9 +420,9 @@ describe("workflow simple logic", () => {
     {
       const { stream, sendEvent } = workflow.createContext();
       sendEvent(startEvent.with("100"));
-      const events: WorkflowEventData<any>[] = await collect(
-        until(stream, stopEvent),
-      );
+      const events: WorkflowEventData<any>[] = await stream
+        .until(stopEvent)
+        .toArray();
       expect(events).toHaveLength(3);
       expect(events.at(-1)!.data).toBe(1);
       expect(events.map((e) => eventSource(e))).toEqual([
@@ -466,9 +464,9 @@ describe("workflow simple logic", () => {
     {
       const { stream, sendEvent } = workflow.createContext();
       sendEvent(startEvent.with("100"));
-      const events: WorkflowEventData<any>[] = await collect(
-        until(stream, stopEvent),
-      );
+      const events: WorkflowEventData<any>[] = await stream
+        .until(stopEvent)
+        .toArray();
       expect(events).toHaveLength(102);
     }
   });
@@ -496,9 +494,9 @@ describe("workflow simple logic", () => {
     });
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("100"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events).toHaveLength(4);
     expect(events.at(-1)!.data).toBe(1);
     expect(events.map((e) => eventSource(e))).toEqual([

packages/core/tests/core/shared/events.ts

@@ -4,3 +4,8 @@ export const messageEvent = workflowEvent({
   debugLabel: "message",
   uniqueId: "message",
 });
+
+export const haltEvent = workflowEvent({
+  debugLabel: "halt",
+  uniqueId: "halt",
+});

packages/core/tests/core/stream.spec.ts

@@ -6,6 +6,7 @@ import {
 } from "@llama-flow/core";
 import { describe, expect, test } from "vitest";
 import * as events from "./shared/events";
+import { haltEvent } from "./shared/events";
 
 describe("stream api", () => {
   test("should able to create stream", async () => {
@@ -37,4 +38,55 @@ describe("stream api", () => {
     expect(eventSource(list[0])).toBe(events.messageEvent);
     //#endregion
   });
+
+  test("stream.until", async () => {
+    const workflow = createWorkflow();
+    const { sendEvent, stream } = workflow.createContext();
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.haltEvent.with());
+    const list = await stream.until(events.haltEvent).toArray();
+    expect(list).toHaveLength(4);
+    expect(list.map(eventSource)).toEqual([
+      events.messageEvent,
+      events.messageEvent,
+      events.messageEvent,
+      events.haltEvent,
+    ]);
+  });
+
+  test("stream.filter", async () => {
+    const workflow = createWorkflow();
+    const { sendEvent, stream } = workflow.createContext();
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.haltEvent.with());
+    const list = await stream
+      .until(events.haltEvent)
+      .filter(events.messageEvent)
+      .toArray();
+    expect(list).toHaveLength(3);
+    expect(list.map(eventSource)).toEqual([
+      events.messageEvent,
+      events.messageEvent,
+      events.messageEvent,
+    ]);
+  });
+
+  test("stream.take", async () => {
+    const workflow = createWorkflow();
+    const { sendEvent, stream } = workflow.createContext();
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.messageEvent.with());
+    sendEvent(events.haltEvent.with());
+    const list = await stream.until(events.haltEvent).take(2).toArray();
+    expect(list).toHaveLength(2);
+    expect(list.map(eventSource)).toEqual([
+      events.messageEvent,
+      events.messageEvent,
+    ]);
+  });
 });

packages/core/tests/core/sub-workflow.spec.ts

@@ -5,9 +5,7 @@ import {
   getContext,
   workflowEvent,
 } from "@llama-flow/core";
-import { runWorkflow } from "@llama-flow/core/stream/run";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
+import { run } from "@llama-flow/core/stream/run";
 
 describe("sub workflow", () => {
   test("basic", async () => {
@@ -23,16 +21,16 @@ describe("sub workflow", () => {
         sendEvent(stopEvent.with());
       });
       await Promise.all([
-        runWorkflow(subWorkflow, startEvent.with(), stopEvent),
-        runWorkflow(subWorkflow, startEvent.with(), stopEvent),
-        runWorkflow(subWorkflow, startEvent.with(), stopEvent),
-      ]).then((evt) => sendEvent(...evt));
+        run(subWorkflow, startEvent.with()).filter(stopEvent).take(1).toArray(),
+        run(subWorkflow, startEvent.with()).filter(stopEvent).take(1).toArray(),
+        run(subWorkflow, startEvent.with()).filter(stopEvent).take(1).toArray(),
+      ]).then((evt) => sendEvent(...evt.flat()));
       sendEvent(haltEvent.with());
     });
     const { sendEvent, stream } = rootWorkflow.createContext();
     sendEvent(startEvent.with());
 
-    const events = await collect(until(stream, haltEvent));
+    const events = await stream.until(haltEvent).toArray();
     expect(events.length).toBe(5);
     expect(events.map(eventSource)).toEqual([
       startEvent,

packages/core/tests/integration/stream-library.spec.ts

@@ -7,7 +7,6 @@ import {
   getContext,
   type WorkflowEventData,
 } from "@llama-flow/core";
-import { until } from "@llama-flow/core/stream/until";
 
 describe("node:stream", () => {
   test("basic usage", async () => {
@@ -58,7 +57,7 @@ describe("stream-chain", () => {
     const outputs: WorkflowEventData<any>[] = [];
     const pipeline = chain([
       // fixme: upstream should be treat it as a stream
-      until(stream, stopEvent)[Symbol.asyncIterator],
+      stream.until(stopEvent)[Symbol.asyncIterator],
       new TransformStream({
         transform: (event: WorkflowEventData<any>, controller) => {
           if (messageEvent.include(event)) {

packages/core/tests/middleware/full-workflow.spec.ts

@@ -11,8 +11,6 @@ import { withValidation } from "@llama-flow/core/middleware/validation";
 import { zodEvent } from "@llama-flow/core/util/zod";
 import { z } from "zod";
 import { webcrypto } from "node:crypto";
-import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
 
 describe("full workflow middleware", () => {
   const createFullWorkflow = <
@@ -75,9 +73,9 @@ describe("full workflow middleware", () => {
     const id = webcrypto.randomUUID();
     const { sendEvent, stream } = workflow.createContext(id);
     sendEvent(startEvent.with("start"));
-    const events: WorkflowEventData<any>[] = await collect(
-      until(stream, stopEvent),
-    );
+    const events: WorkflowEventData<any>[] = await stream
+      .until(stopEvent)
+      .toArray();
     expect(events.length).toBe(2);
     expect(events.map(eventSource)).toEqual([startEvent, stopEvent]);
   });

packages/core/tests/middleware/with-trace-events.spec.ts

@@ -11,9 +11,7 @@ import {
   createHandlerDecorator,
   getEventOrigins,
 } from "@llama-flow/core/middleware/trace-events";
-import { filter } from "@llama-flow/core/stream/filter";
 import { collect } from "@llama-flow/core/stream/consumer";
-import { until } from "@llama-flow/core/stream/until";
 import { pipeline } from "node:stream/promises";
 
 const groupBy = <T>(
@@ -77,18 +75,15 @@ describe("with trace events", () => {
     context.sendEvent(startEvent.with());
 
     const [l, r] = stream.tee();
-    const allEvents = await collect(
-      until(l, (ev) => messageEvent.include(ev) && ev.data === 2),
-    );
-    const events = await collect(
-      filter(
-        workflow.substream(
-          ev,
-          until(r, (ev) => ev.data === 2),
-        ),
-        (e) => messageEvent.include(e),
-      ),
-    );
+    const allEvents = await l.until((ev) => ev.data === 2).toArray();
+
+    const events = await workflow
+      .substream(
+        ev,
+        r.until((ev) => ev.data === 2),
+      )
+      .filter((e) => messageEvent.include(e))
+      .toArray();
     expect(counter).toBe(3);
     expect(allEvents.length).toBe(6);
     expect(events.length).toBe(1);
@@ -113,9 +108,7 @@ describe("with trace events", () => {
     context.sendEvent(ev);
     context.sendEvent(startEvent.with());
     context.sendEvent(startEvent.with());
-    const events = await collect(
-      until(stream, (ev) => messageEvent.include(ev) && ev.data === 2),
-    );
+    const events = await stream.until((ev) => ev.data === 2).toArray();
     expect(events.length).toBe(6);
   });
 
@@ -242,16 +235,10 @@ describe("get event origins", () => {
       sendEvent(branchBEvent.with("Branch B"));
       sendEvent(branchCEvent.with("Branch C"));
 
-      let condition = 0;
-      const results = await collect(
-        until(
-          filter(stream, (ev) => branchCompleteEvent.include(ev)),
-          () => {
-            condition++;
-            return condition === 3;
-          },
-        ),
-      );
+      const results = await stream
+        .filter(branchCompleteEvent)
+        .take(3)
+        .toArray();
 
       const result = groupBy(
         results,

packages/core/vitest.config.ts

@@ -13,6 +13,7 @@ export default defineConfig({
         test: {
           name: "DOM",
           environment: "happy-dom",
+          exclude: ["**/lib/**", "**/dist/**", "**/node_modules/**"],
         },
       },
       {
@@ -24,6 +25,7 @@ export default defineConfig({
         test: {
           name: "Node.js",
           environment: "node",
+          exclude: ["**/lib/**", "**/dist/**", "**/node_modules/**"],
         },
       },
       {
@@ -35,6 +37,7 @@ export default defineConfig({
         test: {
           name: "Edge Runtime",
           environment: "edge-runtime",
+          exclude: ["**/lib/**", "**/dist/**", "**/node_modules/**"],
         },
       },
     ],