diff --git a/.changeset/cool-geese-wave.md b/.changeset/cool-geese-wave.md new file mode 100644 index 000000000..e7a390092 --- /dev/null +++ b/.changeset/cool-geese-wave.md @@ -0,0 +1,9 @@ +--- +"@voltagent/core": minor +--- + +Add `startAsync()` to workflow and workflow chain APIs for fire-and-forget execution. + +`startAsync()` starts a workflow run in the background and returns `{ executionId, workflowId, startAt }` immediately. The run keeps existing execution semantics, respects provided `executionId`, and persists terminal states in memory for later inspection. + +Also adds workflow documentation updates with `startAsync()` usage examples in the workflow overview and streaming docs. diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 48d2226ed..5f931ecbf 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -29,6 +29,7 @@ export type { WorkflowHookStatus, WorkflowHooks, WorkflowStats, + WorkflowStartAsyncResult, WorkflowStateStore, WorkflowStateUpdater, WorkflowStepData, diff --git a/packages/core/src/workflow/chain.spec.ts b/packages/core/src/workflow/chain.spec.ts index 1059af047..13cd8453f 100644 --- a/packages/core/src/workflow/chain.spec.ts +++ b/packages/core/src/workflow/chain.spec.ts @@ -75,6 +75,48 @@ describe.sequential("workflow.run", () => { }); }); +describe.sequential("workflow.startAsync", () => { + beforeEach(() => { + const registry = WorkflowRegistry.getInstance(); + (registry as any).workflows.clear(); + }); + + it("should start chain workflow execution in background", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflowChain({ + id: "chain-start-async", + name: "Chain Start Async", + input: z.object({ value: z.number() }), + result: z.object({ result: z.number() }), + memory, + }).andThen({ + id: "double", + execute: async ({ data }) => ({ result: data.value * 2 }), + }); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow.toWorkflow()); + + const startResult = await workflow.startAsync({ value: 10 }); + + expect(startResult).toEqual({ + executionId: expect.any(String), + workflowId: "chain-start-async", + startAt: expect.any(Date), + }); + + let state = await memory.getWorkflowState(startResult.executionId); + for (let i = 0; i < 100 && state?.status !== "completed"; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + state = await memory.getWorkflowState(startResult.executionId); + } + + expect(state?.status).toBe("completed"); + expect(state?.output).toEqual({ result: 20 }); + }); +}); + describe.sequential("workflow writer API", () => { beforeEach(() => { const registry = WorkflowRegistry.getInstance(); diff --git a/packages/core/src/workflow/chain.ts b/packages/core/src/workflow/chain.ts index 4a258ad6f..49f9917a2 100644 --- a/packages/core/src/workflow/chain.ts +++ b/packages/core/src/workflow/chain.ts @@ -50,6 +50,7 @@ import type { WorkflowExecutionResult, WorkflowInput, WorkflowRunOptions, + WorkflowStartAsyncResult, WorkflowStateStore, WorkflowStateUpdater, WorkflowStepData, @@ -970,6 +971,21 @@ export class WorkflowChain< >; } + /** + * Start the workflow in the background without waiting for completion + */ + async startAsync( + input: WorkflowInput, + options?: WorkflowRunOptions, + ): Promise { + const workflow = createWorkflow( + this.config, + // @ts-expect-error - upstream types work and this is nature of how the createWorkflow function is typed using variadic args + ...this.steps, + ); + return workflow.startAsync(input, options); + } + /** * Execute the workflow with streaming support */ diff --git a/packages/core/src/workflow/core.spec.ts b/packages/core/src/workflow/core.spec.ts index 1bfa94eae..ad1e1eb7d 100644 --- a/packages/core/src/workflow/core.spec.ts +++ b/packages/core/src/workflow/core.spec.ts @@ -463,6 +463,207 @@ describe.sequential("workflow streaming", () => { }); }); +describe.sequential("workflow.startAsync", () => { + beforeEach(() => { + const registry = WorkflowRegistry.getInstance(); + (registry as any).workflows.clear(); + }); + + it("should return immediately and complete in the background", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + let releaseStep: (() => void) | undefined; + const stepGate = new Promise((resolve) => { + releaseStep = resolve; + }); + + const workflow = createWorkflow( + { + id: "start-async-background", + name: "Start Async Background", + input: z.object({ value: z.number() }), + result: z.object({ result: z.number() }), + memory, + }, + andThen({ + id: "wait-for-release", + execute: async ({ data }) => { + await stepGate; + return { result: data.value * 2 }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const startResult = await workflow.startAsync({ value: 21 }); + + expect(startResult).toEqual({ + executionId: expect.any(String), + workflowId: "start-async-background", + startAt: expect.any(Date), + }); + + let runningState = await memory.getWorkflowState(startResult.executionId); + for (let i = 0; i < 100 && runningState?.status !== "running"; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + runningState = await memory.getWorkflowState(startResult.executionId); + } + expect(runningState?.status).toBe("running"); + + releaseStep?.(); + + let completedState = await memory.getWorkflowState(startResult.executionId); + for (let i = 0; i < 100 && completedState?.status !== "completed"; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + completedState = await memory.getWorkflowState(startResult.executionId); + } + + expect(completedState?.status).toBe("completed"); + expect(completedState?.output).toEqual({ result: 42 }); + }); + + it("should persist error state when background execution fails", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflow( + { + id: "start-async-error", + name: "Start Async Error", + input: z.object({ value: z.number() }), + result: z.object({ result: z.number() }), + memory, + }, + andThen({ + id: "throw-error", + execute: async () => { + throw new Error("startAsync failure"); + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const startResult = await workflow.startAsync({ value: 1 }); + let erroredState = await memory.getWorkflowState(startResult.executionId); + + for (let i = 0; i < 100 && erroredState?.status !== "error"; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + erroredState = await memory.getWorkflowState(startResult.executionId); + } + + expect(erroredState?.status).toBe("error"); + expect(erroredState?.metadata).toEqual( + expect.objectContaining({ + errorMessage: "startAsync failure", + }), + ); + }); + + it("should respect executionId passed in options", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflow( + { + id: "start-async-execution-id", + name: "Start Async Execution ID", + input: z.object({ value: z.number() }), + result: z.object({ value: z.number() }), + memory, + }, + andThen({ + id: "echo", + execute: async ({ data }) => data, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const executionId = "execution-id-start-async"; + const startResult = await workflow.startAsync( + { value: 5 }, + { + executionId, + }, + ); + + expect(startResult.executionId).toBe(executionId); + + let state = await memory.getWorkflowState(executionId); + for (let i = 0; i < 100 && state?.status !== "completed"; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + state = await memory.getWorkflowState(executionId); + } + + expect(state?.status).toBe("completed"); + expect(state?.output).toEqual({ value: 5 }); + }); + + it("should reject resumeFrom options to avoid overwriting suspended runs", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflow( + { + id: "start-async-reject-resume-from", + name: "Start Async Reject Resume From", + input: z.object({ value: z.number() }), + result: z.object({ value: z.number() }), + memory, + }, + andThen({ + id: "echo", + execute: async ({ data }) => data, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const executionId = "suspended-run"; + const suspendedAt = new Date(); + await memory.setWorkflowState(executionId, { + id: executionId, + workflowId: "start-async-reject-resume-from", + workflowName: "Start Async Reject Resume From", + status: "suspended", + input: { value: 10 }, + workflowState: { stage: "waiting" }, + suspension: { + suspendedAt, + reason: "awaiting-input", + stepIndex: 0, + }, + metadata: { + marker: "preserve-me", + }, + createdAt: suspendedAt, + updatedAt: suspendedAt, + }); + + await expect( + workflow.startAsync( + { value: 5 }, + { + resumeFrom: { + executionId, + resumeStepIndex: 0, + }, + }, + ), + ).rejects.toThrow("startAsync does not support resumeFrom"); + + const persisted = await memory.getWorkflowState(executionId); + expect(persisted?.status).toBe("suspended"); + expect(persisted?.metadata).toEqual( + expect.objectContaining({ + marker: "preserve-me", + }), + ); + }); +}); + describe.sequential("workflow memory defaults", () => { beforeEach(() => { const registry = AgentRegistry.getInstance(); diff --git a/packages/core/src/workflow/core.ts b/packages/core/src/workflow/core.ts index 5202646a6..89fe52bbc 100644 --- a/packages/core/src/workflow/core.ts +++ b/packages/core/src/workflow/core.ts @@ -44,6 +44,7 @@ import type { WorkflowInput, WorkflowResult, WorkflowRunOptions, + WorkflowStartAsyncResult, WorkflowStateStore, WorkflowStateUpdater, WorkflowStreamResult, @@ -926,36 +927,73 @@ export function createWorkflow< throw error; // Re-throw to prevent creating a new execution } } else { - // Create new execution - ALWAYS create state directly (like Agent does) - - // 1. Create workflow state in Memory V2 (workflow's own memory) - const workflowState = { - id: executionId, - workflowId: id, - workflowName: name, - status: "running" as const, - input, - context: options?.context ? Array.from(contextMap.entries()) : undefined, - workflowState: workflowStateStore, - userId: options?.userId, - conversationId: options?.conversationId, - metadata: { - ...(optionMetadata ?? {}), - traceId: rootSpan.spanContext().traceId, - spanId: rootSpan.spanContext().spanId, - }, - createdAt: new Date(), - updatedAt: new Date(), - }; + if (options?.skipStateInit) { + // startAsync pre-creates running state; only enrich metadata/context here + try { + const existingWorkflowState = await executionMemory.getWorkflowState(executionId); + if (!existingWorkflowState) { + throw new Error(`Workflow state ${executionId} not found`); + } - try { - await executionMemory.setWorkflowState(executionId, workflowState); - runLogger.trace(`Created workflow state in Memory V2 for ${executionId}`); - } catch (error) { - runLogger.error("Failed to create workflow state in Memory V2:", { error }); - throw new Error( - `Failed to create workflow state: ${error instanceof Error ? error.message : String(error)}`, - ); + await executionMemory.updateWorkflowState(executionId, { + status: "running", + input: existingWorkflowState.input ?? input, + context: + options?.context !== undefined + ? Array.from(contextMap.entries()) + : existingWorkflowState.context, + workflowState: + options?.workflowState !== undefined + ? workflowStateStore + : (existingWorkflowState.workflowState ?? workflowStateStore), + userId: options?.userId ?? existingWorkflowState.userId, + conversationId: options?.conversationId ?? existingWorkflowState.conversationId, + metadata: { + ...(existingWorkflowState.metadata ?? {}), + ...(optionMetadata ?? {}), + traceId: rootSpan.spanContext().traceId, + spanId: rootSpan.spanContext().spanId, + }, + updatedAt: new Date(), + }); + + runLogger.trace(`Updated pre-created workflow state in Memory V2 for ${executionId}`); + } catch (error) { + runLogger.error("Failed to update pre-created workflow state in Memory V2:", { error }); + throw new Error( + `Failed to update workflow state: ${error instanceof Error ? error.message : String(error)}`, + ); + } + } else { + // Create new execution - ALWAYS create state directly (like Agent does) + const workflowState = { + id: executionId, + workflowId: id, + workflowName: name, + status: "running" as const, + input, + context: options?.context ? Array.from(contextMap.entries()) : undefined, + workflowState: workflowStateStore, + userId: options?.userId, + conversationId: options?.conversationId, + metadata: { + ...(optionMetadata ?? {}), + traceId: rootSpan.spanContext().traceId, + spanId: rootSpan.spanContext().spanId, + }, + createdAt: new Date(), + updatedAt: new Date(), + }; + + try { + await executionMemory.setWorkflowState(executionId, workflowState); + runLogger.trace(`Created workflow state in Memory V2 for ${executionId}`); + } catch (error) { + runLogger.error("Failed to create workflow state in Memory V2:", { error }); + throw new Error( + `Failed to create workflow state: ${error instanceof Error ? error.message : String(error)}`, + ); + } } } @@ -2130,6 +2168,124 @@ export function createWorkflow< // Simply call executeInternal which handles everything without stream return executeInternal(input, options); }, + startAsync: async ( + input: WorkflowInput, + options?: WorkflowRunOptions, + ): Promise => { + const executionMemory = options?.memory ?? defaultMemory; + + if (options?.resumeFrom) { + const resumeExecutionId = options.resumeFrom.executionId; + const resumeState = await executionMemory.getWorkflowState(resumeExecutionId); + if (resumeState?.status === "suspended") { + throw new Error( + `startAsync does not support resumeFrom for suspended execution ${resumeExecutionId}. Use workflow.run(...) or workflow.stream(...) with resumeFrom.`, + ); + } + + throw new Error( + "startAsync does not support resumeFrom. Use workflow.run(...) or workflow.stream(...) with resumeFrom.", + ); + } + + const executionId = options?.executionId ?? randomUUID(); + const startAt = new Date(); + const contextEntries = + options?.context instanceof Map + ? Array.from(options.context.entries()) + : options?.context + ? Array.from(Object.entries(options.context)) + : undefined; + const optionMetadata = + options?.metadata && + typeof options.metadata === "object" && + !Array.isArray(options.metadata) + ? options.metadata + : undefined; + + await executionMemory.setWorkflowState(executionId, { + id: executionId, + workflowId: id, + workflowName: name, + status: "running", + input, + context: contextEntries, + workflowState: options?.workflowState ?? {}, + userId: options?.userId, + conversationId: options?.conversationId, + metadata: { + ...(optionMetadata ?? {}), + }, + createdAt: startAt, + updatedAt: startAt, + }); + + const executionOptions: WorkflowRunOptions = { + ...options, + executionId, + skipStateInit: true, + }; + + executeInternal(input, executionOptions) + .catch(async (error) => { + const errorMessage = error instanceof Error ? error.message : String(error); + + logger.warn("startAsync execution failed before terminal handling", { + executionId, + error, + }); + + try { + const existingState = await executionMemory.getWorkflowState(executionId); + if (existingState) { + await executionMemory.updateWorkflowState(executionId, { + status: "error", + metadata: { + ...(existingState.metadata ?? {}), + errorMessage, + }, + updatedAt: new Date(), + }); + return; + } + + await executionMemory.setWorkflowState(executionId, { + id: executionId, + workflowId: id, + workflowName: name, + status: "error", + input, + context: contextEntries, + workflowState: options?.workflowState ?? {}, + userId: options?.userId, + conversationId: options?.conversationId, + metadata: { + ...(optionMetadata ?? {}), + errorMessage, + }, + createdAt: startAt, + updatedAt: new Date(), + }); + } catch (persistenceError) { + logger.warn("Failed to persist startAsync background failure", { + executionId, + error: persistenceError, + }); + } + }) + .catch((handlerError) => { + logger.error("Unexpected error while handling startAsync background failure", { + executionId, + error: handlerError, + }); + }); + + return { + executionId, + workflowId: id, + startAt, + }; + }, stream: (input: WorkflowInput, options?: WorkflowRunOptions) => { // Create stream controller for this execution const streamController = new WorkflowStreamController(); diff --git a/packages/core/src/workflow/index.ts b/packages/core/src/workflow/index.ts index c3cec5578..3f0627712 100644 --- a/packages/core/src/workflow/index.ts +++ b/packages/core/src/workflow/index.ts @@ -30,6 +30,7 @@ export type { WorkflowRetryConfig, WorkflowRunOptions, WorkflowResumeOptions, + WorkflowStartAsyncResult, WorkflowSuspensionMetadata, WorkflowSuspendController, WorkflowStateStore, diff --git a/packages/core/src/workflow/types.ts b/packages/core/src/workflow/types.ts index d81d2a48b..3162c0b18 100644 --- a/packages/core/src/workflow/types.ts +++ b/packages/core/src/workflow/types.ts @@ -202,6 +202,24 @@ export interface WorkflowStreamResult< abort: () => void; } +/** + * Result returned when a workflow execution is started asynchronously + */ +export interface WorkflowStartAsyncResult { + /** + * Unique execution ID for this workflow run + */ + executionId: string; + /** + * The workflow ID + */ + workflowId: string; + /** + * When the async execution was started + */ + startAt: Date; +} + export interface WorkflowRetryConfig { /** * Number of retry attempts for a step when it throws an error @@ -293,6 +311,11 @@ export interface WorkflowRunOptions { * Optional agent instance to supply to workflow guardrails */ guardrailAgent?: Agent; + /** + * Internal flag to skip initial state creation when it is already persisted + * @internal + */ + skipStateInit?: boolean; } export interface WorkflowResumeOptions { @@ -612,6 +635,16 @@ export type Workflow< input: WorkflowInput, options?: WorkflowRunOptions, ) => WorkflowStreamResult; + /** + * Start the workflow in the background without waiting for completion + * @param input - The input to the workflow + * @param options - Options for the workflow execution + * @returns Async start metadata with execution ID + */ + startAsync: ( + input: WorkflowInput, + options?: WorkflowRunOptions, + ) => Promise; /** * Create a WorkflowSuspendController that can be used to suspend the workflow * @returns A WorkflowSuspendController instance diff --git a/website/docs/workflows/overview.md b/website/docs/workflows/overview.md index 076bd13f9..c56e975e7 100644 --- a/website/docs/workflows/overview.md +++ b/website/docs/workflows/overview.md @@ -255,7 +255,7 @@ When you use `createWorkflowChain`, you are creating a **builder** object (`Work This builder is not the final, runnable workflow itself. It's the blueprint. -There are two ways to run your workflow: +There are three ways to run your workflow: **1. The Shortcut: `.run()`** @@ -266,7 +266,38 @@ Calling `.run()` directly on the chain is a convenient shortcut. Behind the scen const result = await workflow.run({ name: "World" }); ``` -**2. The Reusable Way: `.toWorkflow()`** +**2. Fire-and-Forget: `.startAsync()`** + +Use `.startAsync()` when you want to trigger a workflow and continue immediately without waiting for completion. It returns execution metadata (`executionId`, `workflowId`, `startAt`) right away. + +```typescript +import { InMemoryStorageAdapter, Memory, createWorkflowChain } from "@voltagent/core"; +import { z } from "zod"; + +const workflowMemory = new Memory({ + storage: new InMemoryStorageAdapter(), +}); + +const greeterChain = createWorkflowChain({ + id: "async-greeter", + name: "Async Greeter", + input: z.object({ name: z.string() }), + result: z.object({ greeting: z.string() }), + memory: workflowMemory, +}).andThen({ + id: "create-greeting", + execute: async ({ data }) => ({ greeting: `Hello, ${data.name}!` }), +}); + +const started = await greeterChain.startAsync({ name: "Alice" }); +console.log(started.executionId, started.startAt); // Track this run later + +// Query execution state later from workflow memory +const state = await greeterChain.toWorkflow().memory.getWorkflowState(started.executionId); +console.log(state?.status); // running | completed | suspended | cancelled | error +``` + +**3. The Reusable Way: `.toWorkflow()`** The `WorkflowChain` builder has a `.toWorkflow()` method that converts your blueprint into a permanent, reusable `Workflow` object. You can store this object, pass it to other functions, or run it multiple times without rebuilding the chain. diff --git a/website/docs/workflows/streaming.md b/website/docs/workflows/streaming.md index bd91b13bc..e6eaa0a33 100644 --- a/website/docs/workflows/streaming.md +++ b/website/docs/workflows/streaming.md @@ -42,10 +42,11 @@ Workflows emit these event types during execution: ### Consuming the Stream -VoltAgent provides two methods for workflow execution: +VoltAgent provides three methods for workflow execution: -- `.run()` - Standard execution without streaming - `.stream()` - Real-time execution with event streaming +- `.run()` - Standard execution without streaming +- `.startAsync()` - Fire-and-forget execution (returns immediately) ```typescript // Method 1: Stream execution for real-time events @@ -75,6 +76,14 @@ console.log("Final result:", result); // Method 2: Standard execution without streaming const execution = await workflow.run(input); console.log("Result:", execution.result); + +// Method 3: Fire-and-forget execution +const started = await workflow.startAsync(input); +console.log("Started execution:", started.executionId); + +// Later, inspect status/output from workflow memory +const state = await workflow.memory.getWorkflowState(started.executionId); +console.log("Current status:", state?.status); ``` ## Writer API @@ -766,15 +775,28 @@ interface WorkflowStreamResult } ``` +### WorkflowStartAsyncResult + +Returned by `.startAsync()` method - starts in the background and returns immediately: + +```typescript +interface WorkflowStartAsyncResult { + executionId: string; + workflowId: string; + startAt: Date; +} +``` + ### Key Differences -| Feature | `.run()` | `.stream()` | -| ---------------- | ------------------------- | ---------------------- | -| Returns | `WorkflowExecutionResult` | `WorkflowStreamResult` | -| Event streaming | No | Yes (AsyncIterable) | -| Field resolution | Immediate | Promise-based | -| Use case | Simple execution | Real-time monitoring | -| Resume behavior | New execution | Same stream continues | +| Feature | `.run()` | `.startAsync()` | `.stream()` | +| ---------------- | ------------------------- | -------------------------- | ---------------------- | +| Returns | `WorkflowExecutionResult` | `WorkflowStartAsyncResult` | `WorkflowStreamResult` | +| Waits for finish | Yes | No | No | +| Event streaming | No | No | Yes (AsyncIterable) | +| Field resolution | Immediate | Immediate metadata | Promise-based | +| Use case | Simple execution | Background trigger | Real-time monitoring | +| Resume behavior | New execution | Via stored execution state | Same stream continues | ### UsageInfo