Workflow Effects

Effects capture side effects declaratively, enabling pure executors that are easy to test and reason about.

Overview

Instead of calling mutation methods directly, executors return WorkflowEffect objects that describe what should happen. The engine's EffectProcessor interprets these effects and applies them to storage.

Benefits

• Pure executors: Node executors return effects instead of mutating state
• Testability: Effects can be asserted without mocking storage
• Batching: Multiple effects can be processed together
• Replay: Effects form an audit trail that can be replayed

// Instead of calling mutation methods...
await storage.userTaskInstances.cancel(taskId);
await storage.events.record(event);

// Return effects declaratively
return TaskSuccess(
  output: {'processed': true},
  effects: [
    const CancelUserTasksEffect(),
    RecordEventEffect(event: event),
  ],
);

Effect Hierarchy

---

Output Effects

SetOutputEffect

Class Signature

Constructor Parameters

Parameter	Type	Required	Description
output	Map<String, dynamic>	Yes	The output data to merge into workflow output
path	String?	No	Optional path to namespace the output

Set or merge output data into the workflow instance.

Usage:

// Merge at root level
SetOutputEffect(output: {'validated': true})

// Namespace under a path to prevent collisions
SetOutputEffect(
  output: {'count': 10, 'total': 500},
  path: 'processingStats',
)
// Result: { processingStats: { count: 10, total: 500 } }

SetVariableEffect

Class Signature

Constructor Parameters

Parameter	Type	Required	Description
name	String	Yes	Variable name
value	dynamic	Yes	Variable value
scope	VariableScope	Yes	Variable scope (local, inherited, shared)

Set or update a workflow variable with explicit scoping.

Usage:

// Set a local variable
SetVariableEffect(name: 'retryCount', value: 3, scope: VariableScope.local)

// Convenience factories
SetVariableEffect.local('retryCount', 3)
SetVariableEffect.inherited('approvalChain', chain)

RouteToPortEffect

Class Signature

Route execution to a specific output port. Used by executors that support multiple outcome paths (e.g., subflow nodes with success/failure/timeout ports).

Usage:

// Route to the success port
RouteToPortEffect(portId: 'success')

// Route to a custom port
RouteToPortEffect(portId: 'needsReview')

---

Status Effects

UpdateStatusEffect

Update the workflow instance status.

class UpdateStatusEffect extends WorkflowEffect {
  const UpdateStatusEffect({required this.status});

  final WorkflowStatus status;
}

CompleteWorkflowEffect

Complete the workflow with final output.

class CompleteWorkflowEffect extends WorkflowEffect {
  const CompleteWorkflowEffect({this.output = const {}});

  /// Final output to merge before completing
  final Map<String, dynamic> output;
}

FailWorkflowEffect

Fail the workflow with an error.

class FailWorkflowEffect extends WorkflowEffect {
  const FailWorkflowEffect({
    required this.error,
    this.nodeId,
  });

  /// The error that caused the failure
  final WorkflowError error;

  /// The node where the failure occurred
  final String? nodeId;
}

---

Token Effects

Tokens track execution position in the workflow. These effects manage token lifecycle.

CreateTokensEffect

Class Signature

Constructor Parameters

Parameter	Type	Required	Description
targetNodeIds	List<String>	Yes	Target node IDs to create tokens for
fromNodeId	String	Yes	The node ID that is the source of these tokens
fromTokenId	String?	No	The token ID that triggered this creation
isParallelSplit	bool	No	Whether this is a parallel split (creates SplitTokens)
output	Map<String, dynamic>	No	Output to carry on the new tokens
splitContext	SplitContext?	No	Split context for parallel/inclusive splits
splitType	SplitType?	No	Type of split (exclusive, parallel, inclusive)

Create new tokens for target nodes. Used when continuing to next nodes or splitting into parallel branches.

ConsumeTokenEffect

Consume a token (mark as terminal). Used when a token reaches an end node or is merged at a join.

class ConsumeTokenEffect extends WorkflowEffect {
  const ConsumeTokenEffect({required this.tokenId});

  final String tokenId;
}

ConsumeJoinTokensEffect

Consume all tokens at a join node when all branches have arrived.

class ConsumeJoinTokensEffect extends WorkflowEffect {
  const ConsumeJoinTokensEffect({
    required this.joinNodeId,
    required this.tokenIds,
  });

  final String joinNodeId;
  final List<String> tokenIds;
}

DropTokenEffect

Drop a token (late arrival at join or explicit drop).

class DropTokenEffect extends WorkflowEffect {
  const DropTokenEffect({
    required this.tokenId,
    this.reason,
  });

  final String tokenId;
  final String? reason;
}

WaitTokenEffect

Mark a token as waiting for an external event.

class WaitTokenEffect extends WorkflowEffect {
  const WaitTokenEffect({required this.tokenId});

  final String tokenId;
}

ActivateTokenEffect

Activate a waiting token (resume from wait state).

class ActivateTokenEffect extends WorkflowEffect {
  const ActivateTokenEffect({required this.tokenId});

  final String tokenId;
}

---

User Task Effects

CreateUserTaskEffect

Create a user task instance.

class CreateUserTaskEffect extends WorkflowEffect {
  const CreateUserTaskEffect({
    required this.taskId,
    required this.nodeId,
    required this.signalName,
    required this.title,
    required this.schemaType,
    this.description,
    this.assignedToRoleId,
    this.assignedToUserId,
    this.assignedToGroupId,
    this.priority = UserTaskPriority.normal,
    this.dueAt,
    this.input = const {},
    this.storeAs,
  });
}

CancelUserTasksEffect

Class Signature

Constructor Parameters

Parameter	Type	Required	Description
nodeId	String?	No	Cancel only tasks for this node. If null, cancel all.
reason	String?	No	Optional reason for cancellation

Cancel pending user tasks for a workflow instance.

Usage:

// Cancel all pending user tasks
const CancelUserTasksEffect()

// Cancel only tasks for a specific node
CancelUserTasksEffect(nodeId: 'approvalNode')

Common Use Case: Cancel Before Revision

class RevisionTaskExecutor extends UserTaskExecutor {
  @override
  Future<NodeResult> execute(ExecutionContext context) async {
    return WaitForUserTaskResult(
      signalName: 'revision_submitted',
      config: UserTaskConfiguration(
        title: 'Revise Document',
        schemaType: 'revision',
        assignedToUserId: context.getRequired<String>('submitterId'),
      ),
      // Cancel any pending approval tasks before creating revision task
      effects: [const CancelUserTasksEffect()],
    );
  }
}

ReassignUserTaskEffect

Class Signature

Constructor Parameters

Parameter	Type	Required	Description
taskId	String	Yes	The user task ID to reassign
newAssignedToUserId	String?	No	New user ID to assign to
newAssignedToRoleId	String?	No	New role ID to assign to
newAssignedToGroupId	String?	No	New group ID to assign to
reason	String?	No	Reason for reassignment
reassignedBy	String?	No	Who initiated the reassignment

Reassign a user task to a different user, role, or group.

Usage:

// Reassign to a specific user
ReassignUserTaskEffect(
  taskId: 'task-123',
  newAssignedToUserId: 'manager-456',
  reason: 'Original assignee on leave',
  reassignedBy: 'admin-789',
)

// Reassign to a role
ReassignUserTaskEffect(
  taskId: 'task-123',
  newAssignedToRoleId: 'senior-managers',
  reason: 'Escalation due to SLA breach',
)

---

Notification Effects

SendNotificationEffect

Class Signature

Constructor Parameters

Parameter	Type	Required	Description
notificationType	String	Yes	Notification type identifier (e.g., "task_assigned")
recipientUserIds	List<String>	Yes	User IDs to notify
title	String	Yes	Notification title
recipientRoleIds	List<String>?	No	Role IDs to notify (optional)
recipientGroupIds	List<String>?	No	Group IDs to notify (optional)
body	String?	No	Notification body (optional)
data	Map<String, dynamic>	No	Additional data payload
channels	List<NotificationChannel>	No	Delivery channels
priority	NotificationPriority	No	Notification priority

Send a notification to users, roles, or groups across multiple channels.

NotificationChannel

enum NotificationChannel { inApp, email, push, sms }

NotificationPriority

enum NotificationPriority { low, normal, high, urgent }

Usage:

SendNotificationEffect(
  notificationType: 'approval_required',
  recipientRoleIds: ['managers'],
  recipientUserIds: [],
  title: 'New Approval Request',
  body: 'Order #1234 requires your approval',
  channels: [NotificationChannel.inApp, NotificationChannel.email],
  priority: NotificationPriority.high,
  data: {'orderId': '1234', 'amount': 5000},
)

---

Event Effects

RecordEventEffect

Record a workflow event to the event log.

class RecordEventEffect extends WorkflowEffect {
  const RecordEventEffect({required this.event});

  final WorkflowEvent event;
}

Usage:

RecordEventEffect(
  event: WorkflowEvent.custom(
    instanceId: context.workflowInstanceId,
    nodeId: context.nodeId,
    eventType: 'approval_completed',
    data: {'decision': 'approved', 'approvedBy': userId},
  ),
)

EmitErrorEffect

Emit an error to the error stream.

class EmitErrorEffect extends WorkflowEffect {
  const EmitErrorEffect({
    required this.instanceId,
    required this.error,
  });

  final String instanceId;
  final WorkflowError error;
}

---

Scheduling Effects

ScheduleExecutionEffect

Schedule execution of target nodes. Used internally by the orchestrator.

class ScheduleExecutionEffect extends WorkflowEffect {
  const ScheduleExecutionEffect({
    required this.nodeIds,
    required this.tokenIds,
  });

  final List<String> nodeIds;
  final List<String> tokenIds;
}

ScheduleRetryEffect

Schedule a retry with delay.

class ScheduleRetryEffect extends WorkflowEffect {
  const ScheduleRetryEffect({
    required this.nodeId,
    required this.tokenId,
    required this.delay,
    required this.attemptNumber,
  });

  final String nodeId;
  final String tokenId;
  final Duration delay;
  final int attemptNumber;
}

StoreAttemptNumberEffect

Store the attempt number for retry tracking.

class StoreAttemptNumberEffect extends WorkflowEffect {
  const StoreAttemptNumberEffect({
    required this.nodeId,
    required this.attemptNumber,
  });

  final String nodeId;
  final int attemptNumber;
}

---

Cleanup Effects

CleanupResourcesEffect

Clean up workflow resources on termination. Deactivates all tokens and cancels pending user tasks.

class CleanupResourcesEffect extends WorkflowEffect {
  const CleanupResourcesEffect();
}

---

Using Effects in Executors

In TaskExecutor

class MyTaskExecutor extends TaskExecutor {
  @override
  String get schemaType => 'task.myTask';

  @override
  String get name => 'My Task';

  @override
  Future<TaskResult> execute(ExecutionContext context) async {
    final result = await processData(context.input);

    return TaskSuccess(
      output: {'processed': true},
      effects: [
        // Set namespaced output
        SetOutputEffect(
          output: {'stats': result.stats},
          path: 'taskResult',
        ),
        // Record audit event
        RecordEventEffect(
          event: WorkflowEvent.custom(
            instanceId: context.workflowInstanceId,
            nodeId: context.nodeId,
            eventType: 'task_completed',
            data: {'itemCount': result.count},
          ),
        ),
      ],
    );
  }
}

In NodeProcessor

class MyNodeProcessor extends NodeProcessor {
  @override
  Future<NodeResult> execute(ExecutionContext context) async {
    // Process node...

    return ContinueResult.single(
      'nextNode',
      output: {'processed': true},
      effects: [
        const CancelUserTasksEffect(),
        RecordEventEffect(event: ...),
      ],
    );
  }
}

In UserTaskExecutor

class ApprovalExecutor extends UserTaskExecutor {
  @override
  Future<NodeResult> execute(ExecutionContext context) async {
    return WaitForUserTaskResult(
      signalName: 'approval_decision',
      config: UserTaskConfiguration(
        title: 'Approve Request',
        schemaType: 'approval',
        assignedToRoleId: 'managers',
      ),
      // Effects run before user task is created
      effects: [
        RecordEventEffect(
          event: WorkflowEvent.custom(
            instanceId: context.workflowInstanceId,
            nodeId: context.nodeId,
            eventType: 'approval_requested',
            data: {},
          ),
        ),
      ],
    );
  }
}

---

Best Practices

1. Keep executors pure: Return effects instead of calling mutation methods
2. Use SetOutputEffect with paths: Namespace output to prevent collisions
3. Cancel tasks before creating new ones: Use CancelUserTasksEffect when transitioning between user tasks
4. Record events for audit: Use RecordEventEffect for important state changes
5. Order effects appropriately: Effects are processed in order

See Also

• Task Executors - Using effects in tasks
• Node Results - Result types with effects
• Data Flow - How data moves through workflows