I'm having error + not sure how to handle HITL of agent that run workflow

[daneatmastra]

This issue was created from Discord post 1450479219783110737:

Heya, I'm having error + not sure how to handle HITL of agent that run workflow.

//Agent call
await emailAgent.generate(prompt);

//Agent definition
export const emailAgent = new Agent({
workflows: {
emailProcessingWorkflow,
},
........

// Workflow step
if (!resumeData) {
return suspend({.....

I get error:
Error processing email: Error: promise 'steps' was not resolved or rejected when stream finished

What would be the correct way to have workflow with HITL in agent? And what if agent is not streaming?

[coderabbitai]

📝 CodeRabbit Plan Mode

Generate an implementation plan and prompts that you can use with your favorite coding agent.

✅ Successfully generated implementation plan.

💡 Use @coderabbitai <feedback> to replan

Examples

• Example 1
• Example 2

_{You can configure auto-planning by selecting labels in the issue_enrichment configuration.}

🔗 Related PRs

#10278 - fix(core): agent workflow and subagent tools [merged]
#10871 - Add human-in-the-loop support for workflows used in agent [merged]

👤 Suggested Assignees

• DanielSLew
• taofeeq-deru

---

🧪 Issue enrichment is currently in early access.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord or schedule a call!

[daneatmastra]

Thank you for reporting this issue! We have assigned it to the trio-tnt and we will look into it as soon as possible.

If you're experiencing an error, please provide a minimal reproducible example whenever possible to help us resolve it quickly.

[coderabbitai]

Implementation Plan

Approach

Enable agents with workflows to properly handle workflow suspension (HITL) in both streaming and non-streaming contexts. Currently, when a workflow step calls suspend() within an agent's workflow tool, the generate() method fails with "promise 'steps' was not resolved or rejected when stream finished" because suspension handling is incomplete.

Observations

The Mastra codebase implements a workflow-based agent architecture where workflows are converted into callable tools. When workflows suspend (for HITL), they return suspension metadata. The streaming infrastructure already has suspension support via 'tool-call-suspended' chunks and suspension promise tracking, but the generate() method lacks proper suspension handling. The Agent class wraps workflows in #convertWorkflowsToTools(), which detects suspension but only works correctly in streaming contexts.

Assumptions

Assumption 1: How should generate() expose suspension state to callers?

Options Considered:

• Return a special status flag and suspension data in the generate() result object, requiring callers to check the status and call a separate resume method
• Throw a special suspension exception that contains the data needed to resume
• Automatically convert to streaming mode when suspension is detected

Chosen Option: Return a special status flag and suspension data in the generate() result object, requiring callers to check the status and call a separate resume method

Rationale: This aligns with workflow patterns where result.status === 'suspended' is checked, feels natural to developers already familiar with workflow suspend/resume, and avoids the complexity of error-based control flow or forced streaming mode changes. The agent result type will include an optional suspended field with suspension metadata and a resume() method reference.

Assumption 2: Should we maintain backward compatibility with existing generate() return types?

Options Considered:

• Add new fields to existing return type (potentially breaking type checks)
• Create a new method like generateWithSuspend() for workflows that might suspend
• Use discriminated unions based on status field

Chosen Option: Use discriminated unions based on status field

Rationale: This provides type safety—TypeScript will know which fields are available based on the status value—and matches existing patterns in the codebase where workflow results use status-based discrimination. Existing code that doesn't check for suspension will continue to work, but TypeScript will encourage proper handling.

💡 User Tips

Regenerate the plan with different choices with @coderabbitai <feedback>.

Plan

Phase 1: Extend Agent Result Types for Suspension Support

Add suspension state to agent result types and establish the structure for communicating suspension to callers in non-streaming contexts.

Task 1: Update Agent Result Type Definitions

Extend the agent result types to include suspension state as a discriminated union.

• Add status field to agent result types with values 'completed', 'suspended', or 'failed'
• For suspended status: Include suspendedToolCalls array containing tool call IDs, suspension payloads, and resume schemas
• Add resume() method reference that callers can use to continue execution
• Update type definitions in packages/core/src/agent/agent.types.ts or create new types in packages/core/src/agent/types.ts

Task 2: Track Execution Context in Agent Methods

Enable the agent to distinguish between streaming and non-streaming contexts during workflow tool execution.

• Add context flag to track whether current execution is streaming or not
• Pass this context through to the workflow tool wrapper's execute function
• Store context in the agent's execution options or as an instance variable during generate() or stream() calls
• This context will determine whether suspension should emit chunks or return state

🤖 Prompt for AI agents

Extend agent result types to support workflow suspension in non-streaming
contexts.

Update type definitions in `packages/core/src/agent/agent.types.ts`:
- Add discriminated union for agent results with `status` field ('completed',
'suspended', or 'failed')
- For suspended status: add `suspendedToolCalls` array with tool call IDs,
suspension payloads, and resume schemas
- Add `resume()` method reference to suspended results

Modify agent execution tracking in `packages/core/src/agent/agent.ts`:
- Add execution context flag to distinguish streaming vs non-streaming mode
- Pass context to workflow tool wrapper's execute function
- Store context in execution options or instance variable during `generate()`
and `stream()` calls

Phase 2: Handle Workflow Suspension in Non-Streaming Contexts

Modify how the workflow tool wrapper handles suspension to prevent unresolved promises in generate() contexts.

Task 1: Detect and Capture Suspension State in Workflow Tool Wrapper

Update the workflow tool's execute function in #convertWorkflowsToTools() to capture suspension state without emitting chunks when in generate() context.

• In packages/core/src/agent/agent.ts, modify the workflow tool execute handler
• When result?.status === 'suspended' is detected, check the execution context
• For non-streaming (generate) context: Store suspension state (tool call ID, suspend payload, resume schema, suspended step IDs) in a tracking structure rather than calling suspend?.()
• For streaming context: Maintain existing behavior of calling suspend?.() to emit chunks
• Return a resolved promise with suspension metadata for non-streaming contexts to prevent the "promise not resolved" error

Task 2: Accumulate Suspension State During Tool Execution

Collect all suspension events that occur during agent execution to return them in the final result.

• Create a suspension tracking structure at the agent execution level (in generate() method scope)
• When workflow tools suspend in generate() context, add their suspension data to this collection
• Track: tool call ID, workflow name, suspended step ID, suspend payload, resume schema
• This collection will be included in the agent result when execution completes

Task 3: Return Suspension State from Generate Method

Modify the generate() method to return suspension information when workflows suspend.

• In packages/core/src/agent/agent.ts, update the generate() method's return handling
• After agent execution completes, check if any suspensions were collected
• If suspensions exist: Return result with status: 'suspended', include suspendedToolCalls array, and attach a resume() method reference
• If no suspensions: Return result with status: 'completed' as normal
• Ensure proper promise resolution for all tracked promises before returning

🤖 Prompt for AI agents

Implement workflow suspension handling in non-streaming generate() contexts to
prevent unresolved promise errors.

Modify workflow tool wrapper in `packages/core/src/agent/agent.ts` within
`#convertWorkflowsToTools()`:
- Detect when `result?.status === 'suspended'` and check execution context
- For non-streaming: store suspension state (tool call ID, suspend payload,
resume schema, suspended step IDs) in tracking structure instead of calling
`suspend?.()`
- For streaming: maintain existing `suspend?.()` behavior
- Return resolved promise with suspension metadata to prevent promise errors

Update `generate()` method in `packages/core/src/agent/agent.ts`:
- Create suspension tracking structure in method scope
- Collect suspension data from workflow tools during execution (tool call ID,
workflow name, suspended step ID, suspend payload, resume schema)
- After execution: check for collected suspensions
- Return result with `status: 'suspended'` and `suspendedToolCalls` array if
suspensions exist, otherwise `status: 'completed'`
- Attach `resume()` method reference to suspended results
- Ensure all promises resolve before returning

Phase 3: Implement Agent-Level Resume Capabilities

Add methods to resume suspended workflow tools from both generate() and stream() contexts.

Task 1: Add Resume Method to Agent Class

Create a method on the Agent class that can resume suspended workflow tool executions.

• Add resumeGenerate() method to the Agent class in packages/core/src/agent/agent.ts
• Accept parameters: original prompt/messages, suspension resume data (mapping tool call IDs to resume payloads), and optional execution options
• Method should reconstruct the execution state from the suspension point
• Resume workflow tools by calling their underlying workflow's resume() method with the provided resume data
• Return the resumed execution result with the same discriminated union type (could complete or suspend again)

Task 2: Support Resume in Streaming Contexts

Extend the existing stream resume functionality to properly handle workflow tool resumption.

• Verify that existing resumeStream() method in packages/core/src/agent/agent.ts properly handles workflow suspension
• Ensure workflow tools can be resumed through the stream approval/decline flow
• When approveToolCall() or declineToolCall() is called for a suspended workflow tool, pass resume data to the workflow's resume method
• Maintain suspension promise resolution in streaming contexts

Task 3: Handle Multi-Step Suspension Scenarios

Account for workflows that may suspend multiple times during a single agent execution.

• Design the suspension/resume flow to handle iterative suspension (suspend → resume → suspend again)
• Track suspension history across resume cycles
• Ensure that multiple workflow tools can suspend in a single agent execution
• Provide clear guidance in suspension metadata about which tool/workflow is suspended

🤖 Prompt for AI agents

Implement agent-level resume capabilities for suspended workflow tools in both
streaming and non-streaming contexts.

Add `resumeGenerate()` method to Agent class in
`packages/core/src/agent/agent.ts`:
- Accept original prompt/messages, suspension resume data (tool call ID to
resume payload mapping), and optional execution options
- Reconstruct execution state from suspension point
- Call underlying workflow's `resume()` method with provided resume data
- Return result with discriminated union type (may complete or suspend again)

Update streaming resume functionality in `packages/core/src/agent/agent.ts`:
- Verify `resumeStream()` handles workflow suspension properly
- Ensure workflow tools resume through `approveToolCall()` and
`declineToolCall()` by passing resume data to workflow's resume method
- Maintain suspension promise resolution

Support multi-step suspension scenarios:
- Handle iterative suspension (suspend → resume → suspend again)
- Track suspension history across resume cycles
- Support multiple workflow tools suspending in single execution
- Include clear tool/workflow identification in suspension metadata

Phase 4: Improve Error Messages and Developer Experience

Update error handling to provide clear guidance when suspension is not properly managed.

Task 1: Replace Generic Promise Error with Actionable Message

Replace the current "promise 'steps' was not resolved" error with suspension-specific guidance.

• In packages/core/src/stream/base/output.ts, enhance the error detection in the flush() method
• If the unresolved promise is related to tool suspension, provide a specific error message
• Error should explain that a workflow tool suspended and guide users to check the result status and use the resume method
• Include documentation links or examples in the error message

Task 2: Add TypeScript Documentation for Suspension APIs

Document the suspension/resume pattern in code comments and type definitions.

• Add JSDoc comments to the agent result types explaining the suspension status and how to handle it
• Document the resume() method with example usage
• Add comments in #convertWorkflowsToTools() explaining the suspension detection logic
• Include inline examples showing: checking for suspension, extracting resume requirements, and calling resume

Task 3: Validate Suspension Schema Compatibility

Add validation to ensure workflow suspend/resume schemas are properly configured when used in agent contexts.

• When workflows are converted to tools, validate that suspended workflows have proper suspendSchema and resumeSchema defined
• Provide helpful errors if schemas are missing or incompatible
• Add warnings if workflow suspension is used without proper storage configuration (needed for workflow state persistence)

🤖 Prompt for AI agents

Improve error messages and developer experience for workflow suspension
handling.

Update error handling in `packages/core/src/stream/base/output.ts`:
- Enhance error detection in `flush()` method for suspension-related unresolved
promises
- Replace generic "promise 'steps' was not resolved" error with
suspension-specific guidance
- Explain that workflow tool suspended and direct users to check result status
and use resume method
- Include documentation links or examples

Add documentation in `packages/core/src/agent/agent.ts` and type files:
- Add JSDoc comments to agent result types explaining suspension status and
handling
- Document `resume()` method with example usage
- Comment suspension detection logic in `#convertWorkflowsToTools()`
- Include inline examples for checking suspension, extracting resume
requirements, and calling resume

Add validation in `packages/core/src/agent/agent.ts` during workflow-to-tool
conversion:
- Validate suspended workflows have `suspendSchema` and `resumeSchema` defined
- Provide helpful errors for missing or incompatible schemas
- Warn if suspension used without proper storage configuration for state
persistence

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Extend agent result types to support workflow suspension in non-streaming
contexts.

Update type definitions in `packages/core/src/agent/agent.types.ts`:
- Add discriminated union for agent results with `status` field ('completed',
'suspended', or 'failed')
- For suspended status: add `suspendedToolCalls` array with tool call IDs,
suspension payloads, and resume schemas
- Add `resume()` method reference to suspended results

Modify agent execution tracking in `packages/core/src/agent/agent.ts`:
- Add execution context flag to distinguish streaming vs non-streaming mode
- Pass context to workflow tool wrapper's execute function
- Store context in execution options or instance variable during `generate()`
and `stream()` calls
===============================================================================

Task: 2

Implement workflow suspension handling in non-streaming generate() contexts to
prevent unresolved promise errors.

Modify workflow tool wrapper in `packages/core/src/agent/agent.ts` within
`#convertWorkflowsToTools()`:
- Detect when `result?.status === 'suspended'` and check execution context
- For non-streaming: store suspension state (tool call ID, suspend payload,
resume schema, suspended step IDs) in tracking structure instead of calling
`suspend?.()`
- For streaming: maintain existing `suspend?.()` behavior
- Return resolved promise with suspension metadata to prevent promise errors

Update `generate()` method in `packages/core/src/agent/agent.ts`:
- Create suspension tracking structure in method scope
- Collect suspension data from workflow tools during execution (tool call ID,
workflow name, suspended step ID, suspend payload, resume schema)
- After execution: check for collected suspensions
- Return result with `status: 'suspended'` and `suspendedToolCalls` array if
suspensions exist, otherwise `status: 'completed'`
- Attach `resume()` method reference to suspended results
- Ensure all promises resolve before returning
===============================================================================

Task: 3

Implement agent-level resume capabilities for suspended workflow tools in both
streaming and non-streaming contexts.

Add `resumeGenerate()` method to Agent class in
`packages/core/src/agent/agent.ts`:
- Accept original prompt/messages, suspension resume data (tool call ID to
resume payload mapping), and optional execution options
- Reconstruct execution state from suspension point
- Call underlying workflow's `resume()` method with provided resume data
- Return result with discriminated union type (may complete or suspend again)

Update streaming resume functionality in `packages/core/src/agent/agent.ts`:
- Verify `resumeStream()` handles workflow suspension properly
- Ensure workflow tools resume through `approveToolCall()` and
`declineToolCall()` by passing resume data to workflow's resume method
- Maintain suspension promise resolution

Support multi-step suspension scenarios:
- Handle iterative suspension (suspend → resume → suspend again)
- Track suspension history across resume cycles
- Support multiple workflow tools suspending in single execution
- Include clear tool/workflow identification in suspension metadata
===============================================================================

Task: 4

Improve error messages and developer experience for workflow suspension
handling.

Update error handling in `packages/core/src/stream/base/output.ts`:
- Enhance error detection in `flush()` method for suspension-related unresolved
promises
- Replace generic "promise 'steps' was not resolved" error with
suspension-specific guidance
- Explain that workflow tool suspended and direct users to check result status
and use resume method
- Include documentation links or examples

Add documentation in `packages/core/src/agent/agent.ts` and type files:
- Add JSDoc comments to agent result types explaining suspension status and
handling
- Document `resume()` method with example usage
- Comment suspension detection logic in `#convertWorkflowsToTools()`
- Include inline examples for checking suspension, extracting resume
requirements, and calling resume

Add validation in `packages/core/src/agent/agent.ts` during workflow-to-tool
conversion:
- Validate suspended workflows have `suspendSchema` and `resumeSchema` defined
- Provide helpful errors for missing or incompatible schemas
- Warn if suspension used without proper storage configuration for state
persistence

💡 Iterate on the plan with: @coderabbitai <feedback>

Example Feedback
- @coderabbitai You can skip phase 3. Add a simple unit test case for phase 2.
- @coderabbitai For assumption 1 go ahead with option 3 and replan.

💬 Have feedback or questions? Drop into our discord or schedule a call!