How to resume workflow to validate data

Hi

I am attempting to create a simple Elsa test application consisting of three sequential activities. Each activity represents a task:

1. Task 1 prompts the user to enter a number via the console.

The workflow should suspend until the user provides input.

When resumed, Task 1 should validate whether the input is a valid number.

If the input is invalid, the task should request the value again and suspend once more.

If the input is valid, the workflow should continue to Task 2.

2. Task 2 behaves the same as Task 1, but stores the result as a second number (Number2) and then proceeds to Task 3.

3. Task 3 multiplies the values collected in Task 1 and Task 2 and outputs the result to the console.

For Tasks 1 and 2, the expected behavior is:

Prompt the user for a number.

Suspend the workflow.

Resume when input is supplied externally.

Validate the input.

Either repeat the request (if invalid) or proceed (if valid).

I am currently experiencing issues with this flow and the resume behavior. I would greatly appreciate guidance on what I might be doing incorrectly.

I am using Elsa 3.5.1.0 and .Net 9, SQL Server 16.0.4215.2

Below is my full test application for reference.

using Elsa.EntityFrameworkCore.Extensions;
using Elsa.EntityFrameworkCore.Modules.Management;
using Elsa.EntityFrameworkCore.Modules.Runtime;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Activities;
using Elsa.Workflows.Attributes;
using Elsa.Workflows.Management;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Options;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Serilog;
using Serilog.Events;

namespace Test.Elsa;

[Activity("Task1", "Get first number from user", Kind = ActivityKind.Task)]
public class Task1Activity : Activity
{
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var logger = context.GetService>();

// Log ALL execution attempts - this should be hit on resume
logger?.LogInformation("═══════════════════════════════════════════════════════");
logger?.LogInformation("[Task1] ExecuteAsync CALLED. ActivityId: {ActivityId}, NodeId: {NodeId}",
context.Id, context.NodeId);
logger?.LogInformation("[Task1] WorkflowExecutionContext.Input: {HasInput}, Keys: {Keys}",
context.WorkflowExecutionContext.Input != null,
context.WorkflowExecutionContext.Input?.Keys != null
? string.Join(", ", context.WorkflowExecutionContext.Input.Keys)
: "null");
logger?.LogInformation("═══════════════════════════════════════════════════════");

var workflowInput = context.WorkflowInput;

// FIRST: Check if we already completed this task
// If Number1 exists, we've already completed and should just complete again
var hasNumber1 = workflowInput.ContainsKey("Number1");
if (hasNumber1)
{
logger?.LogInformation("[Task1] Already completed with Number1={Number1}, completing again", workflowInput["Number1"]);
await context.CompleteActivityAsync();
return;
}

// Get user input from WorkflowExecutionContext.Input (passed via RunInstanceAsync when resuming)
string? userInput = null;
if (context.WorkflowExecutionContext.Input != null)
{
if (context.WorkflowExecutionContext.Input.TryGetValue("UserInput", out var inputObj))
{
userInput = inputObj?.ToString();
if (!string.IsNullOrEmpty(userInput))
{
workflowInput["UserInput"] = userInput;
}
}
}

// Also try WorkflowInput as fallback
if (string.IsNullOrEmpty(userInput) && workflowInput.TryGetValue("UserInput", out var inputValue))
{
userInput = inputValue?.ToString();
}

// Use flag-based approach like ConfirmRequestActivity
// FIRST ENTRY: Check if we've already created the bookmark
bool firstEntry = !workflowInput.ContainsKey("Task1Pending");

if (firstEntry)
{
// First entry: prompt for input and create bookmark
workflowInput["Task1Pending"] = true;
logger?.LogInformation("Task 1: Created bookmark, Please enter a number:");
context.CreateBookmark("WaitForTask1Input");
return;
}

// RESUME ENTRY: We have the flag, so we're resuming
workflowInput.Remove("Task1Pending");

// If we have user input, we're resuming - validate it
if (!string.IsNullOrEmpty(userInput))
{
logger?.LogInformation("[Task1] Resuming with input: '{Input}'", userInput);

// Validate the input
if (string.IsNullOrWhiteSpace(userInput) || !int.TryParse(userInput, out var number))
{
logger?.LogInformation("Task 1: Invalid input. Please enter a valid number:");
workflowInput.Remove("UserInput"); // Clear the invalid input
context.CreateBookmark("WaitForTask1Input");
return; // Suspend and wait for new input
}

// Valid number entered
workflowInput["Number1"] = number;
workflowInput.Remove("UserInput"); // Clear the input after processing
logger?.LogInformation("Task 1: Number {Number} accepted.", number);
await context.CompleteActivityAsync();
return; // Activity completed
}

// If we're resuming but don't have input, something went wrong
logger?.LogWarning("[Task1] Resuming but no user input found");
workflowInput.Remove("Task1Pending");
context.CreateBookmark("WaitForTask1Input");
return;

}
}

[Activity("Task2", "Get second number from user", Kind = ActivityKind.Task)]
public class Task2Activity : Activity
{
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var logger = context.GetService>();

// Log ALL execution attempts - this should be hit on resume
logger?.LogInformation("═══════════════════════════════════════════════════════");
logger?.LogInformation("[Task2] ExecuteAsync CALLED. ActivityId: {ActivityId}, NodeId: {NodeId}",
context.Id, context.NodeId);
logger?.LogInformation("[Task2] WorkflowExecutionContext.Input: {HasInput}, Keys: {Keys}",
context.WorkflowExecutionContext.Input != null,
context.WorkflowExecutionContext.Input?.Keys != null
? string.Join(", ", context.WorkflowExecutionContext.Input.Keys)
: "null");
logger?.LogInformation("═══════════════════════════════════════════════════════");

var workflowInput = context.WorkflowInput;

// FIRST: Check if we already completed this task
// If Number2 exists, we've already completed and should just complete again
var hasNumber2 = workflowInput.ContainsKey("Number2");
if (hasNumber2)
{
logger?.LogInformation("[Task2] Already completed with Number2={Number2}, completing again", workflowInput["Number2"]);
await context.CompleteActivityAsync();
return;
}

// When resuming, input might be in WorkflowExecutionContext.Input or WorkflowInput
string? userInput = null;

// Try to get from WorkflowInput first
if (workflowInput.TryGetValue("UserInput", out var inputValue))
{
userInput = inputValue?.ToString();
}

// If not found, try WorkflowExecutionContext.Input (passed via RunInstanceAsync)
// BUT only if we don't have Number2 (meaning we're actually waiting for input)
if (context.WorkflowExecutionContext.Input != null)
{
if (string.IsNullOrEmpty(userInput) && context.WorkflowExecutionContext.Input.TryGetValue("UserInput", out var inputObj))
{
userInput = inputObj?.ToString() ?? "";
if (!string.IsNullOrEmpty(userInput))
{
workflowInput["UserInput"] = userInput;
}
}
}

// If we have user input, we're resuming - validate it
if (!string.IsNullOrEmpty(userInput))
{
logger?.LogInformation("[Task2] Resuming with input: '{Input}'", userInput);
// Validate the input
if (string.IsNullOrWhiteSpace(userInput) || !int.TryParse(userInput, out var number))
{
logger?.LogInformation("Task 2: Invalid input. Please enter a valid number:");
workflowInput.Remove("UserInput"); // Clear the invalid input
context.CreateBookmark("WaitForTask2Input");
return;
}

// Valid number entered
workflowInput["Number2"] = number;
workflowInput.Remove("UserInput"); // Clear the input after processing
logger?.LogInformation($"Task 2: Number {number} accepted.");
await context.CompleteActivityAsync();
return;
}
else
{
workflowInput.Remove("UserInput");
logger?.LogInformation("Task 2: Created bookmark, Please enter a second number:");
context.CreateBookmark("WaitForTask2Input");
return;
}
}
}

[Activity("Task3", "Multiply the two numbers", Kind = ActivityKind.Task)]
public class Task3Activity : Activity
{
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var logger = context.GetService>();
logger?.LogInformation("[Task3] ExecuteAsync called");

var workflowInput = context.WorkflowInput;

var number1 = workflowInput.GetValue("Number1");
var number2 = workflowInput.GetValue("Number2");

var result = number1 * number2;
logger?.LogInformation("Task 3: {Number1} × {Number2} = {Result}", number1, number2, result);

await context.CompleteActivityAsync();
}
}

public class WorkflowResumeService : BackgroundService
{
private readonly IServiceProvider _serviceProvider;
private readonly ILogger _logger;

public WorkflowResumeService(IServiceProvider serviceProvider, ILogger logger)
{
_serviceProvider = serviceProvider;
_logger = logger;
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
try
{
using var scope = _serviceProvider.CreateScope();
var workflowInstanceStore = scope.ServiceProvider.GetRequiredService();
var workflowRuntime = scope.ServiceProvider.GetRequiredService();

// Find suspended workflows - filter by SubStatus
var allInstances = await workflowInstanceStore.FindManyAsync(
new WorkflowInstanceFilter(),
stoppingToken);

// Filter for suspended workflows - SubStatus is an enum on WorkflowInstance
var suspended = allInstances.Where(i => i.SubStatus.ToString() == "Suspended");

foreach (var instance in suspended)
{
if (stoppingToken.IsCancellationRequested)
break;

var bookmarks = instance.WorkflowState?.Bookmarks?.ToList() ?? new();
if (!bookmarks.Any())
continue;

// Log all bookmarks for debugging BEFORE getting user input
_logger.LogInformation("All bookmarks for workflow {InstanceId} BEFORE input: {BookmarkList}",
instance.Id,
string.Join(", ", bookmarks.Select(b => $"{b.Name} (ActivityId: {b.ActivityId}, CreatedAt: {b.CreatedAt})")));

// Get the MOST RECENT bookmark (the one that was created last) BEFORE getting user input
var bookmarkBeforeInput = bookmarks
.OrderByDescending(b => b.CreatedAt)
.FirstOrDefault();

if (bookmarkBeforeInput == null)
continue;

_logger.LogInformation("Found suspended workflow {InstanceId} with bookmark {BookmarkName} (ActivityId: {ActivityId}, CreatedAt: {CreatedAt})",
instance.Id, bookmarkBeforeInput.Name, bookmarkBeforeInput.ActivityId, bookmarkBeforeInput.CreatedAt);

// Wait for user input
Console.WriteLine($"\n[Workflow {instance.Id}] Waiting for input for bookmark: {bookmarkBeforeInput.Name}");
Console.Write("Enter value: ");
var userInput = Console.ReadLine();

if (string.IsNullOrWhiteSpace(userInput))
continue;

// IMPORTANT: Re-query the workflow instance and bookmarks AFTER getting user input
// The workflow state may have changed (e.g., Activity 2 created a new bookmark)
var updatedInstance = await workflowInstanceStore.FindAsync(
new WorkflowInstanceFilter { Id = instance.Id },
stoppingToken);

if (updatedInstance == null)
{
_logger.LogWarning("Workflow instance {InstanceId} not found after input", instance.Id);
continue;
}

var updatedBookmarks = updatedInstance.WorkflowState?.Bookmarks?.ToList() ?? new();

// Get the MOST RECENT bookmark AFTER getting user input
// This ensures we're resuming from the correct activity (e.g., Activity 2 if it just created a bookmark)
var bookmark = updatedBookmarks
.OrderByDescending(b => b.CreatedAt)
.FirstOrDefault();

if (bookmark == null)
{
_logger.LogWarning("No bookmarks found for workflow {InstanceId} after input", instance.Id);
continue;
}

_logger.LogInformation("Using bookmark AFTER input: {BookmarkName} (ActivityId: {ActivityId}, CreatedAt: {CreatedAt})",
bookmark.Name, bookmark.ActivityId, bookmark.CreatedAt);

// Log all bookmarks for debugging AFTER getting user input
_logger.LogInformation("All bookmarks for workflow {InstanceId} AFTER input: {BookmarkList}",
instance.Id,
string.Join(", ", updatedBookmarks.Select(b => $"{b.Name} (ActivityId: {b.ActivityId}, CreatedAt: {b.CreatedAt})")));

_logger.LogInformation("═══════════════════════════════════════════════════════");
_logger.LogInformation("RESUMING workflow {InstanceId} from bookmark {BookmarkId} (Name: {BookmarkName}, ActivityId: {ActivityId}, ActivityInstanceId: {ActivityInstanceId}) with input: {Input}",
instance.Id, bookmark.Id, bookmark.Name, bookmark.ActivityId, bookmark.ActivityInstanceId, userInput);
_logger.LogInformation("═══════════════════════════════════════════════════════");

// Use RunInstanceAsync with BookmarkId to resume from the specific bookmark
// This targets the exact bookmark by ID, ensuring the correct activity resumes
var client = await workflowRuntime.CreateClientAsync(instance.Id);

_logger.LogInformation("Resuming workflow {InstanceId} from bookmark {BookmarkId} (Name: {BookmarkName}, ActivityId: {ActivityId}, ActivityInstanceId: {ActivityInstanceId}) with input: {Input}",
instance.Id, bookmark.Id, bookmark.Name, bookmark.ActivityId, bookmark.ActivityInstanceId, userInput);

var result = await client.RunInstanceAsync(
new global::Elsa.Workflows.Runtime.Messages.RunWorkflowInstanceRequest
{
BookmarkId = bookmark.Id, // Target the SPECIFIC bookmark by ID
Input = new Dictionary
{
["UserInput"] = userInput
},
Properties = new Dictionary
{
["UserInput"] = userInput
}
},
stoppingToken);

_logger.LogInformation("RunInstanceAsync result: SubStatus={SubStatus}, Status={Status}",
result?.SubStatus, result?.Status);

_logger.LogInformation("═══════════════════════════════════════════════════════");
_logger.LogInformation("RESUMED workflow {InstanceId} from bookmark {BookmarkId}", instance.Id, bookmark.Id);
_logger.LogInformation("═══════════════════════════════════════════════════════");
}

await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken);
}
catch (Exception ex)
{
_logger.LogError(ex, "Error in WorkflowResumeService");
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
}
}
}
}

public class NumberMultiplicationWorkflow : WorkflowBase
{
protected override void Build(IWorkflowBuilder builder)
{
builder.Root = new Sequence
{
Activities =
{
new Task1Activity(),
new Task2Activity(),
new Task3Activity()
}
};
}
}

class Program
{
static async Task Main(string[] args)
{
Console.WriteLine("╔══════════════════════════════════════════════════════════════╗");
Console.WriteLine("║ ║");
Console.WriteLine("║ ELSA BOOKMARK TEST APPLICATION ║");
Console.WriteLine("║ ║");
Console.WriteLine("╚══════════════════════════════════════════════════════════════╝");
Console.WriteLine();

// Configure Serilog to remove source information
Log.Logger = new LoggerConfiguration()
.MinimumLevel.Information()
.MinimumLevel.Override("Microsoft", LogEventLevel.Warning)
.MinimumLevel.Override("System", LogEventLevel.Warning)
.MinimumLevel.Override("Microsoft.EntityFrameworkCore", LogEventLevel.Warning)
.MinimumLevel.Override("Microsoft.EntityFrameworkCore.Database.Command", LogEventLevel.Warning)
.MinimumLevel.Override("Elsa.Mediator", LogEventLevel.Warning)
.MinimumLevel.Override("Elsa.Mediator.Middleware.Command.Components.CommandLoggingMiddleware", LogEventLevel.Warning)
.MinimumLevel.Override("Microsoft.Hosting.Lifetime", LogEventLevel.Warning)
.WriteTo.Console(outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj}{NewLine}{Exception}")
.CreateLogger();

// Use in-memory database for testing
var connectionString = "Server=localhost,1433;Database=Lumo;User ID=SA;Password=XXXXXXXX;TrustServerCertificate=True;";

var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddSerilog();

builder.Services
.AddElsa(elsa =>
{
elsa.UseWorkflowRuntime(runtime =>
{
runtime.UseEntityFrameworkCore(ef => ef.UseSqlServer(connectionString));
});
elsa.UseWorkflowManagement(management =>
{
management.UseEntityFrameworkCore(ef => ef.UseSqlServer(connectionString));
});
})
.AddWorkflowsFrom();

// Add the resume service
builder.Services.AddHostedService();

var host = builder.Build();

// Ensure database is created - Elsa will handle this automatically
// We can skip manual database creation as Elsa handles it

// Start the host first to ensure all services are initialized
await host.StartAsync();

// Get services for workflow management
var workflowDefinitionStore = host.Services.GetRequiredService();
var workflowDefinitionService = host.Services.GetRequiredService();
var workflowDispatcher = host.Services.GetRequiredService();

// Find the workflow definition by DefinitionId (class name)
// In Elsa 3.x, workflows discovered via AddWorkflowsFrom are automatically registered
global::Elsa.Workflows.Management.Entities.WorkflowDefinition? workflowDefinition = null;

// Try to find by DefinitionId first
workflowDefinition = await workflowDefinitionStore.FindAsync(
new global::Elsa.Workflows.Management.Filters.WorkflowDefinitionFilter
{
DefinitionId = "NumberMultiplicationWorkflow"
});

// If not found, try by Name
if (workflowDefinition == null)
{
workflowDefinition = await workflowDefinitionStore.FindAsync(
new global::Elsa.Workflows.Management.Filters.WorkflowDefinitionFilter
{
Name = "NumberMultiplicationWorkflow"
});
}

if (workflowDefinition == null)
{
// List all available workflows for debugging
var allWorkflows = await workflowDefinitionStore.FindManyAsync(
new global::Elsa.Workflows.Management.Filters.WorkflowDefinitionFilter(),
CancellationToken.None);

Console.WriteLine($"\nAvailable workflows in database: {allWorkflows.Count()}");
foreach (var wf in allWorkflows)
{
Console.WriteLine($" - DefinitionId: {wf.DefinitionId}, Name: {wf.Name}, Id: {wf.Id}");
}

Console.WriteLine("\n⚠️ Workflow definition not found in database.");
Console.WriteLine("Workflows discovered via AddWorkflowsFrom may need to be published to the database.");
Console.WriteLine("Press any key to exit...");
Console.ReadKey();
await host.StopAsync();
return;
}

Console.WriteLine($"✓ Found workflow definition: {workflowDefinition.DefinitionId} (Name: {workflowDefinition.Name})");

Console.WriteLine("Starting workflow...");
var dispatchRequest = new global::Elsa.Workflows.Runtime.Requests.DispatchWorkflowDefinitionRequest
{
DefinitionVersionId = workflowDefinition.Id ?? throw new InvalidOperationException("Workflow definition ID is null")
};

var dispatchResult = await workflowDispatcher.DispatchAsync(dispatchRequest, new global::Elsa.Workflows.Runtime.DispatchWorkflowOptions(), CancellationToken.None);

Console.WriteLine($"Workflow dispatched successfully.");
Console.WriteLine("The workflow will wait for your input...");
Console.WriteLine("(The WorkflowResumeService will handle resuming when you enter values)");
Console.WriteLine();

// Run the host
await host.RunAsync();
}
}

Start by tracing Task1Activity.ExecuteAsync and Task2Activity.ExecuteAsync, especially bookmark creation and input handling during resume. Then inspect WorkflowResumeService.ExecuteAsync and its use of IWorkflowInstanceStore and IWorkflowRuntime. Done means invalid input suspends again, valid values advance through both tasks, and Task3Activity outputs their product.

Written by the indexing model from the issue text.