feat(core): add workflow startAsync fire-and-forget API#1097
Merged
Conversation
🦋 Changeset detectedLatest commit: ce7eb63 The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
This comment has been minimized.
This comment has been minimized.
Contributor
📝 WalkthroughWalkthroughAdds a fire-and-forget startAsync() API to Workflow and WorkflowChain that returns execution metadata immediately, launches background execution, and persists terminal states (completed/error) in memory for later inspection. Changes
Sequence Diagram(s)sequenceDiagram
actor Caller
participant Workflow
participant Memory
participant Background
Caller->>Workflow: startAsync(input, options)
Workflow->>Workflow: generate/use executionId, prepare start metadata
Workflow->>Memory: persist initial "running" state (executionId, workflowId, startAt)
Workflow->>Background: kick off executeInternal (non-blocking, skipStateInit)
Workflow-->>Caller: return { executionId, workflowId, startAt }
par Background execution
Background->>Background: run workflow steps
Background->>Memory: persist terminal state (completed/error) and error metadata if any
and Caller continues
Caller->>Caller: proceed without waiting
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Deploying voltagent with
|
| Latest commit: |
ce7eb63
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://64d34b01.voltagent.pages.dev |
| Branch Preview URL: | https://feat-workflow-start-async-fi.voltagent.pages.dev |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
packages/core/src/workflow/types.ts (1)
205-221:startedAtis inconsistent withstartAton related types.
WorkflowExecutionResultandWorkflowStreamResultboth expose astartAt: Datefield. IntroducingstartedAtonWorkflowStartAsyncResultcreates a naming inconsistency in the public API that will trip up callers switching between methods.🔧 Proposed rename
export interface WorkflowStartAsyncResult { executionId: string; workflowId: string; - startedAt: Date; + startAt: Date; }Also update the corresponding field in
core.tsstartAsyncreturn value and the assertions in both spec files.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/core/src/workflow/types.ts` around lines 205 - 221, Rename the field startedAt to startAt to match the existing API (WorkflowExecutionResult and WorkflowStreamResult): update the WorkflowStartAsyncResult interface to use startAt: Date, update the startAsync implementation in core.ts to populate startAt instead of startedAt, and adjust all tests/spec assertions that reference startedAt to expect startAt (search for WorkflowStartAsyncResult, startAsync, and any spec files asserting startedAt and change them accordingly).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/core/src/workflow/core.spec.ts`:
- Around line 507-508: Replace the immediate assertion that checks runningState
with a polling loop (same pattern as the completion check) that repeatedly calls
memory.getWorkflowState(startResult.executionId) until status === "running" or a
timeout is reached; keep the gate held while polling and only call
releaseStep?.() after the polling confirms "running" (or times out) so the test
is robust to async adapters. Use the same timeout/interval pattern as the
existing completion check to implement the loop.
In `@packages/core/src/workflow/core.ts`:
- Around line 2146-2208: The outer .catch(() => {}) swallows any exceptions
thrown by the async error handler; move the initial logger.warn call (the one
logging "startAsync execution failed before terminal handling") out of the try
block so it always executes, then keep the try/catch solely around the
persistence calls (executionMemory.getWorkflowState / updateWorkflowState /
setWorkflowState) to handle persistence failures; finally replace the trailing
.catch(() => {}) with a handler that logs unexpected errors from the async
handler (or rethrows) so failures in executeInternal's error-handling logic
aren't silently discarded.
- Around line 2134-2215: startAsync currently returns an executionId before
executeInternal creates the initial "running" state, causing callers to see no
state for real DB backends; fix by having startAsync synchronously/awaitably
create the initial workflow state via executionMemory.setWorkflowState(...)
(with status: "running", id: executionId, workflowId: id, workflowName: name,
input, metadata, createdAt/updatedAt, etc.) before invoking executeInternal, and
pass a new flag on executionOptions (e.g., skipStateInit: true) so
executeInternal checks options.skipStateInit and skips its own initial
setWorkflowState call and instead only updates trace IDs/metadata via
updateWorkflowState to avoid double-creation.
---
Nitpick comments:
In `@packages/core/src/workflow/types.ts`:
- Around line 205-221: Rename the field startedAt to startAt to match the
existing API (WorkflowExecutionResult and WorkflowStreamResult): update the
WorkflowStartAsyncResult interface to use startAt: Date, update the startAsync
implementation in core.ts to populate startAt instead of startedAt, and adjust
all tests/spec assertions that reference startedAt to expect startAt (search for
WorkflowStartAsyncResult, startAsync, and any spec files asserting startedAt and
change them accordingly).
Comment on lines
+2146
to
+2208
| 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; | ||
| } | ||
|
|
||
| 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: "error", | ||
| input, | ||
| context: contextEntries, | ||
| workflowState: options?.workflowState ?? {}, | ||
| userId: options?.userId, | ||
| conversationId: options?.conversationId, | ||
| metadata: { | ||
| ...(optionMetadata ?? {}), | ||
| errorMessage, | ||
| }, | ||
| createdAt: startedAt, | ||
| updatedAt: new Date(), | ||
| }); | ||
| } catch (persistenceError) { | ||
| logger.warn("Failed to persist startAsync background failure", { | ||
| executionId, | ||
| error: persistenceError, | ||
| }); | ||
| } | ||
| }) | ||
| .catch(() => { | ||
| // Prevent unhandled rejections from the async catch handler. | ||
| }); |
Contributor
There was a problem hiding this comment.
Second .catch() silently discards errors from the async error-handler itself.
The pattern .catch(async handler).catch(() => {}) is correct for suppressing unhandled-rejection warnings, but the outer .catch(() => {}) swallows all exceptions thrown by the async inner handler — including any unanticipated logic errors that escape the inner try/catch. If, for example, logger.warn(...) throws or error instanceof Error check somehow fails, the problem would be invisible.
Consider promoting the inner logger call outside the try-catch so at minimum the error is logged before any persistence attempt:
.catch(async (error) => {
const errorMessage = error instanceof Error ? error.message : String(error);
+
+ logger.warn("startAsync execution failed before terminal handling", {
+ executionId,
+ error,
+ });
try {
- logger.warn("startAsync execution failed before terminal handling", {
- executionId,
- error,
- });
const existingState = await executionMemory.getWorkflowState(executionId);This way, even if the persistence block fails, the warning is always emitted.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/core/src/workflow/core.ts` around lines 2146 - 2208, The outer
.catch(() => {}) swallows any exceptions thrown by the async error handler; move
the initial logger.warn call (the one logging "startAsync execution failed
before terminal handling") out of the try block so it always executes, then keep
the try/catch solely around the persistence calls
(executionMemory.getWorkflowState / updateWorkflowState / setWorkflowState) to
handle persistence failures; finally replace the trailing .catch(() => {}) with
a handler that logs unexpected errors from the async handler (or rethrows) so
failures in executeInternal's error-handling logic aren't silently discarded.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
website/docs/workflows/streaming.md (1)
45-86:⚠️ Potential issue | 🟡 MinorMethod ordering in the bullet list and code block is inconsistent.
The bullet list (lines 47–49) orders the methods as
.run()→.startAsync()→.stream(), but the code block labels them Method 1 =.stream(), Method 2 =.run(), Method 3 =.startAsync(). Align either the bullet list order or the "Method N" comments so readers can cross-reference without confusion.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@website/docs/workflows/streaming.md` around lines 45 - 86, The documentation lists methods in one order but labels them differently in the code block; align the bullet list and the code comments so readers can cross-reference. Update either the bullets to list `.stream()`, `.run()`, `.startAsync()` or change the code comments so Method 1 = `.run()`, Method 2 = `.startAsync()`, Method 3 = `.stream()` (ensure references to `workflow.stream`, `workflow.run`, `workflow.startAsync`, and `stream.result` remain correct and consistent).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@website/docs/workflows/overview.md`:
- Around line 273-282: The snippet incorrectly references
greeterChain.toWorkflow() before greeterChain is defined; remove the unnecessary
toWorkflow() call and call startAsync directly on greeterChain (use
greeterChain.startAsync({...})) and then use
greeterChain.memory.getWorkflowState(executionId) to query status; update the
console.log lines accordingly so the snippet is self-contained and consistent
with the earlier examples that use WorkflowChain.startAsync and
memory.getWorkflowState.
In `@website/docs/workflows/streaming.md`:
- Around line 783-788: The interface WorkflowStartAsyncResult is inconsistent
with WorkflowExecutionResult and WorkflowStreamResult because it exposes
startedAt instead of startAt; change the property name in
WorkflowStartAsyncResult from startedAt: Date to startAt: Date, update all
references/usages and any serialization/deserialization or tests that rely on
startedAt, and run/typecheck to ensure consumers now use workflowResult.startAt
consistently across WorkflowStartAsyncResult, WorkflowExecutionResult, and
WorkflowStreamResult.
---
Outside diff comments:
In `@website/docs/workflows/streaming.md`:
- Around line 45-86: The documentation lists methods in one order but labels
them differently in the code block; align the bullet list and the code comments
so readers can cross-reference. Update either the bullets to list `.stream()`,
`.run()`, `.startAsync()` or change the code comments so Method 1 = `.run()`,
Method 2 = `.startAsync()`, Method 3 = `.stream()` (ensure references to
`workflow.stream`, `workflow.run`, `workflow.startAsync`, and `stream.result`
remain correct and consistent).
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
1 issue found across 7 files (changes from recent commits).
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name=".changeset/cool-geese-wave.md">
<violation number="1" location=".changeset/cool-geese-wave.md:7">
P3: The changeset now documents `startAt`, but the API uses `startedAt` per the PR description. This inconsistency will mislead consumers about the correct response field.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
|
|
||
| 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. |
Contributor
There was a problem hiding this comment.
P3: The changeset now documents startAt, but the API uses startedAt per the PR description. This inconsistency will mislead consumers about the correct response field.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At .changeset/cool-geese-wave.md, line 7:
<comment>The changeset now documents `startAt`, but the API uses `startedAt` per the PR description. This inconsistency will mislead consumers about the correct response field.</comment>
<file context>
@@ -4,6 +4,6 @@
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, startedAt }` immediately. The run keeps existing execution semantics, respects provided `executionId`, and persists terminal states in memory for later inspection.
+`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.
</file context>
Suggested change
| `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. | |
| `startAsync()` starts a workflow run in the background and returns `{ executionId, workflowId, startedAt }` immediately. The run keeps existing execution semantics, respects provided `executionId`, and persists terminal states in memory for later inspection. |
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/core/src/workflow/core.ts`:
- Around line 2171-2175: The startAsync implementation currently uses
options?.resumeFrom?.executionId to derive executionId which allows callers to
pass resumeFrom into the start API and accidentally overwrite a suspended run
when setWorkflowState writes status:"running"; update startAsync (and related
logic around executionId derivation) to reject or ignore resumeFrom:
specifically, remove using resumeFrom.executionId when deriving executionId (use
options.executionId or randomUUID only), and add a guard at the top of
startAsync to detect if options.resumeFrom is present and either throw an error
or refuse to proceed if a stored execution exists with status "suspended" (check
via the same storage lookup used by setWorkflowState) so the suspended
checkpoint and events are not overwritten; keep resume behavior confined to the
dedicated resume/executeInternal path.
In `@website/docs/workflows/overview.md`:
- Around line 292-297: The docs example incorrectly accesses greeterChain.memory
which doesn’t exist on WorkflowChain; update the snippet to call
greeterChain.toWorkflow() and then use the public memory API to query state
(i.e., call
greeterChain.toWorkflow().memory.getWorkflowState(started.executionId)); keep
use of started = await greeterChain.startAsync({ name: "Alice" }) and
started.executionId/startAt unchanged and only replace the memory access with
the toWorkflow().memory.getWorkflowState call.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
packages/core/src/workflow/core.ts (1)
2177-2189: Unnecessary async storage round-trip inresumeFromguard.Both branches always throw — the
getWorkflowStatecall is made solely to pick between two error messages. If storage itself fails here, the caller receives a confusing storage error instead of the expected "resumeFrom not supported" error.♻️ Simplified guard (no storage I/O on the fast-reject path)
- 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.", - ); - } + if (options?.resumeFrom) { + throw new Error( + "startAsync does not support resumeFrom. Use workflow.run(...) or workflow.stream(...) with resumeFrom.", + ); + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/core/src/workflow/core.ts` around lines 2177 - 2189, The resumeFrom guard is performing unnecessary storage I/O by calling executionMemory.getWorkflowState (resumeExecutionId/resumeState) even though both branches always throw; remove the executionMemory.getWorkflowState call and the suspended-status branch and instead immediately throw the generic "startAsync does not support resumeFrom. Use workflow.run(...) or workflow.stream(...) with resumeFrom." when options?.resumeFrom is present (references: options?.resumeFrom, startAsync, executionMemory.getWorkflowState, resumeExecutionId, resumeState).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/core/src/workflow/core.ts`:
- Around line 2223-2227: Public API type WorkflowRunOptions currently exposes
internal flag skipStateInit, allowing callers of run()/stream()/startAsync() to
bypass state initialization; fix by creating a separate internal options type
(e.g., InternalWorkflowRunOptions) that extends WorkflowRunOptions but adds
skipStateInit, update executeInternal signature to accept
InternalWorkflowRunOptions (or a private overload), and ensure all internal call
sites such as startAsync and the call at executionOptions creation replace
WorkflowRunOptions with the new internal type so skipStateInit is no longer
present on the public WorkflowRunOptions surface.
---
Nitpick comments:
In `@packages/core/src/workflow/core.ts`:
- Around line 2177-2189: The resumeFrom guard is performing unnecessary storage
I/O by calling executionMemory.getWorkflowState (resumeExecutionId/resumeState)
even though both branches always throw; remove the
executionMemory.getWorkflowState call and the suspended-status branch and
instead immediately throw the generic "startAsync does not support resumeFrom.
Use workflow.run(...) or workflow.stream(...) with resumeFrom." when
options?.resumeFrom is present (references: options?.resumeFrom, startAsync,
executionMemory.getWorkflowState, resumeExecutionId, resumeState).
Comment on lines
+2223
to
+2227
| const executionOptions: WorkflowRunOptions = { | ||
| ...options, | ||
| executionId, | ||
| skipStateInit: true, | ||
| }; |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check where skipStateInit is declared
rg -n 'skipStateInit' --type=ts -B 2 -A 2
# Confirm whether WorkflowRunOptions is exported from the public index
rg -n 'WorkflowRunOptions' packages/core/src/index.ts packages/core/src/workflow/index.ts 2>/dev/null || trueRepository: VoltAgent/voltagent
Length of output: 1019
🏁 Script executed:
#!/bin/bash
# Get the full WorkflowRunOptions interface definition
rg -n 'interface WorkflowRunOptions|export.*WorkflowRunOptions' --type=ts -A 30 packages/core/src/workflow/types.tsRepository: VoltAgent/voltagent
Length of output: 1068
🏁 Script executed:
#!/bin/bash
# Get more of the WorkflowRunOptions interface, including skipStateInit
rg -n 'export interface WorkflowRunOptions' --type=ts -A 80 packages/core/src/workflow/types.ts | head -100Repository: VoltAgent/voltagent
Length of output: 2651
skipStateInit is exposed in the public WorkflowRunOptions type without type-level enforcement.
skipStateInit is declared in WorkflowRunOptions (line 318, types.ts) with only an @internal JSDoc comment. Since WorkflowRunOptions is publicly exported and used as the parameter type for run(), stream(), and startAsync(), any caller can pass skipStateInit: true. This will bypass initial state creation (line 930, core.ts) and fail with a confusing "Workflow state X not found" error if no pre-created state exists.
Recommend creating a separate internal type for executeInternal:
♻️ Proposed fix — separate internal options type
+// In types.ts (internal, not exported):
+export interface WorkflowInternalRunOptions extends WorkflowRunOptions {
+ skipStateInit?: boolean;
+}Then update executeInternal's signature:
- const executeInternal = async (
- input: WorkflowInput<INPUT_SCHEMA>,
- options?: WorkflowRunOptions,
+ const executeInternal = async (
+ input: WorkflowInput<INPUT_SCHEMA>,
+ options?: WorkflowInternalRunOptions,And the startAsync call site (line 2223):
- const executionOptions: WorkflowRunOptions = {
+ const executionOptions: WorkflowInternalRunOptions = {
...options,
executionId,
skipStateInit: true,
};This enforces type safety by keeping internal flags out of the public API surface.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/core/src/workflow/core.ts` around lines 2223 - 2227, Public API type
WorkflowRunOptions currently exposes internal flag skipStateInit, allowing
callers of run()/stream()/startAsync() to bypass state initialization; fix by
creating a separate internal options type (e.g., InternalWorkflowRunOptions)
that extends WorkflowRunOptions but adds skipStateInit, update executeInternal
signature to accept InternalWorkflowRunOptions (or a private overload), and
ensure all internal call sites such as startAsync and the call at
executionOptions creation replace WorkflowRunOptions with the new internal type
so skipStateInit is no longer present on the public WorkflowRunOptions surface.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PR Checklist
Please check if your PR fulfills the following requirements:
Bugs / Features
What is the current behavior?
Workflow APIs support
run()andstream(), but there is no fire-and-forget start method. Callers must wait for completion or manage stream lifecycle directly.What is the new behavior?
Adds
startAsync()to both workflow surfaces:Workflow.startAsync(input, options?)WorkflowChain.startAsync(input, options?)startAsync()starts execution in the background and immediately returns:executionIdworkflowIdstartedAtImplementation details:
executeInternalexecution path.executionIdfrom options when provided.fixes (issue)
N/A
Notes for reviewers
01-async-start-fire-and-forget+ changeset.docs/workflow-parity-planswas intentionally excluded from this commit/PR.Summary by cubic
Adds startAsync to workflows and chains for fire-and-forget execution that returns immediately while the run continues in the background. It returns executionId, workflowId, and startAt, persists terminal state for later inspection, and rejects resumeFrom to avoid overwriting suspended runs.
New Features
Bug Fixes
Written for commit ce7eb63. Summary will update on new commits.
Summary by CodeRabbit
New Features
Tests
Documentation
Chores