feat: multithreading implementation

Author: Spelkington

CONTRIBUTING.md

@@ -104,9 +104,9 @@ This doesn't mean *ignoring* these concerns — it just means extending the API
 The project uses NX for task orchestration. Common commands:
 
 ```bash
-nx run flowthru:build                   # Build the solution
-nx run flowthru:test                    # Run all tests with coverage
-nx run flowthru:format:csharp           # Format code with CSharpier
+dotnet build # Confirm solution builds fully
+nx run affected -t test # Run all test projects affected by current changes
+dotnet test # Run all tests for the project.
 ```
 
 To run a subset of tests by category:

src/core/Flowthru.Core/Flows/ExecutionOptions.cs

@@ -1,5 +1,5 @@
-using Flowthru.Core.Results;
 using Flowthru.Core.Graph;
+using Flowthru.Core.Results;
 
 namespace Flowthru.Core.Flows;
 
@@ -34,13 +34,23 @@ public class ExecutionOptions
   public bool StopOnFirstError { get; set; } = true;
 
   /// <summary>
-  /// Whether to enable parallel execution of nodes within the same layer.
+  /// Maximum number of steps that may execute concurrently.
   /// </summary>
   /// <remarks>
-  /// Phase 2 feature - currently not implemented.
-  /// When true, nodes in the same execution layer run concurrently.
+  /// <para>
+  /// Controls the degree of parallelism in the task-graph scheduler. Steps whose
+  /// dependencies are all satisfied are dispatched immediately, up to this limit.
+  /// </para>
+  /// <para>
+  /// <list type="bullet">
+  /// <item><c>1</c> (default) — sequential execution; one step at a time in topological order.</item>
+  /// <item><c>N &gt; 1</c> — up to N independent steps run concurrently.</item>
+  /// <item><c>-1</c> or <see cref="int.MaxValue"/> — unbounded parallelism; all ready steps
+  /// are dispatched immediately.</item>
+  /// </list>
+  /// </para>
   /// </remarks>
-  public bool EnableParallelExecution { get; set; } = false;
+  public int MaxDegreeOfParallelism { get; set; } = 1;
 
   /// <summary>
   /// The result formatter to use for displaying execution results.

src/core/Flowthru.Core/Flows/Flow.cs

@@ -553,20 +553,40 @@ public DagMetadata ExportDag()
   }
 
   /// <summary>
-  /// /// Builds and executes the flow, returning comprehensive execution results.
+  /// Builds and executes the flow, returning comprehensive execution results.
   /// </summary>
   /// <param name="cancellationToken">Cancellation token to signal graceful shutdown</param>
-  /// <returns>FlowResult containing execution status, timing, and Flow results</returns>
+  /// <returns>FlowResult containing execution status, timing, and step results</returns>
   /// <remarks>
-  /// <para>
   /// This is the primary high-level API for executing flows. It automatically
-  /// calls Build() if the Flow hasn't been built yet, then executes and tracks results.
+  /// calls Build() if the Flow hasn't been built yet, then executes via the
+  /// task-graph scheduler with default options (sequential, stop on first error).
+  /// </remarks>
+  public Task<FlowResult> RunAsync(CancellationToken cancellationToken) =>
+    RunAsync(new ExecutionOptions(), cancellationToken);
+
+  /// <summary>
+  /// Builds and executes the flow with the supplied execution options.
+  /// </summary>
+  /// <param name="options">Controls parallelism, error policy, and other execution behaviour.</param>
+  /// <param name="cancellationToken">Cancellation token to signal graceful shutdown</param>
+  /// <returns>FlowResult containing execution status, timing, and step results</returns>
+  /// <remarks>
+  /// <para>
+  /// Steps are dispatched by the task-graph scheduler as soon as all their dependencies
+  /// complete, up to <see cref="ExecutionOptions.MaxDegreeOfParallelism"/> concurrent steps.
+  /// </para>
+  /// <para>
+  /// With <c>MaxDegreeOfParallelism = 1</c> (default) execution is sequential and
+  /// behaviourally equivalent to the previous layer-by-layer loop.
   /// </para>
   /// </remarks>
-  public async Task<FlowResult> RunAsync(CancellationToken cancellationToken)
+  public async Task<FlowResult> RunAsync(
+    ExecutionOptions options,
+    CancellationToken cancellationToken
+  )
   {
     var stopwatch = Stopwatch.StartNew();
-    var stepResults = new Dictionary<string, StepResult>();
 
     try
     {
@@ -577,33 +597,34 @@ public async Task<FlowResult> RunAsync(CancellationToken cancellationToken)
         Build();
       }
 
-      Logger?.LogInformation("Starting flow execution via RunAsync()");
+      var stepList = (IReadOnlyList<FlowStep>)(_slicedSteps ?? _steps);
 
-      // Execute all layers
-      foreach (var layer in ExecutionLayers!)
-      {
-        Logger?.LogInformation("Executing layer with {StepCount} steps", layer.Count);
+      Logger?.LogInformation(
+        "Starting flow execution via RunAsync() ({StepCount} steps, parallelism={Parallelism})",
+        stepList.Count,
+        options.MaxDegreeOfParallelism
+      );
 
-        foreach (var flowStep in layer)
-        {
-          // Check for cancellation before starting each step
-          cancellationToken.ThrowIfCancellationRequested();
+      var executor = new Graph.TaskGraphExecutor(
+        stepList,
+        options.MaxDegreeOfParallelism,
+        ExecuteStepWithTrackingAsync,
+        Logger
+      );
 
-          var stepResult = await ExecuteStepWithTrackingAsync(flowStep, cancellationToken);
-          stepResults[flowStep.Label] = stepResult;
+      var stepResults = await executor.RunAsync(options.StopOnFirstError, cancellationToken);
 
-          // If step failed, stop execution
-          if (!stepResult.Success)
-          {
-            stopwatch.Stop();
-            return FlowResult.CreateFailure(
-              stopwatch.Elapsed,
-              stepResult.Exception!,
-              stepResults,
-              Name
-            );
-          }
-        }
+      // Surface the first failure when StopOnFirstError is true.
+      var firstFailure = stepResults.Values.FirstOrDefault(r => !r.Success);
+      if (firstFailure != null)
+      {
+        stopwatch.Stop();
+        return FlowResult.CreateFailure(
+          stopwatch.Elapsed,
+          firstFailure.Exception!,
+          stepResults,
+          Name
+        );
       }
 
       stopwatch.Stop();
@@ -616,40 +637,31 @@ public async Task<FlowResult> RunAsync(CancellationToken cancellationToken)
     }
     catch (OperationCanceledException)
     {
-      // Re-throw cancellation exceptions so they propagate to the caller
-      // Cancellation is not a failure but a requested abort
+      // Re-throw — cancellation is not a failure but a requested abort.
       stopwatch.Stop();
       throw;
     }
     catch (Exception ex)
     {
       stopwatch.Stop();
       Logger?.LogError(ex, "Flow execution failed: {ErrorMessage}", ex.Message);
-      return FlowResult.CreateFailure(stopwatch.Elapsed, ex, stepResults, Name);
+      return FlowResult.CreateFailure(
+        stopwatch.Elapsed,
+        ex,
+        new Dictionary<string, StepResult>(),
+        Name
+      );
     }
   }
 
   /// <summary>
-  /// Executes the Flow sequentially, layer by layer.
+  /// Executes the flow in topological order, throwing on the first step failure.
   /// </summary>
   /// <param name="cancellationToken">Cancellation token to signal graceful shutdown</param>
-  /// <returns>Task representing the Flow execution</returns>
-  /// <exception cref="InvalidOperationException">Thrown if Flow has not been built</exception>
+  /// <returns>Task representing the flow execution</returns>
+  /// <exception cref="InvalidOperationException">Thrown if the flow has not been built</exception>
   /// <remarks>
-  /// <para>
-  /// This method executes Flow in topological order:
-  /// 1. Execute all Flow in layer 0 sequentially
-  /// 2. Execute all Flow in layer 1 sequentially
-  /// 3. Continue until all layers are complete
-  /// </para>
-  /// <para>
-  /// <strong>Note:</strong> This method throws exceptions on failure. For result-based
-  /// execution with error handling, use RunAsync() instead.
-  /// </para>
-  /// <para>
-  /// In Phase 2, this will be replaced with a parallel executor that can run
-  /// steps within the same layer concurrently.
-  /// </para>
+  /// For structured result-based execution (including parallel), use <see cref="RunAsync(CancellationToken)"/>.
   /// </remarks>
   public async Task ExecuteAsync(CancellationToken cancellationToken)
   {
@@ -664,22 +676,17 @@ public async Task ExecuteAsync(CancellationToken cancellationToken)
 
     try
     {
-      foreach (var layer in ExecutionLayers!)
-      {
-        Logger?.LogInformation("Executing layer with {StepCount} steps", layer.Count);
+      var result = await RunAsync(cancellationToken);
 
-        foreach (var flowStep in layer)
-        {
-          // Check for cancellation before starting each step
-          cancellationToken.ThrowIfCancellationRequested();
-
-          await ExecuteStepAsync(flowStep, cancellationToken);
-        }
+      if (!result.Success)
+      {
+        throw result.Exception
+          ?? new InvalidOperationException("Flow execution failed with no exception details.");
       }
 
       Logger?.LogInformation("Flow execution completed successfully");
     }
-    catch (Exception ex)
+    catch (Exception ex) when (ex is not OperationCanceledException)
     {
       Logger?.LogError(ex, "Flow execution failed: {ErrorMessage}", ex.Message);
       throw;