feat: recoverable data with error handling (#1476)

.changeset/cool-cows-wonder.md

@@ -0,0 +1,5 @@
+---
+"@llamaindex/workflow": patch
+---
+
+feat: recoverable context with error handling

packages/workflow/src/workflow-context.ts

@@ -453,6 +453,9 @@ export class WorkflowContext<Start = string, Stop = string, Data = unknown>
                       }),
                     )
                     .catch((err) => {
+                      // when the step raise an error, should go back to the previous step
+                      this.#sendEvent(event);
+                      isPendingEvents.add(event);
                       controller.error(err);
                     });
                 }

unit/workflow/workflow.test.ts

@@ -884,3 +884,83 @@ describe("snapshot", async () => {
     expect(fn).toHaveBeenCalledTimes(1);
   });
 });
+
+describe("error", () => {
+  test("error in handler", async () => {
+    const myFlow = new Workflow<boolean, string, string>({ verbose: true });
+    myFlow.addStep(
+      {
+        inputs: [StartEvent<string>],
+        outputs: [StopEvent<string>],
+      },
+      async ({ data }) => {
+        if (!data) {
+          throw new Error("Something went wrong");
+        } else {
+          return new StopEvent(`Hello ${data}!`);
+        }
+      },
+    );
+    await expect(myFlow.run("world")).rejects.toThrow("Something went wrong");
+    {
+      const context = myFlow.run("world");
+      try {
+        for await (const _ of context) {
+          // do nothing
+        }
+      } catch (error) {
+        expect(error).toBeInstanceOf(Error);
+        expect((error as Error).message).toBe("Something went wrong");
+        const snapshot = context.snapshot();
+        const newContext = myFlow.recover(snapshot).with(true);
+        expect((await newContext).data).toBe("Hello true!");
+      }
+    }
+  });
+
+  test("recover in the middle of workflow", async () => {
+    const myFlow = new Workflow<string | undefined, string, string>({
+      verbose: true,
+    });
+
+    class AEvent extends WorkflowEvent<string> {}
+
+    myFlow.addStep(
+      {
+        inputs: [StartEvent<string>],
+        outputs: [AEvent],
+      },
+      async ({ data }) => {
+        if (data !== undefined) {
+          throw new Error("Something went wrong");
+        }
+        return new AEvent("world");
+      },
+    );
+    myFlow.addStep(
+      {
+        inputs: [AEvent],
+        outputs: [StopEvent],
+      },
+      async ({ data }, ev) => {
+        if (data === undefined) {
+          throw new Error("Something went wrong");
+        }
+        return new StopEvent(`Hello, ${data}!`);
+      },
+    );
+    // no context, so will throw error
+    const context = myFlow.run("world");
+    try {
+      for await (const _ of context) {
+        // do nothing
+      }
+    } catch (error) {
+      expect(error).toBeInstanceOf(Error);
+      expect((error as Error).message).toBe("Something went wrong");
+      const snapshot = context.snapshot();
+      const newContext = myFlow.recover(snapshot).with("Recovered Data");
+      expect((await newContext).data).toBe("Hello, Recovered Data!");
+    }
+  });
+});