deer-workflow 动态 Agent 工作流运行时

90export async function workflow<TOutput = unknown, TArgs = unknown>(
91  target: WorkflowTarget,
92  args?: TArgs,
93): Promise<TOutput> {
94  const parentContext = getWorkflowContext();
95  const depth = parentContext ? parentContext.depth + 1 : 0;
96
97  if (depth > MAX_NESTED_WORKFLOW_DEPTH) {
98    throw new WorkflowNestingError(depth);
99  }
100
101  const scriptPath = resolveWorkflowPath(target, parentContext);
102  const context: WorkflowExecutionContext<TArgs | undefined> = {
103    id: crypto.randomUUID(),
104    parentId: parentContext?.id,
105    depth,
106    scriptPath,
107    args,
108  };
109
110  return runInWorkflowContext(context, async () => {
111    const startedAt = performance.now();
112    emitWorkflowEvent({
113      type: "workflow:start",
114      ...toWorkflowEventContext(context),
115    });
116
117    try {
118      const definition = await loadWorkflowDefinition<
119        TArgs | undefined,
120        TOutput
121      >(scriptPath);
122      if (definition.meta !== undefined) {
123        emitWorkflowEvent({
124          type: "workflow:meta",
125          ...toWorkflowEventContext(context),
126          meta: definition.meta,
127        });
128      }
129      const result = await definition.handler(args, context);
130      endCurrentPhase(context);
131      emitWorkflowEvent({
132        type: "workflow:end",
133        ...toWorkflowEventContext(context),
134        durationMs: elapsedMilliseconds(startedAt),
135      });
136      return result;
137    } catch (error) {
138      endCurrentPhase(context);
139      emitWorkflowEvent({
140        type: "workflow:error",
141        ...toWorkflowEventContext(context),
142        durationMs: elapsedMilliseconds(startedAt),
143        error: serializeWorkflowError(error),
144      });
145      throw error;
146    }
147  });

22export async function parallel<const TTasks extends readonly ParallelTask[]>(
23  tasks: TTasks,
24): Promise<ParallelResults<TTasks>> {
25  const results = await Promise.all(
26    Array.from(tasks, async (task) => {
27      try {
28        return await task();
29      } catch {
30        return null;
31      }
32    }),
33  );
34
35  return results as ParallelResults<TTasks>;
36}

78export async function pipeline<TOriginal>(
79  items: readonly TOriginal[],
80  ...stages: RuntimePipelineStage<TOriginal>[]
81): Promise<Array<unknown | null>> {
82  return Promise.all(
83    Array.from(items, async (original, index) => {
84      let value: unknown = original;
85
86      try {
87        for (const stage of stages) {
88          value = await stage(value as never, original, index);
89        }
90        return value;
91      } catch {
92        return null;
93      }
94    }),
95  );
96}

52  run<TOutput = unknown, TArgs = unknown>(
53    target: WorkflowTarget,
54    args?: TArgs,
55  ): Promise<TOutput> {
56    if (this.#disposed) {
57      throw new Error("WorkflowRunner has been disposed.");
58    }
59
60    return runWithWorkflowEventEmitter(this.events, () =>
61      withLogSink(
62        (message) => {
63          const context = getWorkflowContext();
64          if (!context) {
65            return;
66          }
67
68          this.events.emit({
69            type: "log",
70            ...toWorkflowEventContext(context),
71            message,
72            ...(context.phase === undefined ? {} : { phase: context.phase }),
73          });
74        },
75        () => workflow<TOutput, TArgs>(target, args),
76      ),
77    );
78  }

14export class WorkflowEventEmitter {
15  readonly #listeners = new Set<WorkflowEventListener>();
16  #sequence = 0;
17
18  /**
19   * Subscribes to every Workflow event.
20   *
21   * @param listener - Callback invoked for each event.
22   * @returns A function that removes this subscription.
23   */
24  on(listener: WorkflowEventListener): () => void {
25    this.#listeners.add(listener);
26    return () => this.off(listener);
27  }
28
29  /**
30   * Removes a previously registered listener.
31   *
32   * @param listener - Callback originally passed to {@link on}.
33   * @returns `true` when the listener was registered.
34   */
35  off(listener: WorkflowEventListener): boolean {
36    return this.#listeners.delete(listener);
37  }
38
39  /**
40   * Adds envelope metadata and synchronously emits an event.
41   *
42   * @param input - Typed event payload without sequence or timestamp.
43   * @returns The complete immutable event delivered to listeners.
44   */
45  emit(input: WorkflowEventInput): WorkflowEvent {
46    const event = Object.freeze({
47      ...input,
48      sequence: ++this.#sequence,
49      timestamp: new Date().toISOString(),
50    }) as WorkflowEvent;
51
52    for (const listener of [...this.#listeners]) {
53      listener(event);
54    }
55
56    return event;
57  }

73  test("writes ordered lifecycle, phase, and log events as JSON Lines", async () => {
74    const lines: string[] = [];
75    const runner = new WorkflowRunner({
76      logWriter: (line) => lines.push(line),
77    });
78
79    const result = await runner.run<string>(successfulWorkflowPath);
80    const events = lines.map(parseEvent);
81
82    expect(result).toBe("done");
83    expect(events.map((event) => event.type)).toEqual([
84      "workflow:start",
85      "workflow:meta",
86      "workflow:phase:start",
87      "log",
88      "workflow:phase:end",
89      "workflow:phase:start",
90      "log",
91      "workflow:phase:end",
92      "workflow:end",
93    ]);
94    expect(events.map((event) => event.sequence)).toEqual([
95      1, 2, 3, 4, 5, 6, 7, 8, 9,
96    ]);
97    expect(new Set(events.map((event) => event.workflowId)).size).toBe(1);
98
99    const metadata = events[1];
100    expect(metadata?.type).toBe("workflow:meta");
101    if (metadata?.type === "workflow:meta") {
102      expect(metadata.meta.name).toBe("successful");
103      expect(metadata.meta.phases.map((phase) => phase.title)).toEqual([
104        "Research",
105        "Synthesis",
106      ]);
107      expect(metadata.meta.exampleArgs).toEqual({ topic: "Agent runtimes" });
108    }
109
110    const firstLog = events[3];
111    expect(firstLog?.type).toBe("log");
112    if (firstLog?.type === "log") {
113      expect(firstLog.phase).toBe("Research");
114      expect(firstLog.message).toBe("Collecting sources");
115    }
116
117    runner.dispose();
118  });

39  test("isolates an item failure and skips its remaining stages", async () => {
40    const secondStageItems: number[] = [];
41
42    const results = await pipeline(
43      [1, 2, 3],
44      (value) => {
45        if (value === 2) {
46          throw new Error("item failure");
47        }
48        return value * 10;
49      },
50      (value, original) => {
51        secondStageItems.push(original);
52        return value + 1;
53      },
54    );
55
56    expect(results).toEqual([11, null, 31]);
57    expect(secondStageItems).toEqual([1, 3]);
58  });