feat: add WorkflowStream.fromResponse/toResponse API (#92)

.changeset/olive-phones-invite.md

@@ -0,0 +1,5 @@
+---
+"@llama-flow/core": patch
+---
+
+feat: add `WorkflowStream.fromResponse/toResponse` API

demo/deno/deno.lock

@@ -3,7 +3,7 @@
   "specifiers": {
     "jsr:@std/assert@1": "1.0.12",
     "jsr:@std/internal@^1.0.6": "1.0.6",
-    "npm:@llama-flow/core@*": "0.3.4"
+    "npm:@llama-flow/core@*": "0.3.9"
   },
   "jsr": {
     "@std/assert@1.0.12": {
@@ -17,7 +17,7 @@
     }
   },
   "npm": {
-    "@llama-flow/core@0.3.4": {}
+    "@llama-flow/core@0.3.9": {}
   },
   "workspace": {
     "dependencies": [
@@ -25,12 +25,13 @@
       "npm:@llama-flow/core@*"
     ],
     "patches": {
-      "npm:@llama-flow/core@0.3.4": {
+      "npm:@llama-flow/core@0.3.9": {
         "peerDependencies": [
           "npm:@modelcontextprotocol/sdk@^1.7.0",
           "npm:hono@^4.7.4",
           "npm:next@^15.2.2",
           "npm:p-retry@^6.2.1",
+          "npm:rxjs@^7.8.2",
           "npm:zod@^3.24.2"
         ],
         "peerDependenciesMeta": {
@@ -46,6 +47,9 @@
           "p-retry": {
             "optional": true
           },
+          "rxjs": {
+            "optional": true
+          },
           "zod": {
             "optional": true
           }

demo/deno/main_test.ts

@@ -5,7 +5,7 @@ Deno.test(function workflowRun() {
   const { sendEvent, stream } = workflow.createContext();
   sendEvent(startEvent.with());
   stream
-    .pipeThrough(
+    .pipeThrough<string>(
       new TransformStream({
         transform: (event, controller) => {
           if (endEvent.include(event)) {

packages/core/src/core/context.ts

@@ -54,7 +54,7 @@ export type ContextNext = (
 ) => void;
 
 export type WorkflowContext = {
-  get stream(): WorkflowStream<WorkflowEvent<any>>;
+  get stream(): WorkflowStream;
   get signal(): AbortSignal;
   sendEvent: (...events: WorkflowEventData<any>[]) => void;
 

packages/core/src/core/index.ts

@@ -11,3 +11,5 @@ export {
   type WorkflowEventConfig,
   type InferWorkflowEventData,
 } from "./event";
+// stream
+export { WorkflowStream } from "./stream";

packages/core/src/core/stream.ts

@@ -1,35 +1,104 @@
-import { type WorkflowEvent, type WorkflowEventData } from "./event";
-import type { Subscribable } from "./utils";
+import {
+  eventSource,
+  type WorkflowEvent,
+  type WorkflowEventData,
+} from "./event";
+import { createSubscribable, type Subscribable } from "./utils";
+
+export class WorkflowStream<R = any>
+  implements
+    AsyncIterable<WorkflowEventData<any>>,
+    ReadableStream<WorkflowEventData<any>>
+{
+  #stream: ReadableStream<WorkflowEventData<any>>;
 
-export class WorkflowStream<Event extends WorkflowEvent<any>> {
   #subscribable: Subscribable<[event: WorkflowEventData<any>], void>;
-  #stream: ReadableStream<ReturnType<Event["with"]>>;
 
-  on<T extends Event>(
-    event: T,
-    handler: (event: ReturnType<T["with"]>) => void,
+  on(
+    event: WorkflowEvent<any>,
+    handler: (event: WorkflowEventData<any>) => void,
   ): () => void {
     return this.#subscribable.subscribe((ev) => {
       if (event.include(ev)) {
-        handler(ev as ReturnType<T["with"]>);
+        handler(ev);
       }
     });
   }
 
   constructor(
     subscribable: Subscribable<[event: WorkflowEventData<any>], void>,
-    stream: ReadableStream<ReturnType<Event["with"]>>,
+    stream: ReadableStream<WorkflowEventData<any>>,
   ) {
     this.#subscribable = subscribable;
     this.#stream = stream;
   }
 
+  static fromResponse(
+    response: Response,
+    eventMap: Record<string, WorkflowEvent<any>>,
+  ): WorkflowStream {
+    const body = response.body;
+    if (!body) {
+      throw new Error("Response body is not readable");
+    }
+    const subscribable = createSubscribable<
+      [event: WorkflowEventData<any>],
+      void
+    >();
+    const events = Object.values(eventMap);
+    const stream = body
+      .pipeThrough(new TextDecoderStream())
+      .pipeThrough<WorkflowEventData<any>>(
+        new TransformStream({
+          transform: (chunk, controller) => {
+            const eventData = JSON.parse(chunk) as {
+              data: ReturnType<WorkflowEvent<any>["with"]>;
+              uniqueId: string;
+            };
+            const targetEvent = events.find(
+              (e) => e.uniqueId === eventData.uniqueId,
+            );
+            if (targetEvent) {
+              const ev = targetEvent.with(
+                eventData.data,
+              ) as WorkflowEventData<any>;
+              subscribable.publish(ev);
+              controller.enqueue(ev);
+            }
+          },
+        }),
+      );
+    return new WorkflowStream(subscribable, stream);
+  }
+
+  toResponse(
+    init?: ResponseInit,
+  ): R extends WorkflowEventData<any> ? Response : never {
+    return new Response(
+      this.#stream
+        .pipeThrough(
+          new TransformStream({
+            transform: (event, controller) => {
+              controller.enqueue(
+                JSON.stringify({
+                  data: event.data,
+                  uniqueId: eventSource(event)!.uniqueId,
+                }) + "\n",
+              );
+            },
+          }),
+        )
+        .pipeThrough(new TextEncoderStream()),
+      init,
+    ) as any;
+  }
+
   get locked() {
     return this.#stream.locked;
   }
 
   [Symbol.asyncIterator](): ReadableStreamAsyncIterator<
-    ReturnType<Event["with"]>
+    WorkflowEventData<any>
   > {
     return this.#stream[Symbol.asyncIterator]();
   }
@@ -40,29 +109,38 @@ export class WorkflowStream<Event extends WorkflowEvent<any>> {
 
   // make type compatible with Web ReadableStream API
   getReader(options: { mode: "byob" }): ReadableStreamBYOBReader;
-  getReader(): ReadableStreamDefaultReader<ReturnType<Event["with"]>>;
+  getReader(): ReadableStreamDefaultReader<WorkflowEventData<any>>;
   getReader(
     options?: ReadableStreamGetReaderOptions,
-  ): ReadableStreamReader<ReturnType<Event["with"]>>;
+  ): ReadableStreamReader<WorkflowEventData<any>>;
   getReader(): any {
     return this.#stream.getReader();
   }
 
-  pipeThrough<T = WorkflowEventData<any>>(
-    transform: ReadableWritablePair<T, ReturnType<Event["with"]>>,
+  // @ts-expect-error
+  pipeThrough<T>(
+    transform: ReadableWritablePair<T, R>,
     options?: StreamPipeOptions,
-  ): ReadableStream<T> {
-    return this.#stream.pipeThrough(transform, options);
+  ): WorkflowStream<T> {
+    const stream = this.#stream.pipeThrough(
+      // @ts-expect-error
+      transform,
+      options,
+    ) as any;
+    return new WorkflowStream(this.#subscribable, stream);
   }
 
+  // @ts-expect-error
   pipeTo(
-    destination: WritableStream<ReturnType<Event["with"]>>,
+    destination: WritableStream<R>,
     options?: StreamPipeOptions,
   ): Promise<void> {
+    // @ts-expect-error
     return this.#stream.pipeTo(destination, options);
   }
 
-  tee(): [WorkflowStream<Event>, WorkflowStream<Event>] {
+  // @ts-expect-error
+  tee(): [WorkflowStream, WorkflowStream] {
     const [l, r] = this.#stream.tee();
     return [
       new WorkflowStream(this.#subscribable, l),
@@ -72,7 +150,7 @@ export class WorkflowStream<Event extends WorkflowEvent<any>> {
 
   values(
     options?: ReadableStreamIteratorOptions,
-  ): ReadableStreamAsyncIterator<ReturnType<Event["with"]>> {
+  ): ReadableStreamAsyncIterator<WorkflowEventData<any>> {
     return this.#stream.values(options);
   }
 }

packages/core/src/core/utils.ts

@@ -46,6 +46,9 @@ export function getSubscribers<Args extends any[], R>(
   return __internal__subscribesSourcemap.get(subscribable)!;
 }
 
+/**
+ * @internal
+ */
 export function createSubscribable<Args extends any[], R>(): Subscribable<
   Args,
   R

packages/core/src/observable.ts

@@ -1,8 +1,8 @@
 import { Observable } from "rxjs";
-import type { WorkflowEventData } from "@llama-flow/core";
+import { type WorkflowEventData, WorkflowStream } from "@llama-flow/core";
 
 export const toObservable = (
-  stream: ReadableStream<WorkflowEventData<any>>,
+  stream: ReadableStream<WorkflowEventData<any>> | WorkflowStream,
 ): Observable<WorkflowEventData<any>> => {
   return new Observable((subscriber) => {
     const reader = stream.getReader();

packages/core/src/stream/consumer.ts

@@ -1,3 +1,9 @@
+import {
+  type WorkflowEvent,
+  type WorkflowEventData,
+  WorkflowStream,
+} from "@llama-flow/core";
+
 /**
  * A no-op function that consumes a stream of events and does nothing with them.
  *
@@ -5,7 +11,7 @@
  * or `getContext()`, it's infinite and will never finish
  */
 export const nothing = async (
-  stream: ReadableStream<unknown>,
+  stream: ReadableStream | WorkflowStream,
 ): Promise<void> => {
   await stream.pipeTo(
     new WritableStream<unknown>({
@@ -22,11 +28,13 @@ export const nothing = async (
  * Do not collect the raw stream from `workflow.createContext()`
  * or getContext()`, it's infinite and will never finish.
  */
-export const collect = async <T>(stream: ReadableStream<T>): Promise<T[]> => {
-  const events: T[] = [];
+export const collect = async <T extends WorkflowEventData<any>>(
+  stream: ReadableStream<T> | WorkflowStream,
+): Promise<WorkflowEventData<any>[]> => {
+  const events: WorkflowEventData<any>[] = [];
   await stream.pipeTo(
-    new WritableStream<T>({
-      write: (event) => {
+    new WritableStream({
+      write: (event: WorkflowEventData<any>) => {
         events.push(event);
       },
     }),

packages/core/src/stream/filter.ts

@@ -1,10 +1,14 @@
-import type { WorkflowEventData } from "@llama-flow/core";
+import type {
+  WorkflowEvent,
+  WorkflowEventData,
+  WorkflowStream,
+} from "@llama-flow/core";
 
 export function filter<
   Event extends WorkflowEventData<any>,
   Final extends Event,
 >(
-  stream: ReadableStream<Event>,
+  stream: ReadableStream<Event> | WorkflowStream,
   cond: (event: Event) => event is Final,
 ): ReadableStream<Final> {
   return stream.pipeThrough(
@@ -15,5 +19,5 @@ export function filter<
         }
       },
     }),
-  );
+  ) as ReadableStream<Final>;
 }

packages/core/src/stream/find.ts

@@ -1,10 +1,14 @@
-import type { WorkflowEvent, WorkflowEventData } from "@llama-flow/core";
+import {
+  type WorkflowEvent,
+  type WorkflowEventData,
+  WorkflowStream,
+} from "@llama-flow/core";
 
 /**
  * Consume a stream of events with a given event and time.
  */
 export async function find<T>(
-  stream: ReadableStream<WorkflowEventData<any>>,
+  stream: ReadableStream<WorkflowEventData<any>> | WorkflowStream,
   event: WorkflowEvent<T>,
 ): Promise<WorkflowEventData<T>> {
   const reader = stream.getReader();

packages/core/src/stream/until.ts

@@ -1,4 +1,8 @@
-import type { WorkflowEvent, WorkflowEventData } from "@llama-flow/core";
+import {
+  type WorkflowEvent,
+  type WorkflowEventData,
+  WorkflowStream,
+} from "@llama-flow/core";
 
 const isWorkflowEvent = (value: unknown): value is WorkflowEvent<any> =>
   value != null &&
@@ -7,15 +11,15 @@ const isWorkflowEvent = (value: unknown): value is WorkflowEvent<any> =>
   "include" in value;
 
 export function until(
-  stream: ReadableStream<WorkflowEventData<any>>,
+  stream: ReadableStream<WorkflowEventData<any>> | WorkflowStream,
   cond: (event: WorkflowEventData<any>) => boolean | Promise<boolean>,
 ): ReadableStream<WorkflowEventData<any>>;
 export function until<Stop>(
-  stream: ReadableStream<WorkflowEventData<any>>,
+  stream: ReadableStream<WorkflowEventData<any>> | WorkflowStream,
   cond: WorkflowEvent<Stop>,
 ): ReadableStream<WorkflowEventData<any> | WorkflowEventData<Stop>>;
 export function until(
-  stream: ReadableStream<WorkflowEventData<any>>,
+  stream: ReadableStream<WorkflowEventData<any>> | WorkflowStream,
   cond:
     | ((event: WorkflowEventData<any>) => boolean | Promise<boolean>)
     | WorkflowEvent<any>,

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

@@ -0,0 +1,6 @@
+import { workflowEvent } from "@llama-flow/core";
+
+export const messageEvent = workflowEvent({
+  debugLabel: "message",
+  uniqueId: "message",
+});

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

@@ -0,0 +1,40 @@
+import {
+  WorkflowStream,
+  createWorkflow,
+  type WorkflowEventData,
+  eventSource,
+} from "@llama-flow/core";
+import { describe, expect, test } from "vitest";
+import * as events from "./shared/events";
+
+describe("stream api", () => {
+  test("should able to create stream", async () => {
+    //#region server
+    const workflow = createWorkflow();
+    const { sendEvent, stream } = workflow.createContext();
+    sendEvent(events.messageEvent.with());
+    const response = stream
+      .pipeThrough(
+        new TransformStream({
+          transform(event, controller) {
+            controller.enqueue(event);
+            controller.terminate();
+          },
+        }),
+      )
+      .toResponse();
+    //#endregion
+
+    //#region client
+    const clientSideStream = WorkflowStream.fromResponse(response, {
+      message: events.messageEvent,
+    });
+    const list: WorkflowEventData<any>[] = [];
+    for await (const event of clientSideStream) {
+      list.push(event);
+    }
+    expect(list).toHaveLength(1);
+    expect(eventSource(list[0])).toBe(events.messageEvent);
+    //#endregion
+  });
+});

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

@@ -1,5 +1,10 @@
 import { describe, test, vi, expectTypeOf, type Mock, expect } from "vitest";
-import { createWorkflow, getContext, workflowEvent } from "@llama-flow/core";
+import {
+  createWorkflow,
+  getContext,
+  workflowEvent,
+  type WorkflowEventData,
+} from "@llama-flow/core";
 import {
   withTraceEvents,
   runOnce,
@@ -297,7 +302,7 @@ describe("get event origins", () => {
     const { stream, sendEvent } = workflow.createContext();
     sendEvent(startEvent.with("initial data"));
 
-    const result = await pipeline(stream, async function (source) {
+    await pipeline(stream, async function (source) {
       for await (const event of source) {
         if (stopEvent.include(event)) {
           return `Result: ${event.data}`;

packages/llamaindex/src/index.ts

@@ -4,6 +4,7 @@ import {
   type WorkflowEventData as CoreWorkflowEventData,
   workflowEvent,
   getContext,
+  WorkflowStream,
 } from "@llama-flow/core";
 import { collect } from "@llama-flow/core/stream/consumer";
 import { withStore } from "@llama-flow/core/middleware/store";
@@ -19,7 +20,7 @@ type Handler<
 
 export type StepContext<T = unknown> = {
   sendEvent: (event: WorkflowEvent<any>) => void;
-  get stream(): ReadableStream<WorkflowEvent<any>>;
+  get stream(): WorkflowStream;
   get data(): T;
 };
 
@@ -113,7 +114,7 @@ export class Workflow<ContextData, Start, Stop> {
               context.sendEvent(coreEvent);
             },
             get stream() {
-              return context.stream.pipeThrough<WorkflowEvent<any>>(
+              return context.stream.pipeThrough(
                 new TransformStream({
                   transform: (event, controller) => {
                     controller.enqueue(coreEventWeakMap.get(event)!);