diff --git a/.changeset/fresh-otters-melt.md b/.changeset/fresh-otters-melt.md new file mode 100644 index 000000000..102965f17 --- /dev/null +++ b/.changeset/fresh-otters-melt.md @@ -0,0 +1,85 @@ +--- +"@voltagent/core": minor +--- + +feat: add workflow execution primitives (`bail`, `abort`, `getStepResult`, `getInitData`) + +### What's New + +Step execution context now includes four new primitives: + +- `bail(result?)`: complete the workflow early with a custom final result +- `abort()`: cancel the workflow immediately +- `getStepResult(stepId)`: get a prior step output directly (returns `null` if not available) +- `getInitData()`: get the original workflow input (stable across resume paths) + +These primitives are available in all step contexts, including nested step flows. + +### Example: Early Complete with `bail` + +```ts +const workflow = createWorkflowChain({ + id: "bail-demo", + input: z.object({ amount: z.number() }), + result: z.object({ status: z.string() }), +}) + .andThen({ + id: "risk-check", + execute: async ({ data, bail }) => { + if (data.amount > 10_000) { + bail({ status: "rejected" }); + } + return { status: "approved" }; + }, + }) + .andThen({ + id: "never-runs-on-bail", + execute: async () => ({ status: "approved" }), + }); +``` + +### Example: Cancel with `abort` + +```ts +const workflow = createWorkflowChain({ + id: "abort-demo", + input: z.object({ requestId: z.string() }), + result: z.object({ done: z.boolean() }), +}) + .andThen({ + id: "authorization", + execute: async ({ abort }) => { + abort(); // terminal status: cancelled + }, + }) + .andThen({ + id: "never-runs-on-abort", + execute: async () => ({ done: true }), + }); +``` + +### Example: Use `getStepResult` + `getInitData` + +```ts +const workflow = createWorkflowChain({ + id: "introspection-demo", + input: z.object({ userId: z.string(), value: z.number() }), + result: z.object({ total: z.number(), userId: z.string() }), +}) + .andThen({ + id: "step-1", + execute: async ({ data }) => ({ partial: data.value + 1 }), + }) + .andThen({ + id: "step-2", + execute: async ({ getStepResult, getInitData }) => { + const s1 = getStepResult<{ partial: number }>("step-1"); + const init = getInitData(); + + return { + total: (s1?.partial ?? 0) + init.value, + userId: init.userId, + }; + }, + }); +``` diff --git a/packages/core/src/test-utils/mocks/workflows.ts b/packages/core/src/test-utils/mocks/workflows.ts index a2db2cef9..27869cfea 100644 --- a/packages/core/src/test-utils/mocks/workflows.ts +++ b/packages/core/src/test-utils/mocks/workflows.ts @@ -24,7 +24,19 @@ export function createMockWorkflowExecuteContext( data: overrides.data ?? ({} as DangerouslyAllowAny), state: overrides.state ?? ({} as DangerouslyAllowAny), getStepData: overrides.getStepData ?? (() => undefined), + getStepResult: overrides.getStepResult ?? (() => null), + getInitData: overrides.getInitData ?? (() => ({}) as DangerouslyAllowAny), suspend: overrides.suspend ?? vi.fn(), + bail: + overrides.bail ?? + (() => { + throw new Error("WORKFLOW_BAIL_NOT_CONFIGURED"); + }), + abort: + overrides.abort ?? + (() => { + throw new Error("WORKFLOW_ABORT_NOT_CONFIGURED"); + }), workflowState: overrides.workflowState ?? {}, setWorkflowState: (() => undefined) as MockWorkflowExecuteContext["setWorkflowState"], logger: overrides.logger ?? { diff --git a/packages/core/src/workflow/core.spec-d.ts b/packages/core/src/workflow/core.spec-d.ts index 2debb2bf2..4b8ba2d80 100644 --- a/packages/core/src/workflow/core.spec-d.ts +++ b/packages/core/src/workflow/core.spec-d.ts @@ -104,7 +104,11 @@ describe("non-chaining API type inference", () => { (update: WorkflowStateUpdater) => void >(); expectTypeOf().toBeFunction(); + expectTypeOf().toBeFunction(); + expectTypeOf().toBeFunction(); expectTypeOf().toBeFunction(); + expectTypeOf().toBeFunction(); + expectTypeOf().toBeFunction(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); }); diff --git a/packages/core/src/workflow/core.spec.ts b/packages/core/src/workflow/core.spec.ts index 6bfd21c9e..7cc7f5e81 100644 --- a/packages/core/src/workflow/core.spec.ts +++ b/packages/core/src/workflow/core.spec.ts @@ -7,7 +7,7 @@ import { InMemoryStorageAdapter } from "../memory/adapters/storage/in-memory"; import { AgentRegistry } from "../registries/agent-registry"; import { VOLTAGENT_RESTART_CHECKPOINT_KEY, createWorkflow } from "./core"; import { WorkflowRegistry } from "./registry"; -import { andAgent, andThen } from "./steps"; +import { andAgent, andThen, andWhen } from "./steps"; describe.sequential("workflow.run", () => { beforeEach(() => { @@ -333,6 +333,283 @@ describe.sequential("workflow.run", () => { }), ); }); + + it("should support bail(result) to complete early with custom output", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + let finalStepReached = false; + + const workflow = createWorkflow( + { + id: "execution-primitives-bail", + name: "Execution Primitives Bail", + input: z.object({ value: z.number() }), + result: z.object({ final: z.number() }), + memory, + }, + andThen({ + id: "prepare", + execute: async ({ data }) => ({ prepared: data.value + 1 }), + }), + andThen({ + id: "bail-step", + execute: async ({ bail, getStepResult }) => { + const prepared = getStepResult<{ prepared: number }>("prepare"); + bail({ final: (prepared?.prepared ?? 0) * 10 }); + }, + }), + andThen({ + id: "should-not-run", + execute: async () => { + finalStepReached = true; + return { final: -1 }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const result = await workflow.run({ value: 2 }); + + expect(result.status).toBe("completed"); + expect(result.result).toEqual({ final: 30 }); + expect(finalStepReached).toBe(false); + + const persisted = await memory.getWorkflowState(result.executionId); + expect(persisted?.status).toBe("completed"); + expect(persisted?.output).toEqual({ final: 30 }); + }); + + it("should support bail() without result to complete early without output", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + let finalStepReached = false; + + const workflow = createWorkflow( + { + id: "execution-primitives-bail-no-result", + name: "Execution Primitives Bail No Result", + input: z.object({ value: z.number() }), + result: z.object({ final: z.number() }), + memory, + }, + andThen({ + id: "prepare", + execute: async ({ data }) => ({ prepared: data.value + 1 }), + }), + andThen({ + id: "bail-step", + execute: async ({ bail }) => { + bail(); + }, + }), + andThen({ + id: "should-not-run", + execute: async () => { + finalStepReached = true; + return { final: -1 }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const result = await workflow.run({ value: 2 }); + + expect(result.status).toBe("completed"); + expect(result.result).toBeNull(); + expect(finalStepReached).toBe(false); + + const persisted = await memory.getWorkflowState(result.executionId); + expect(persisted?.status).toBe("completed"); + expect(persisted?.output).toBeNull(); + }); + + it("should support abort() to cancel execution and persist cancellation metadata", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + let finalStepReached = false; + + const workflow = createWorkflow( + { + id: "execution-primitives-abort", + name: "Execution Primitives Abort", + input: z.object({ value: z.number() }), + result: z.object({ value: z.number() }), + memory, + }, + andThen({ + id: "abort-step", + execute: async ({ abort }) => { + abort(); + }, + }), + andThen({ + id: "should-not-run", + execute: async () => { + finalStepReached = true; + return { value: -1 }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const result = await workflow.run({ value: 1 }); + + expect(result.status).toBe("cancelled"); + expect(result.result).toBeNull(); + expect(result.cancellation?.reason).toBe("Workflow aborted by step: abort-step"); + expect(finalStepReached).toBe(false); + + const persisted = await memory.getWorkflowState(result.executionId); + expect(persisted?.status).toBe("cancelled"); + expect(persisted?.metadata).toEqual( + expect.objectContaining({ + cancellationReason: "Workflow aborted by step: abort-step", + }), + ); + }); + + it("should provide getStepResult and getInitData helpers in step context", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflow( + { + id: "execution-primitives-step-result-init", + name: "Execution Primitives Step Result Init", + input: z.object({ value: z.number() }), + result: z.object({ + computed: z.number(), + unknownIsNull: z.boolean(), + initValue: z.number(), + }), + memory, + }, + andThen({ + id: "step-1", + execute: async ({ data }) => ({ stepValue: data.value + 1 }), + }), + andThen({ + id: "step-2", + execute: async ({ getStepResult, getInitData }) => { + const prior = getStepResult<{ stepValue: number }>("step-1"); + const unknown = getStepResult("missing-step"); + const init = getInitData<{ value: number }>(); + + return { + computed: (prior?.stepValue ?? 0) + init.value, + unknownIsNull: unknown === null, + initValue: init.value, + }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const result = await workflow.run({ value: 5 }); + + expect(result.status).toBe("completed"); + expect(result.result).toEqual({ + computed: 11, + unknownIsNull: true, + initValue: 5, + }); + }); + + it("should keep getInitData stable across suspend/resume", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + + const workflow = createWorkflow( + { + id: "execution-primitives-init-resume", + name: "Execution Primitives Init Resume", + input: z.object({ requestId: z.string() }), + result: z.object({ requestId: z.string(), approved: z.boolean() }), + memory, + }, + andThen({ + id: "approval-gate", + resumeSchema: z.object({ approved: z.boolean() }), + execute: async ({ data, suspend, resumeData }) => { + if (resumeData) { + return { + ...data, + approved: resumeData.approved, + }; + } + + await suspend("Manual approval required"); + }, + }), + andThen({ + id: "finalize", + execute: async ({ data, getInitData }) => { + const init = getInitData<{ requestId: string }>(); + return { + requestId: init.requestId, + approved: (data as { approved?: boolean }).approved === true, + }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const suspended = await workflow.run({ requestId: "req-42" }); + expect(suspended.status).toBe("suspended"); + + const resumed = await suspended.resume({ approved: true }); + expect(resumed.status).toBe("completed"); + expect(resumed.result).toEqual({ + requestId: "req-42", + approved: true, + }); + }); + + it("should allow bail from nested steps", async () => { + const memory = new Memory({ storage: new InMemoryStorageAdapter() }); + let finalStepReached = false; + + const workflow = createWorkflow( + { + id: "execution-primitives-nested-bail", + name: "Execution Primitives Nested Bail", + input: z.object({ value: z.number() }), + result: z.object({ value: z.number() }), + memory, + }, + andWhen({ + id: "conditional-bail", + condition: async () => true, + step: andThen({ + id: "inner-bail", + execute: async ({ bail }) => { + bail({ value: 99 }); + }, + }), + }), + andThen({ + id: "should-not-run", + execute: async () => { + finalStepReached = true; + return { value: -1 }; + }, + }), + ); + + const registry = WorkflowRegistry.getInstance(); + registry.registerWorkflow(workflow); + + const result = await workflow.run({ value: 1 }); + + expect(result.status).toBe("completed"); + expect(result.result).toEqual({ value: 99 }); + expect(finalStepReached).toBe(false); + }); }); describe.sequential("workflow streaming", () => { diff --git a/packages/core/src/workflow/core.ts b/packages/core/src/workflow/core.ts index 5e1e630b4..55c0199f6 100644 --- a/packages/core/src/workflow/core.ts +++ b/packages/core/src/workflow/core.ts @@ -59,6 +59,38 @@ import type { export const VOLTAGENT_RESTART_CHECKPOINT_KEY = "__voltagent_restart_checkpoint"; const workflowReplayLogger = new LoggerProxy({ component: "workflow-core-replay" }); +const WORKFLOW_BAIL_SIGNAL = "WORKFLOW_BAIL_SIGNAL"; +const WORKFLOW_CANCELLED = "WORKFLOW_CANCELLED"; +const WORKFLOW_ABORT_REASON_DEFAULT = "Workflow aborted by step"; + +class WorkflowBailSignal extends Error { + readonly result: RESULT | undefined; + + constructor(result?: RESULT) { + super(WORKFLOW_BAIL_SIGNAL); + this.name = "WorkflowBailSignal"; + this.result = result; + } +} + +// Cancellation detection in execution paths depends on `error.message === WORKFLOW_CANCELLED`. +// Keep this signal message aligned with those checks. +class WorkflowAbortSignal extends Error { + readonly reason?: string; + + constructor(reason?: string) { + super(WORKFLOW_CANCELLED); + this.name = "WorkflowAbortSignal"; + this.reason = reason; + } +} + +const isWorkflowBailSignal = ( + error: unknown, +): error is WorkflowBailSignal => error instanceof WorkflowBailSignal; + +const isWorkflowAbortSignal = (error: unknown): error is WorkflowAbortSignal => + error instanceof WorkflowAbortSignal; const isObjectRecord = (value: unknown): value is Record => typeof value === "object" && value !== null && !Array.isArray(value); @@ -1549,6 +1581,183 @@ export function createWorkflow< } }; + const completeSuccessfulExecution = async ( + result: z.infer | null, + bailInfo?: { + stepId: string; + stepName: string; + stepIndex: number; + }, + ): Promise> => { + if (result === null) { + stateManager.update({ + result: null, + }); + } else { + stateManager.update({ + data: result, + result, + }); + } + + const finalState = stateManager.finish(); + + traceContext.setOutput(finalState.result); + traceContext.setUsage(stateManager.state.usage); + if (bailInfo) { + rootSpan.setAttribute("workflow.bailed", true); + rootSpan.setAttribute("workflow.bailed.step.id", bailInfo.stepId); + rootSpan.setAttribute("workflow.bailed.step.name", bailInfo.stepName); + rootSpan.setAttribute("workflow.bailed.step.index", bailInfo.stepIndex); + } + traceContext.end("completed"); + + await safeFlushOnFinish(observability); + + try { + await executionMemory.updateWorkflowState(executionContext.executionId, { + status: "completed", + workflowState: stateManager.state.workflowState, + events: collectedEvents, + output: finalState.result, + updatedAt: new Date(), + }); + } catch (memoryError) { + runLogger.warn("Failed to update workflow state to completed in Memory V2:", { + error: memoryError, + }); + } + + await runTerminalHooks("completed"); + + const duration = finalState.endAt.getTime() - finalState.startAt.getTime(); + runLogger.debug( + `Workflow completed | user=${options?.userId || "anonymous"} conv=${options?.conversationId || "none"} duration=${duration}ms`, + { + duration, + output: finalState.result !== undefined ? finalState.result : null, + ...(bailInfo + ? { + bailed: true, + bailStepId: bailInfo.stepId, + bailStepIndex: bailInfo.stepIndex, + } + : {}), + }, + ); + + emitAndCollectEvent({ + type: "workflow-complete", + executionId, + from: name, + output: finalState.result, + status: "success", + context: contextMap, + timestamp: new Date().toISOString(), + metadata: bailInfo + ? { + bailed: true, + bailStepId: bailInfo.stepId, + bailStepName: bailInfo.stepName, + bailStepIndex: bailInfo.stepIndex, + } + : undefined, + }); + + streamController?.close(); + return createWorkflowExecutionResult( + id, + executionId, + finalState.startAt, + finalState.endAt, + "completed", + finalState.result as z.infer | null, + stateManager.state.usage, + undefined, + stateManager.state.cancellation, + undefined, + effectiveResumeSchema, + ); + }; + + const completeBail = async ({ + bailSignal, + step, + stepName, + stepIndex, + span, + }: { + bailSignal: WorkflowBailSignal>; + step: BaseStep; + stepName: string; + stepIndex: number; + span?: ReturnType; + }): Promise> => { + const finalResult = bailSignal.result !== undefined ? bailSignal.result : null; + const spanToEnd = span ?? executionContext.currentStepSpan; + + if (spanToEnd) { + traceContext.endStepSpan(spanToEnd, "completed", { + output: finalResult, + attributes: { + "workflow.step.bailed": true, + }, + }); + + if (executionContext.currentStepSpan === spanToEnd) { + executionContext.currentStepSpan = undefined; + } + } + + const stepData = executionContext.stepData.get(step.id); + if (stepData) { + stepData.output = finalResult; + stepData.status = "success"; + stepData.error = null; + } + + if (finalResult === null) { + stateManager.update({ + result: null, + }); + } else { + stateManager.update({ + data: finalResult, + result: finalResult, + }); + } + + emitAndCollectEvent({ + type: "step-complete", + executionId, + from: stepName, + input: stateManager.state.data, + output: finalResult, + status: "success", + context: contextMap, + timestamp: new Date().toISOString(), + stepIndex, + stepType: step.type, + metadata: { + bailed: true, + }, + }); + + await hooks?.onStepEnd?.(stateManager.state); + + runLogger.debug(`Workflow bailed at step ${stepIndex + 1}: ${stepName}`, { + stepIndex, + stepName, + output: finalResult, + }); + + return completeSuccessfulExecution(finalResult, { + stepId: step.id, + stepName, + stepIndex, + }); + }; + try { if (workflowGuardrailRuntime && guardrailSets.input.length > 0) { if (!isWorkflowGuardrailInput(input)) { @@ -1691,8 +1900,13 @@ export function createWorkflow< const resolveCancellationReason = (abortValue?: unknown): string => { const reasonFromSignal = typeof abortValue === "string" && abortValue !== "cancelled" ? abortValue : undefined; + const reasonFromAbortError = + isWorkflowAbortSignal(abortValue) && abortValue.reason + ? abortValue.reason + : undefined; return ( + reasonFromAbortError ?? options?.suspendController?.getCancelReason?.() ?? activeController?.getCancelReason?.() ?? reasonFromSignal ?? @@ -2042,12 +2256,19 @@ export function createWorkflow< ...(stepRetryLimit > 0 && { "workflow.step.retry.count": retryCount }), }, }); + executionContext.currentStepSpan = attemptSpan; try { // Create execution context for the step with typed suspend function const typedSuspendFn = ( reason?: string, suspendData?: z.infer, ) => suspendFn(reason, suspendData); + const bailFn = (result?: z.infer): never => { + throw new WorkflowBailSignal>(result); + }; + const abortFn = (): never => { + throw new WorkflowAbortSignal(`${WORKFLOW_ABORT_REASON_DEFAULT}: ${stepName}`); + }; // Only pass resumeData if we're on the step that was suspended and we have resume input const isResumingThisStep = @@ -2065,26 +2286,23 @@ export function createWorkflow< ) : new NoOpWorkflowStreamWriter(); - // Create a modified execution context with the current step span - const stepExecutionContext = { - ...executionContext, - currentStepSpan: attemptSpan, // Add the current step span for agent integration - }; - const stepContext = createStepExecutionContext< WorkflowInput, typeof stateManager.state.data, + RESULT_SCHEMA, z.infer, z.infer >( stateManager.state.data, convertWorkflowStateToParam( stateManager.state, - stepExecutionContext, + executionContext, options?.suspendController?.signal, ), - stepExecutionContext, + executionContext, typedSuspendFn, + bailFn, + abortFn, isResumingThisStep ? resumeInputData : undefined, retryCount, ); @@ -2182,8 +2400,18 @@ export function createWorkflow< } break; } catch (stepError) { - if (stepError instanceof Error && stepError.message === "WORKFLOW_CANCELLED") { - const cancellationReason = resolveCancellationReason(); + if (isWorkflowBailSignal>(stepError)) { + return completeBail({ + bailSignal: stepError, + step, + stepName, + stepIndex: index, + span: attemptSpan, + }); + } + + if (stepError instanceof Error && stepError.message === WORKFLOW_CANCELLED) { + const cancellationReason = resolveCancellationReason(stepError); return completeCancellation(attemptSpan, cancellationReason); } @@ -2237,10 +2465,7 @@ export function createWorkflow< }, }, ); - if ( - delayError instanceof Error && - delayError.message === "WORKFLOW_CANCELLED" - ) { + if (delayError instanceof Error && delayError.message === WORKFLOW_CANCELLED) { const cancellationReason = resolveCancellationReason(); return completeCancellation(interruptionSpan, cancellationReason); } @@ -2270,6 +2495,10 @@ export function createWorkflow< }); throw stepError; // Re-throw the original error + } finally { + if (executionContext.currentStepSpan === attemptSpan) { + executionContext.currentStepSpan = undefined; + } } } } @@ -2288,72 +2517,34 @@ export function createWorkflow< }); } - const finalState = stateManager.finish(); - - // Record workflow completion in trace - traceContext.setOutput(finalState.result); - traceContext.setUsage(stateManager.state.usage); - traceContext.end("completed"); - - // Ensure spans are flushed (critical for serverless environments) - await safeFlushOnFinish(observability); + const finalResult = (stateManager.state.result ?? + stateManager.state.data) as z.infer; + return completeSuccessfulExecution(finalResult); + } catch (error) { + // Check if this is a cancellation or suspension, not an error + if (isWorkflowBailSignal>(error)) { + const bailStepIndex = executionContext.currentStepIndex; + const bailStep = (steps as BaseStep[])[bailStepIndex]; + const bailStepName = bailStep?.name || bailStep?.id || `Step ${bailStepIndex + 1}`; + if (!bailStep) { + const finalResult = error.result !== undefined ? error.result : null; + return completeSuccessfulExecution(finalResult); + } - // Update Memory V2 state to completed with events and output - try { - await executionMemory.updateWorkflowState(executionContext.executionId, { - status: "completed", - workflowState: stateManager.state.workflowState, - events: collectedEvents, - output: finalState.result, - updatedAt: new Date(), - }); - } catch (memoryError) { - runLogger.warn("Failed to update workflow state to completed in Memory V2:", { - error: memoryError, + return completeBail({ + bailSignal: error, + step: bailStep, + stepName: bailStepName, + stepIndex: bailStepIndex, + span: executionContext.currentStepSpan, }); } - await runTerminalHooks("completed"); - - // Log workflow completion with context - const duration = finalState.endAt.getTime() - finalState.startAt.getTime(); - runLogger.debug( - `Workflow completed | user=${options?.userId || "anonymous"} conv=${options?.conversationId || "none"} duration=${duration}ms`, - { - duration, - output: finalState.result !== undefined ? finalState.result : null, - }, - ); - - // Emit workflow complete event - emitAndCollectEvent({ - type: "workflow-complete", - executionId, - from: name, - output: finalState.result, - status: "success", - context: contextMap, - timestamp: new Date().toISOString(), - }); - - streamController?.close(); - return createWorkflowExecutionResult( - id, - executionId, - finalState.startAt, - finalState.endAt, - "completed", - finalState.result as z.infer, - stateManager.state.usage, - undefined, - stateManager.state.cancellation, - undefined, - effectiveResumeSchema, - ); - } catch (error) { - // Check if this is a cancellation or suspension, not an error - if (error instanceof Error && error.message === "WORKFLOW_CANCELLED") { + if (error instanceof Error && error.message === WORKFLOW_CANCELLED) { + const reasonFromAbortError = + isWorkflowAbortSignal(error) && error.reason ? error.reason : undefined; const cancellationReason = + reasonFromAbortError ?? options?.suspendController?.getCancelReason?.() ?? workflowRegistry.activeExecutions.get(executionId)?.getCancelReason?.() ?? options?.suspendController?.getReason?.() ?? @@ -3480,11 +3671,11 @@ async function executeWithSignalCheck( if (reason && typeof reason === "object" && reason !== null && "type" in reason) { const typedReason = reason as { type?: string }; if (typedReason.type === "cancelled") { - return new Error("WORKFLOW_CANCELLED"); + return new Error(WORKFLOW_CANCELLED); } } if (reason === "cancelled") { - return new Error("WORKFLOW_CANCELLED"); + return new Error(WORKFLOW_CANCELLED); } return new Error("WORKFLOW_SUSPENDED"); }; diff --git a/packages/core/src/workflow/internal/types.ts b/packages/core/src/workflow/internal/types.ts index 23534d3f0..0c3fe67ae 100644 --- a/packages/core/src/workflow/internal/types.ts +++ b/packages/core/src/workflow/internal/types.ts @@ -28,11 +28,21 @@ export type InternalBaseWorkflowInputSchema = * Context object for new execute API with helper functions * @private - INTERNAL USE ONLY */ -export interface WorkflowExecuteContext { +export interface WorkflowExecuteContext< + INPUT, + DATA, + SUSPEND_DATA, + RESUME_DATA, + WORKFLOW_RESULT = unknown, +> { data: DATA; state: WorkflowStepState; getStepData: (stepId: string) => WorkflowStepData | undefined; + getStepResult: (stepId: string) => T | null; + getInitData: >() => T; suspend: (reason?: string, suspendData?: SUSPEND_DATA) => Promise; + bail: (result?: WORKFLOW_RESULT) => never; + abort: () => never; resumeData?: RESUME_DATA; retryCount?: number; workflowState: WorkflowStateStore; @@ -54,8 +64,15 @@ export interface WorkflowExecuteContext * Uses context-based API with data, state, and helper functions * @private - INTERNAL USE ONLY */ -export type InternalWorkflowFunc = ( - context: WorkflowExecuteContext, +export type InternalWorkflowFunc< + INPUT, + DATA, + RESULT, + SUSPEND_DATA, + RESUME_DATA, + WORKFLOW_RESULT = unknown, +> = ( + context: WorkflowExecuteContext, ) => Promise; export type InternalWorkflowStepConfig = { @@ -82,7 +99,14 @@ export type InternalWorkflowStepConfig = { * Base step interface for building new steps * @private - INTERNAL USE ONLY */ -export interface InternalBaseWorkflowStep { +export interface InternalBaseWorkflowStep< + INPUT, + DATA, + RESULT, + SUSPEND_DATA, + RESUME_DATA, + WORKFLOW_RESULT = unknown, +> { /** * Unique identifier for the step */ @@ -125,7 +149,7 @@ export interface InternalBaseWorkflowStep, + context: WorkflowExecuteContext, ) => Promise; } @@ -139,9 +163,13 @@ export type InternalAnyWorkflowStep< RESULT = DangerouslyAllowAny, SUSPEND_DATA = DangerouslyAllowAny, RESUME_DATA = DangerouslyAllowAny, + WORKFLOW_RESULT = unknown, > = - | InternalBaseWorkflowStep - | Omit, "type">; + | InternalBaseWorkflowStep + | Omit< + InternalBaseWorkflowStep, + "type" + >; /** * Infer the result type from a list of steps diff --git a/packages/core/src/workflow/internal/utils.ts b/packages/core/src/workflow/internal/utils.ts index 4c390cd5d..66ea236a3 100644 --- a/packages/core/src/workflow/internal/utils.ts +++ b/packages/core/src/workflow/internal/utils.ts @@ -5,11 +5,16 @@ import { createUIMessageStream, createUIMessageStreamResponse, } from "ai"; +import type { z } from "zod"; import type { WorkflowExecutionContext } from "../context"; import type { WorkflowStreamController } from "../stream"; import type { WorkflowStateUpdater, WorkflowStepState, WorkflowStreamEvent } from "../types"; import type { WorkflowState } from "./state"; -import type { InternalWorkflowStepConfig, WorkflowExecuteContext } from "./types"; +import type { + InternalExtractWorkflowInputData, + InternalWorkflowStepConfig, + WorkflowExecuteContext, +} from "./types"; /** * Convert a workflow state to a parameter for a step or hook @@ -64,20 +69,46 @@ export function defaultStepConfig(con * @param suspendFn - The suspend function for the step * @returns The execution context for the step */ -export function createStepExecutionContext( +export function createStepExecutionContext< + INPUT, + DATA, + RESULT_SCHEMA extends z.ZodTypeAny, + SUSPEND_DATA, + RESUME_DATA, +>( data: DATA, state: WorkflowStepState, executionContext: WorkflowExecutionContext, suspendFn: (reason?: string, suspendData?: SUSPEND_DATA) => Promise, + bailFn?: (result?: z.infer) => never, + abortFn?: () => never, resumeData?: RESUME_DATA, retryCount = 0, setWorkflowState?: (update: WorkflowStateUpdater) => void, -): WorkflowExecuteContext { +): WorkflowExecuteContext> { return { data, state, getStepData: (stepId: string) => executionContext?.stepData.get(stepId), + getStepResult: (stepId: string) => { + const stepData = executionContext?.stepData.get(stepId); + if (!stepData || stepData.output === undefined) { + return null; + } + return stepData.output as T; + }, + getInitData: >() => state.input as T, suspend: suspendFn, + bail: + bailFn ?? + (() => { + throw new Error("WORKFLOW_BAIL_NOT_CONFIGURED"); + }), + abort: + abortFn ?? + (() => { + throw new Error("WORKFLOW_ABORT_NOT_CONFIGURED"); + }), resumeData, retryCount, workflowState: executionContext.workflowState, diff --git a/website/docs/workflows/overview.md b/website/docs/workflows/overview.md index a242b8172..c7c242bda 100644 --- a/website/docs/workflows/overview.md +++ b/website/docs/workflows/overview.md @@ -433,6 +433,60 @@ This allows the agent to maintain a persistent, contextual conversation with eac // used by the agent's memory to provide context-aware responses. ``` +### Execution Primitives in Step Context + +Each workflow step now has control/introspection helpers for early exits and history lookups: + +- `bail(result?)`: Complete the workflow immediately with a final result +- `abort()`: Cancel the workflow immediately +- `getStepResult(stepId)`: Read a previous step's output (or `null`) +- `getInitData()`: Read the original workflow input, even after resume + +```typescript +import { createWorkflowChain } from "@voltagent/core"; +import { z } from "zod"; + +const workflow = createWorkflowChain({ + id: "execution-primitives-demo", + name: "Execution Primitives Demo", + input: z.object({ userId: z.string(), amount: z.number() }), + result: z.object({ + status: z.enum(["approved", "rejected", "cancelled"]), + userId: z.string(), + }), +}) + .andThen({ + id: "risk-check", + execute: async ({ data }) => ({ + score: data.amount > 1000 ? 95 : 20, + userId: data.userId, + }), + }) + .andThen({ + id: "decision", + execute: async ({ getStepResult, getInitData, bail, abort }) => { + const risk = getStepResult<{ score: number; userId: string }>("risk-check"); + const init = getInitData(); + + if (!risk) { + abort(); // terminal status: cancelled + } + + if (risk.score >= 90) { + bail({ + status: "rejected", + userId: init.userId, + }); // terminal status: completed + } + + return { + status: "approved", + userId: init.userId, + }; + }, + }); +``` + ### Workflow Retry Policies Set a workflow-wide default retry policy with `retryConfig`. Steps inherit it unless they define `retries` (use `retries: 0` to opt out). `delayMs` waits between retry attempts. diff --git a/website/docs/workflows/suspend-resume.md b/website/docs/workflows/suspend-resume.md index ff38b13e5..0191ed2d0 100644 --- a/website/docs/workflows/suspend-resume.md +++ b/website/docs/workflows/suspend-resume.md @@ -287,6 +287,35 @@ await exec2.resume( - `suspend` - Function to pause the workflow - `resumeData` - Data provided when resuming (undefined on first run) - `suspendData` - Data that was saved during suspension +- `getInitData` - The original workflow input (stable across resume) + +### Access Original Input with `getInitData` + +Use `getInitData()` when you need the first input payload after multiple transforms or resume cycles. + +```typescript +.andThen({ + id: "approval-step", + resumeSchema: z.object({ approved: z.boolean() }), + execute: async ({ data, suspend, resumeData }) => { + if (resumeData) { + return { ...data, approved: resumeData.approved }; + } + + await suspend("Approval required"); + }, +}) +.andThen({ + id: "finalize", + execute: async ({ data, getInitData }) => { + const init = getInitData(); + return { + requestId: init.requestId, + approved: data.approved === true, + }; + }, +}); +``` ## Common Patterns