User Task Executors

User task executors prepare human tasks for the inbox and process user responses.

UserTaskExecutor Interface

abstract class UserTaskExecutor implements SchemaItem {
  /// Schema type identifier (e.g., 'userTask.approval')
  String get schemaType;

  /// Human-readable name
  String get name;

  /// Optional description
  String get description => '';

  /// Execute the user task executor
  ///
  /// Returns WaitForUserTaskResult - this is MANDATORY for user tasks.
  /// User tasks ALWAYS wait for a human signal.
  Future<WaitForUserTaskResult> execute(ExecutionContext context);

  /// Process the user's output when task is completed (optional override)
  Map<String, dynamic> processOutput(Map<String, dynamic> userOutput) => userOutput;
}

Key Difference from TaskExecutor

// TaskExecutor - returns immediately with result
abstract class TaskExecutor {
  Future<TaskResult> execute(ExecutionContext context);
  // Returns: TaskSuccess or TaskFailure
}

// UserTaskExecutor - ALWAYS waits for human signal
abstract class UserTaskExecutor {
  Future<WaitForUserTaskResult> execute(ExecutionContext context);
  // Returns: ALWAYS WaitForUserTaskResult
}

WaitForUserTaskResult

The result returned by user task executors:

class WaitForUserTaskResult extends NodeResult {
  final String signalName;
  final UserTaskConfiguration config;
  final List<WorkflowEffect> effects;  // Side effects before creating task

  WaitForUserTaskResult({
    required this.signalName,
    required this.config,
    this.effects = const [],
  });
}

UserTaskConfiguration

The configuration for creating a user task instance:

class UserTaskConfiguration {
  final String title;
  final String? description;
  final String schemaType;
  final String? assignedToRoleId;
  final String? assignedToUserId;
  final String? assignedToGroupId;
  final UserTaskPriority priority;
  final DateTime? dueDate;
  final Map<String, dynamic> input;

  UserTaskConfiguration({
    required this.title,
    this.description,
    required this.schemaType,
    this.assignedToRoleId,
    this.assignedToUserId,
    this.assignedToGroupId,
    this.priority = UserTaskPriority.normal,
    this.dueDate,
    this.input = const {},
  });
}

ExecutionContext for User Tasks

The context provides access to all execution state:

// Get data from previous node output
final entityType = context.get<String>('entityType');
final entityId = context.getRequired<String>('entityId');

// Get original workflow input
final tenantId = context.getInitial<String>('tenantId');

// Get accumulated output from any previous node
final validation = context.getAny<Map>('validationResult');

// Get node configuration
final template = context.getConfig<String>('template');

// Signal name is provided for user tasks
final signalName = context.signalName!;

Implementing a User Task Executor

Approval Task Executor

class ApprovalTaskExecutor extends UserTaskExecutor {
  static const _schemaType = 'userTask.approval';

  static final typeDescriptor = TypeDescriptor<UserTaskExecutor>(
    schemaType: _schemaType,
    fromJson: (json) => ApprovalTaskExecutor(),
    title: 'Approval Task',
  );

  @override
  String get schemaType => _schemaType;

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

  @override
  Future<WaitForUserTaskResult> execute(ExecutionContext context) async {
    final input = context.input;
    final config = context.config;

    // Build title with template variables
    final title = _resolveTemplate(
      config['title'] as String? ?? 'Approval Required',
      input,
    );

    return WaitForUserTaskResult(
      signalName: context.signalName!,
      config: UserTaskConfiguration(
        title: title,
        description: config['description'] as String?,
        schemaType: schemaType,
        assignedToRoleId: _resolveAssignee(config, input),
        priority: _resolvePriority(config, input),
        dueDate: _resolveDueDate(config),
        input: {
          'entityId': input['entityId'],
          'entityType': input['entityType'],
          'entityName': input['entityName'],
          'submittedBy': input['submittedBy'],
          'submittedAt': input['submittedAt'],
        },
        storeAs: config['storeAs'] as String?,
      ),
    );
  }

  String? _resolveAssignee(Map<String, dynamic> config, Map<String, dynamic> input) {
    // Static assignment from config
    if (config['assignToRole'] != null) {
      return config['assignToRole'] as String;
    }
    // Dynamic assignment from workflow data
    if (config['assignToRoleField'] != null) {
      return input[config['assignToRoleField']] as String?;
    }
    return null;
  }

  UserTaskPriority _resolvePriority(Map<String, dynamic> config, Map<String, dynamic> input) {
    final priority = config['priority'] ?? input['priority'] ?? 'normal';
    return UserTaskPriority.values.firstWhere(
      (p) => p.name == priority,
      orElse: () => UserTaskPriority.normal,
    );
  }

  DateTime? _resolveDueDate(Map<String, dynamic> config) {
    final hoursUntilDue = config['hoursUntilDue'] as int?;
    if (hoursUntilDue != null) {
      return DateTime.now().add(Duration(hours: hoursUntilDue));
    }
    return null;
  }

  String _resolveTemplate(String template, Map<String, dynamic> data) {
    return template.replaceAllMapped(
      RegExp(r'\{\{(\w+)\}\}'),
      (match) => data[match.group(1)]?.toString() ?? '',
    );
  }
}

Document Review Executor

class DocumentReviewTaskExecutor extends UserTaskExecutor {
  static const _schemaType = 'userTask.documentReview';

  static final typeDescriptor = TypeDescriptor<UserTaskExecutor>(
    schemaType: _schemaType,
    fromJson: (json) => DocumentReviewTaskExecutor(),
    title: 'Document Review',
  );

  @override
  String get schemaType => _schemaType;

  @override
  String get name => 'Document Review Task';

  @override
  Future<WaitForUserTaskResult> execute(ExecutionContext context) async {
    final input = context.input;

    return WaitForUserTaskResult(
      signalName: context.signalName!,
      config: UserTaskConfiguration(
        title: 'Review Document: ${input['documentTitle']}',
        description: 'Please review the attached document and provide feedback.',
        schemaType: schemaType,
        assignedToRoleId: input['reviewerRoleId'] as String?,
        priority: UserTaskPriority.normal,
        input: {
          'documentId': input['documentId'],
          'documentTitle': input['documentTitle'],
          'documentUrl': input['documentUrl'],
          'submittedBy': input['submittedBy'],
          'version': input['version'],
        },
      ),
    );
  }
}

TypedUserTaskExecutor

For type-safe input handling with compile-time guarantees:

/// Type-safe base class for user task executors
abstract class TypedUserTaskExecutor<TInput> extends UserTaskExecutor {
  /// Deserialize input from workflow variables
  TInput fromInput(Map<String, dynamic> input);

  /// Type-safe execute method
  Future<WaitForUserTaskResult> executeTyped(
    TInput input,
    ExecutionContext context,
  );

  @override
  Future<WaitForUserTaskResult> execute(ExecutionContext context) async {
    final input = fromInput(context.input);
    return executeTyped(input, context);
  }
}

Example with JsonSerializable

Use @JsonSerializable from json_annotation for input models:

import 'package:json_annotation/json_annotation.dart';

part 'approval_task_input.g.dart';

@JsonSerializable()
class ApprovalTaskInput {
  final String entityId;
  final String entityType;
  final String entityName;
  final int currentLevel;
  final String currentLevelRoleId;
  final String submittedBy;

  ApprovalTaskInput({
    required this.entityId,
    required this.entityType,
    required this.entityName,
    required this.currentLevel,
    required this.currentLevelRoleId,
    required this.submittedBy,
  });

  factory ApprovalTaskInput.fromJson(Map<String, dynamic> json) =>
      _$ApprovalTaskInputFromJson(json);

  Map<String, dynamic> toJson() => _$ApprovalTaskInputToJson(this);
}

// Type-safe executor
class ApprovalTaskExecutor extends TypedUserTaskExecutor<ApprovalTaskInput> {
  static const _schemaType = 'userTask.approval';

  static final typeDescriptor = TypeDescriptor<UserTaskExecutor>(
    schemaType: _schemaType,
    fromJson: (json) => ApprovalTaskExecutor(),
    title: 'Approval Task',
  );

  @override
  String get schemaType => _schemaType;

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

  @override
  ApprovalTaskInput fromInput(Map<String, dynamic> input) =>
      ApprovalTaskInput.fromJson(input);

  @override
  Future<WaitForUserTaskResult> executeTyped(
    ApprovalTaskInput input,
    ExecutionContext context,
  ) async {
    // Fully typed access to input properties!
    return WaitForUserTaskResult(
      signalName: context.signalName!,
      config: UserTaskConfiguration(
        title: 'Approve ${input.entityType}: Level ${input.currentLevel}',
        description: 'Review and approve this ${input.entityType}',
        schemaType: schemaType,
        assignedToRoleId: input.currentLevelRoleId,
        priority: UserTaskPriority.high,
        input: input.toJson(),  // Pass typed data for UI
        storeAs: 'level${input.currentLevel}Decision',
      ),
    );
  }
}

Benefits of TypedUserTaskExecutor

Aspect	UserTaskExecutor	TypedUserTaskExecutor
Input access	context.get<T>('field')	input.field
Type safety	Runtime checks	Compile-time checks
Refactoring	Error-prone	IDE-supported
Auto-complete	None	Full support

When to Use

• Use TypedUserTaskExecutor when:

  • Complex input structures
  • Multi-level approval workflows
  • Input is passed to UI components
  • Building reusable executors

• Use basic UserTaskExecutor when:

  • Simple input (1-2 fields)
  • Quick inline tasks

Using in Workflows

Inline Execute (Recommended for Simple Cases)

builder.userTask(
  'managerApproval',
  name: 'Manager Approval',
  signal: 'manager_approval',
  execute: (ctx) async {
    return UserTaskConfiguration(
      title: 'Approve Request',
      schemaType: 'approval',
      assignedToRoleId: 'managers',
    );
  },
  storeAs: 'approvalResult',
);

Executor-Based

// Register via descriptor
final descriptor = WorkflowDescriptor(
  title: 'User Tasks',
  userTasks: [ApprovalTaskExecutor.typeDescriptor],
);

// Create context with descriptors
final context = RegistryDeserializationContext(
  descriptors: [DefaultWorkflowDescriptor(), descriptor],
);

// Create engine with context and storage
final engine = WorkflowEngine(
  context: context,
  storage: InMemoryStorage(context: context),
);
await engine.initialize();

// Use in workflow
builder.userTask(
  'managerApproval',
  name: 'Manager Approval',
  executor: ApprovalTaskExecutor(),
);

With Effects

class RevisionTaskExecutor extends UserTaskExecutor {
  @override
  String get schemaType => 'userTask.revision';

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

  @override
  Future<WaitForUserTaskResult> execute(ExecutionContext context) async {
    final submitterId = context.getRequired<String>('submitterId');

    return WaitForUserTaskResult(
      signalName: context.signalName!,
      config: UserTaskConfiguration(
        title: 'Revise Document',
        schemaType: schemaType,
        assignedToUserId: submitterId,
      ),
      // Cancel any pending approval tasks before creating revision task
      effects: [const CancelUserTasksEffect()],
    );
  }
}

Builder-Based (with Config)

builder.userTask(
  'managerApproval',
  name: 'Manager Approval',
  signal: 'manager_approval',
  schemaType: 'approvalTask',
  title: 'Approve {{entityType}}: {{entityName}}',
  description: 'Please review and approve this request.',
  assignToRole: 'manager',
  storeAs: 'managerDecision',
);

Completing User Tasks

From Your Application

// User completes task in UI
Future<void> completeUserTask(
  String taskId,
  Map<String, dynamic> response,
) async {
  // Get task details
  final task = await storage.userTaskInstances.getById(taskId);

  // Send signal to workflow using the node ID
  await engine.sendSignal(
    workflowInstanceId: task.workflowInstanceId,
    node: task.nodeId,  // The user task node ID
    payload: response,
  );
}

// Example: Approval decision
await completeUserTask(taskId, {
  'decision': 'approved',
  'comments': 'Looks good, approved!',
  'approvedBy': currentUserId,
});

Task Status Updates

// Claim task (optional)
await storage.userTaskInstances.claim(taskId, userId);

// Complete task
await storage.userTaskInstances.complete(taskId, output);

// Task status flow:
// pending -> claimed -> completed

Assignment Strategies

Role-Based

UserTaskConfiguration(
  assignedToRoleId: 'finance_approvers',
  ...
);

Direct User

UserTaskConfiguration(
  assignedToUserId: specificUserId,
  ...
);

Dynamic (from workflow data)

UserTaskConfiguration(
  // Use get<T> for previous node output
  assignedToUserId: context.get<String>('managerId'),
  ...
);

Best Practices

1. Use templates for titles - Dynamic, contextual titles
2. Include relevant input - Help users make decisions
3. Set appropriate priorities - Guide user attention
4. Use due dates - Track SLAs
5. Process output - Normalize user responses
6. Use effects to cancel previous tasks - Before creating revision/resubmit tasks
7. Use get<T> for previous node output - This is your primary data source
8. Use getInitial<T> for original workflow input - For configuration that persists

Next Steps

• Condition Executors - Gateway conditions
• Workflow Effects - Effect types reference
• Signals - Signal processing
• Approval Workflows - Approval patterns