feat(workflow): add observer/watch stream APIs (#1100)

Author: omeraplak

.changeset/blue-bags-remember.md

@@ -0,0 +1,73 @@
+---
+"@voltagent/core": minor
+---
+
+feat: add workflow observer/watch APIs for stream results
+
+### What's New
+
+- Added run-level observer APIs on `WorkflowStreamResult`:
+  - `watch(cb)`
+  - `watchAsync(cb)`
+  - `observeStream()`
+  - `streamLegacy()`
+- These APIs are now available on:
+  - `workflow.stream(...)`
+  - `workflow.timeTravelStream(...)`
+  - resumed stream results returned from `.resume(...)`
+- Added test coverage for event ordering, unsubscribe behavior, multiple watchers, callback isolation, and `observeStream()` close semantics.
+
+### SDK Example
+
+```ts
+const stream = workflow.stream({ value: 3 });
+
+const unwatch = stream.watch((event) => {
+  console.log("[watch]", event.type, event.from);
+});
+
+const unwatchAsync = await stream.watchAsync(async (event) => {
+  if (event.type === "workflow-error") {
+    await notifyOps(event);
+  }
+});
+
+const reader = stream.observeStream().getReader();
+const observerTask = (async () => {
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    console.log("[observeStream]", value.type);
+  }
+})();
+
+for await (const event of stream) {
+  console.log("[main iterator]", event.type);
+}
+
+unwatch();
+unwatchAsync();
+await observerTask;
+
+const state = await stream.streamLegacy().getWorkflowState();
+console.log("final status:", state?.status);
+```
+
+### Time Travel Stream Example
+
+```ts
+const replayStream = workflow.timeTravelStream({
+  executionId: sourceExecutionId,
+  stepId: "step-approval",
+});
+
+const stopReplayWatch = replayStream.watch((event) => {
+  console.log("[replay]", event.type, event.from);
+});
+
+for await (const event of replayStream) {
+  console.log("[replay iterator]", event.type);
+}
+
+stopReplayWatch();
+```

packages/core/src/workflow/core.spec.ts

@@ -382,6 +382,97 @@ describe.sequential("workflow streaming", () => {
     expect(result).toEqual({ result: 10 });
   });
 
+  it("should expose watch/watchAsync and streamLegacy on workflow stream results", async () => {
+    const memory = new Memory({ storage: new InMemoryStorageAdapter() });
+    const workflow = createWorkflow(
+      {
+        id: "stream-observer-surface",
+        name: "Stream Observer Surface",
+        input: z.object({ value: z.number() }),
+        result: z.object({ result: z.number() }),
+        memory,
+      },
+      andThen({
+        id: "multiply",
+        execute: async ({ data }) => ({ result: data.value * 3 }),
+      }),
+    );
+
+    const registry = WorkflowRegistry.getInstance();
+    registry.registerWorkflow(workflow);
+
+    const stream = workflow.stream({ value: 7 });
+    const watchedEvents: string[] = [];
+    const watchedAsyncEvents: string[] = [];
+
+    const unsubscribe = stream.watch((event) => {
+      watchedEvents.push(event.type);
+    });
+    const unsubscribeAsync = await stream.watchAsync((event) => {
+      watchedAsyncEvents.push(event.type);
+    });
+
+    for await (const _event of stream) {
+      // Drain the iterator to completion.
+    }
+
+    unsubscribe();
+    unsubscribeAsync();
+
+    expect(watchedEvents.length).toBeGreaterThan(0);
+    expect(watchedAsyncEvents.length).toBeGreaterThan(0);
+    expect(watchedEvents).toContain("workflow-start");
+    expect(watchedEvents).toContain("workflow-complete");
+    expect(watchedAsyncEvents).toContain("workflow-start");
+    expect(watchedAsyncEvents).toContain("workflow-complete");
+
+    const legacyState = await stream.streamLegacy().getWorkflowState();
+    expect(legacyState).toEqual(
+      expect.objectContaining({
+        id: stream.executionId,
+        status: "completed",
+      }),
+    );
+  });
+
+  it("should expose observeStream as readable stream on workflow stream results", async () => {
+    const memory = new Memory({ storage: new InMemoryStorageAdapter() });
+    const workflow = createWorkflow(
+      {
+        id: "stream-observe-readable",
+        name: "Stream Observe Readable",
+        input: z.object({ value: z.number() }),
+        result: z.object({ result: z.number() }),
+        memory,
+      },
+      andThen({
+        id: "multiply",
+        execute: async ({ data }) => ({ result: data.value * 2 }),
+      }),
+    );
+
+    const registry = WorkflowRegistry.getInstance();
+    registry.registerWorkflow(workflow);
+
+    const stream = workflow.stream({ value: 4 });
+    const reader = stream.observeStream().getReader();
+    const observedTypes: string[] = [];
+
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) {
+        break;
+      }
+      if (value) {
+        observedTypes.push(value.type);
+      }
+    }
+
+    expect(observedTypes).toContain("workflow-start");
+    expect(observedTypes).toContain("workflow-complete");
+    expect(await stream.result).toEqual({ result: 8 });
+  });
+
   it("should have usage with default values", async () => {
     const memory = new Memory({ storage: new InMemoryStorageAdapter() });
     const workflow = createWorkflow(

packages/core/src/workflow/core.ts

@@ -3116,6 +3116,14 @@ export function createWorkflow<
             resumedSuspendController.cancel(reason);
           },
           abort: () => streamController.abort(),
+          watch: (cb) => streamController.watch(cb),
+          watchAsync: (cb) => streamController.watchAsync(cb),
+          observeStream: () => streamController.observeStream(),
+          streamLegacy: () => ({
+            stream: streamController.observeStream(),
+            getWorkflowState: () =>
+              replayExecutionMemory.getWorkflowState(suspendedResult.executionId),
+          }),
           toUIMessageStreamResponse: eventToUIMessageStreamResponse(streamController),
           [Symbol.asyncIterator]: () => streamController.getStream(),
         };
@@ -3148,6 +3156,13 @@ export function createWorkflow<
         abort: () => {
           streamController.abort();
         },
+        watch: (cb) => streamController.watch(cb),
+        watchAsync: (cb) => streamController.watchAsync(cb),
+        observeStream: () => streamController.observeStream(),
+        streamLegacy: () => ({
+          stream: streamController.observeStream(),
+          getWorkflowState: () => replayExecutionMemory.getWorkflowState(executionId),
+        }),
         [Symbol.asyncIterator]: () => streamController.getStream(),
       };
 
@@ -3294,6 +3309,14 @@ export function createWorkflow<
             resumedSuspendController.cancel(reason);
           },
           abort: () => streamController.abort(),
+          watch: (cb) => streamController.watch(cb),
+          watchAsync: (cb) => streamController.watchAsync(cb),
+          observeStream: () => streamController.observeStream(),
+          streamLegacy: () => ({
+            stream: streamController.observeStream(),
+            getWorkflowState: () =>
+              streamExecutionMemory.getWorkflowState(suspendedResult.executionId),
+          }),
           toUIMessageStreamResponse: eventToUIMessageStreamResponse(streamController),
           [Symbol.asyncIterator]: () => streamController.getStream(),
         };
@@ -3327,6 +3350,13 @@ export function createWorkflow<
         abort: () => {
           streamController.abort();
         },
+        watch: (cb) => streamController.watch(cb),
+        watchAsync: (cb) => streamController.watchAsync(cb),
+        observeStream: () => streamController.observeStream(),
+        streamLegacy: () => ({
+          stream: streamController.observeStream(),
+          getWorkflowState: () => streamExecutionMemory.getWorkflowState(executionId),
+        }),
         // AsyncIterable implementation
         [Symbol.asyncIterator]: () => streamController.getStream(),
       };