fix: extends standard readable stream (#103)

Author: himself65

.changeset/gentle-rings-talk.md

@@ -0,0 +1,5 @@
+---
+"@llama-flow/core": patch
+---
+
+fix: workflow stream extends standard readable stream

packages/core/src/core/stream.ts

@@ -8,10 +8,10 @@ import {
 import { createSubscribable, type Subscribable } from "./utils";
 
 export class WorkflowStream<R = any>
-  implements AsyncIterable<R>, ReadableStream<R>
+  extends ReadableStream<R>
+  implements AsyncIterable<R>
 {
   #stream: ReadableStream<R>;
-
   #subscribable: Subscribable<[data: R], void>;
 
   on(
@@ -40,6 +40,7 @@ export class WorkflowStream<R = any>
         "Either subscribable or root stream must be provided",
       );
     }
+    super();
     if (!subscribable) {
       this.#subscribable = createSubscribable<[data: R], void>();
       this.#stream = rootStream!.pipeThrough(

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

@@ -56,8 +56,7 @@ describe("stream-chain", () => {
     sendEvent(startEvent.with());
     const outputs: WorkflowEventData<any>[] = [];
     const pipeline = chain([
-      // fixme: upstream should be treat it as a stream
-      stream.until(stopEvent)[Symbol.asyncIterator],
+      stream.until(stopEvent),
       new TransformStream({
         transform: (event: WorkflowEventData<any>, controller) => {
           if (messageEvent.include(event)) {

packages/llamaindex/src/index.ts

@@ -5,7 +5,6 @@ import {
   workflowEvent,
   WorkflowStream,
 } from "@llama-flow/core";
-import { collect } from "@llama-flow/core/stream/consumer";
 import { createStatefulMiddleware } from "@llama-flow/core/middleware/state";
 
 type Handler<
@@ -179,22 +178,22 @@ export class Workflow<ContextData, Start, Stop> {
     Object.assign(result, {
       then: async (resolve: any, reject: any) => {
         try {
-          const events = await collect(result);
+          const events = await result.toArray();
           resolve(events.at(-1)!);
         } catch (error) {
           reject(error);
         }
       },
       catch: async (reject: any) => {
         try {
-          await collect(result);
+          await result.toArray();
         } catch (error) {
           reject(error);
         }
       },
       finally: async (resolve: any) => {
         try {
-          await collect(result);
+          await result.toArray();
         } finally {
           resolve();
         }