WorkflowBuilder

WorkflowBuilder is the fluent API for creating executable Workflow objects in code. It is useful for tests, local examples, and simple programmatic definitions. Persisted production templates usually use WorkflowDefinition JSON seeded through storage.

Constructor

WorkflowBuilder(
  String code,
  String name, {
  String? description,
});

final builder = WorkflowBuilder(
  'document-approval',
  'Document Approval',
  description: 'Review and approve a document',
)
    .version(2)
    .withMetadata({'owner': 'quality'});

build() generates the workflow ID as ${code.toLowerCase()}-v${version}.

Methods

Method	Purpose
version(int)	Set the workflow version. Defaults to 1.
withMetadata(Map<String, dynamic>)	Merge arbitrary metadata.
start(String id, {String? name})	Add a start node.
end(String id, {String? name, String? outcome})	Add an end node. outcome is accepted but the current builder stores the node as a normal end node.
task(...)	Add a service task. Requires an executor, inline function, or schema type.
userTask(...)	Add a human task. Uses configuration, a custom executor, or an inline task configuration function.
signalWait(...)	Add an external signal wait node.
oneOf(...)	Exclusive split. Evaluates branches in order.
allOf(...)	Parallel split to all targets.
anyOf(...)	Inclusive split to all targets in the current builder. Race behavior is handled by downstream join/processor semantics.
from(String)	Set the current source node for the next connection.
to(String)	Connect the current source node to a target and make the target current.
connect(String from, String to, {String? label})	Add an explicit edge.
build()	Validate minimum structure and return a Workflow.

Service Tasks

Use inline functions for simple cases:

builder.task(
  'load-document',
  execute: (ctx) async {
    final documentId = ctx.getInitialRequired<String>('documentId');
    final document = await documentService.get(documentId);
    return {
      'documentId': document.id,
      'title': document.title,
    };
  },
);

Use a TaskExecutor when the workflow may be serialized and loaded through descriptors:

class SendNotificationExecutor extends TaskExecutor {
  static const type = 'task.notification.send';

  static final typeDescriptor = TypeDescriptor<TaskExecutor>(
    schemaType: type,
    fromJson: (_) => SendNotificationExecutor(),
    title: 'Send Notification',
  );

  @override
  String get schemaType => type;

  @override
  String get name => 'Send Notification';

  @override
  Future<TaskResult> execute(ExecutionContext context) async {
    final subject = context.getAny<String>('title') ?? 'Workflow update';
    await notificationService.send(subject: subject);

    return TaskSuccess([
      SetOutputEffect(output: {
        'notificationSent': true,
        'sentAt': DateTime.now().toUtc().toIso8601String(),
      }),
    ]);
  }
}

builder.task(
  'notify',
  executor: SendNotificationExecutor(),
  storeAs: 'notification',
);

The builder stores the executor's schemaType in ServiceTaskNodeConfiguration. At runtime, TaskNodeProcessor resolves the executor from the engine's WorkflowTypeResolver.

User Tasks

A configuration-only user task uses DefaultUserTaskExecutor:

builder.userTask(
  'approval',
  signal: 'approval_decision',
  schemaType: 'userTask.approval',
  title: 'Approve {{title}}',
  assignToRole: 'document-approvers',
  storeAs: 'approval',
);

A custom user task executor prepares the inbox task and later processes completion:

class ApprovalUserTaskExecutor extends UserTaskExecutor {
  static const type = 'userTask.approval';

  static final typeDescriptor = TypeDescriptor<UserTaskExecutor>(
    schemaType: type,
    fromJson: (_) => ApprovalUserTaskExecutor(),
    title: 'Approval Task',
  );

  @override
  String get schemaType => type;

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

  @override
  Future<WaitForUserTaskResult> execute(ExecutionContext context) async {
    final title = context.getAny<String>('title') ?? 'item';

    return WaitForUserTaskResult(
      signalName: context.signalName ?? 'approval_decision',
      config: UserTaskConfiguration(
        title: 'Approve $title',
        schemaType: schemaType,
        assignment: TaskAssignment.role('document-approvers'),
        input: context.accumulated,
      ),
    );
  }

  @override
  UserTaskCompletionResult processCompletion(
    Map<String, dynamic> userOutput,
    Map<String, dynamic> taskInput,
    ExecutionContext context,
  ) {
    final approved = userOutput['approved'] == true;

    return UserTaskCompletionResult.routeTo(
      approved ? 'approve' : 'reject',
      output: {
        'approved': approved,
        'completedBy': userOutput['completedBy'],
      },
    );
  }
}

Wire it through the builder:

builder.userTask(
  'approval',
  signal: 'approval_decision',
  executor: ApprovalUserTaskExecutor(),
  storeAs: 'approval',
);

For descriptor-backed loading, include ApprovalUserTaskExecutor.typeDescriptor in a WorkflowDescriptor.

Signal Wait

builder.signalWait(
  'wait-for-payment',
  signal: 'payment_completed',
  storeAs: 'payment',
);

signalWait creates a SignalEventNodeConfiguration. The engine stores the received payload under storeAs.

Branches

builder.oneOf('route', [
  Branch.when("approval.approved == true", then: 'approved-path'),
  Branch.whenEquals('approval.status', 'rejected', then: 'rejected-path'),
  Branch.whenTrue('requiresReview', then: 'review-path'),
  Branch.whenNotNull('escalation', then: 'escalate-path'),
  Branch.otherwise(then: 'fallback-path'),
]);

Branch.whenFn is convenient for code-built workflows:

Branch.whenFn(
  (vars) => (vars['amount'] as num? ?? 0) > 10000,
  then: 'high-value',
  label: 'High value',
)

Inline function conditions are not meaningfully serializable. Use a registered ConditionExecutor or expression condition for stored definitions.

Edges

The builder auto-connects from the current node to the next added node. You can also control flow explicitly:

builder
  .from('route')
  .to('approved-path')
  .connect('approved-path', 'approved-end', label: 'Approved');

Branch methods add edges from the split node to each branch target.

Complete Example

final workflow = WorkflowBuilder('document-approval', 'Document Approval')
    .start('start')
    .task('load-document', execute: (ctx) async {
      final documentId = ctx.getInitialRequired<String>('documentId');
      return {
        'documentId': documentId,
        'title': 'Document $documentId',
      };
    })
    .userTask(
      'approval',
      signal: 'approval_decision',
      schemaType: 'userTask.approval',
      title: 'Approve {{title}}',
      assignToRole: 'document-approvers',
      storeAs: 'approval',
    )
    .oneOf('route', [
      Branch.whenEquals('approval.approved', true, then: 'approve'),
      Branch.otherwise(then: 'reject'),
    ])
    .task('approve', execute: (ctx) async {
      return {'finalStatus': 'approved'};
    })
    .end('approved')
    .from('route')
    .to('reject')
    .task('reject', execute: (ctx) async {
      return {'finalStatus': 'rejected'};
    })
    .end('rejected')
    .build();

Validation

The builder itself checks the minimum structure:

• At least one start() node.
• At least one end() node.
• No duplicate node IDs.

WorkflowEngine.registerWorkflow() and loadWorkflowDefinition() run the workflow model validation before execution.

See Also

• WorkflowEngine
• TaskExecutor
• UserTaskExecutor
• Loading and Registration