feat: scheduling strategy system

Author: Spelkington

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

@@ -1,4 +1,5 @@
 using Flowthru.Core.Graph;
+using Flowthru.Core.Graph.Scheduling;
 using Flowthru.Core.Results;
 
 namespace Flowthru.Core.Flows;
@@ -69,6 +70,23 @@ public class ExecutionOptions
   /// </remarks>
   public FlowSliceStrategy? SliceStrategy { get; set; }
 
+  /// <summary>
+  /// Priority strategy used to order ready steps on each dispatch cycle.
+  /// </summary>
+  /// <remarks>
+  /// <para>
+  /// When <c>null</c> (default), the executor selects a strategy automatically:
+  /// <see cref="Graph.Scheduling.FifoSchedulingStrategy"/> for sequential execution
+  /// (<see cref="MaxDegreeOfParallelism"/> = 1), and
+  /// <see cref="Graph.Scheduling.CriticalPathSchedulingStrategy"/> for parallel execution.
+  /// </para>
+  /// <para>
+  /// Provide an explicit value to override this default — for example, to force FIFO
+  /// ordering even under parallelism, or to supply a custom strategy.
+  /// </para>
+  /// </remarks>
+  public ISchedulingStrategy? SchedulingStrategy { get; set; }
+
   /// <summary>
   /// Gets the configured formatter or creates a default one.
   /// </summary>

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

@@ -5,6 +5,7 @@
 using Flowthru.Core.Graph;
 using Flowthru.Core.Graph.Meta;
 using Flowthru.Core.Graph.Meta.Models;
+using Flowthru.Core.Graph.Scheduling;
 using Flowthru.Core.Graph.Validation;
 using Microsoft.Extensions.Logging;
 
@@ -266,7 +267,11 @@ public void Build(FlowSliceStrategy? sliceStrategy = null)
     var stepsToExecute = _slicedSteps ?? _steps;
     DependencyAnalyzer.AssignLayers(stepsToExecute);
 
-    // Step 4: Group steps by layer for execution
+    // Step 4: Compute heights for critical-path scheduling
+    // Height = longest path to a sink; used by CriticalPathSchedulingStrategy.
+    DependencyAnalyzer.ComputeHeights(stepsToExecute);
+
+    // Step 5: Group steps by layer for execution
     ExecutionLayers = DependencyAnalyzer.GroupByLayer(stepsToExecute).ToList();
 
     Logger?.LogInformation(
@@ -642,6 +647,10 @@ CancellationToken cancellationToken
         stepList,
         parallelism,
         ExecuteStepWithTrackingAsync,
+        options.SchedulingStrategy
+          ?? (
+            parallelism == 1 ? new FifoSchedulingStrategy() : new CriticalPathSchedulingStrategy()
+          ),
         Logger
       );
 

src/core/Flowthru.Core/Graph/DependencyAnalyzer.cs

@@ -516,6 +516,51 @@ List<FlowStep> allSteps
     return result;
   }
 
+  /// <summary>
+  /// Computes the height of every node in the DAG: the length of the longest path
+  /// from that node to any sink (a node with no dependents).
+  /// </summary>
+  /// <param name="nodes">Steps whose heights should be computed (must already have
+  /// dependencies resolved via <see cref="BuildDependencyGraph"/>).</param>
+  /// <remarks>
+  /// <para>
+  /// Height is defined recursively:
+  /// <code>
+  ///   height(sink) = 0
+  ///   height(n)    = 1 + max(height(d) for d in dependents(n))
+  /// </code>
+  /// This is computed iteratively in reverse-topological order (sinks first) in O(V+E).
+  /// </para>
+  /// <para>
+  /// Used by <see cref="Scheduling.CriticalPathSchedulingStrategy"/> to prioritise ready
+  /// steps that gate the most downstream work.
+  /// </para>
+  /// </remarks>
+  public static void ComputeHeights(List<FlowStep> nodes)
+  {
+    // Build reverse adjacency: step → steps that depend on it.
+    var dependents = nodes.ToDictionary(n => n, _ => new List<FlowStep>());
+    foreach (var node in nodes)
+    {
+      foreach (var dep in node.Dependencies)
+      {
+        if (dependents.TryGetValue(dep, out var list))
+        {
+          list.Add(node);
+        }
+      }
+    }
+
+    // Process nodes in reverse topological order (sinks first).
+    // Re-derive topological order from Layer assignments: highest Layer = processed first.
+    var ordered = nodes.OrderByDescending(n => n.Layer);
+    foreach (var node in ordered)
+    {
+      var deps = dependents[node];
+      node.Height = deps.Count == 0 ? 0 : 1 + deps.Max(d => d.Height);
+    }
+  }
+
   /// <summary>
   /// Builds a reverse dependency map (node → nodes that depend on it).
   /// </summary>

src/core/Flowthru.Core/Graph/FlowStep.cs

@@ -100,6 +100,17 @@ public class FlowStep
   /// </summary>
   public int Layer { get; set; } = -1; // -1 indicates not yet assigned
 
+  /// <summary>
+  /// Height in the DAG: the length of the longest path from this step to any sink (leaf).
+  /// Sinks have height 0. Used by critical-path scheduling to prioritise steps that unblock
+  /// the most downstream work.
+  /// </summary>
+  /// <remarks>
+  /// Populated by <see cref="DependencyAnalyzer.ComputeHeights"/> after the dependency
+  /// graph has been built. A value of -1 indicates heights have not yet been computed.
+  /// </remarks>
+  public int Height { get; set; } = -1; // -1 indicates not yet computed
+
   /// <summary>
   /// Creates a new Flow step with a transformation function.
   /// </summary>

src/core/Flowthru.Core/Graph/Scheduling/CriticalPathSchedulingStrategy.cs

@@ -0,0 +1,41 @@
+namespace Flowthru.Core.Graph.Scheduling;
+
+/// <summary>
+/// Scheduling strategy that prioritises steps with the longest remaining critical path
+/// (Highest Level First / HLF).
+/// </summary>
+/// <remarks>
+/// <para>
+/// When multiple steps are ready simultaneously, this strategy dispatches the step
+/// with the greatest <see cref="FlowStep.Height"/> first — where height is the length
+/// of the longest path from that step to any leaf in the DAG.
+/// </para>
+/// <para>
+/// <strong>Rationale.</strong> Starting a high-height step unblocks more downstream
+/// parallelism sooner, keeping worker threads saturated. Graham (1966) proved that any
+/// list-scheduling algorithm using this priority order achieves a makespan within a
+/// factor of <c>2 − 1/m</c> of optimal on <c>m</c> identical machines — the best
+/// polynomial-time guarantee known for <c>P|prec|C_max</c>.
+/// </para>
+/// <para>
+/// <strong>Prerequisites.</strong> <see cref="FlowStep.Height"/> values must have been
+/// populated by <see cref="DependencyAnalyzer.ComputeHeights"/> before this strategy
+/// is used. Steps with <c>Height == -1</c> are treated as height 0.
+/// </para>
+/// <para>
+/// Steps with equal height retain their relative arrival order (stable sort), so the
+/// strategy degrades gracefully to FIFO when all ready steps share the same height.
+/// </para>
+/// </remarks>
+public sealed class CriticalPathSchedulingStrategy : ISchedulingStrategy
+{
+  /// <inheritdoc/>
+  public IReadOnlyList<FlowStep> Prioritize(
+    IReadOnlyList<FlowStep> readySteps,
+    SchedulingContext context
+  )
+  {
+    // OrderByDescending is a stable sort — equal-height steps keep arrival order.
+    return readySteps.OrderByDescending(s => s.Height >= 0 ? s.Height : 0).ToList();
+  }
+}

src/core/Flowthru.Core/Graph/Scheduling/FifoSchedulingStrategy.cs

@@ -0,0 +1,18 @@
+namespace Flowthru.Core.Graph.Scheduling;
+
+/// <summary>
+/// Scheduling strategy that preserves arrival order (first-in, first-out).
+/// </summary>
+/// <remarks>
+/// Equivalent to the behaviour of the original <c>ConcurrentQueue</c>-based
+/// dispatcher: steps become eligible for dispatch in the order their last
+/// dependency completes, and that order is preserved when claiming worker slots.
+/// </remarks>
+public sealed class FifoSchedulingStrategy : ISchedulingStrategy
+{
+  /// <inheritdoc/>
+  public IReadOnlyList<FlowStep> Prioritize(
+    IReadOnlyList<FlowStep> readySteps,
+    SchedulingContext context
+  ) => readySteps;
+}

src/core/Flowthru.Core/Graph/Scheduling/ISchedulingStrategy.cs

@@ -0,0 +1,35 @@
+namespace Flowthru.Core.Graph.Scheduling;
+
+/// <summary>
+/// Defines the priority ordering for ready steps in the task-graph scheduler.
+/// </summary>
+/// <remarks>
+/// <para>
+/// When multiple steps are ready to dispatch simultaneously (all dependencies satisfied),
+/// the scheduler delegates to an <see cref="ISchedulingStrategy"/> to determine which
+/// step should be dispatched first. This affects which steps claim a worker slot when
+/// the degree of parallelism is limited.
+/// </para>
+/// <para>
+/// Implementations receive the currently ready steps and a <see cref="SchedulingContext"/>
+/// containing graph structure and any available historical data, then return the steps in
+/// dispatch-priority order (highest priority first).
+/// </para>
+/// <para>
+/// The strategy is invoked each time the dispatch loop drains the ready queue, ensuring
+/// newly-unblocked steps are ranked relative to any that were already waiting.
+/// </para>
+/// </remarks>
+public interface ISchedulingStrategy
+{
+  /// <summary>
+  /// Returns <paramref name="readySteps"/> sorted in dispatch-priority order,
+  /// highest priority first.
+  /// </summary>
+  /// <param name="readySteps">Steps whose dependencies have all completed and that
+  /// are eligible for immediate dispatch.</param>
+  /// <param name="context">Read-only graph context available to inform ordering decisions.</param>
+  /// <returns>The same steps in priority order. Must contain exactly the same elements
+  /// as <paramref name="readySteps"/>.</returns>
+  IReadOnlyList<FlowStep> Prioritize(IReadOnlyList<FlowStep> readySteps, SchedulingContext context);
+}

src/core/Flowthru.Core/Graph/Scheduling/SchedulingContext.cs

@@ -0,0 +1,18 @@
+namespace Flowthru.Core.Graph.Scheduling;
+
+/// <summary>
+/// Read-only graph context passed to <see cref="ISchedulingStrategy.Prioritize"/> on
+/// each dispatch cycle.
+/// </summary>
+/// <remarks>
+/// Carries structural information about the DAG that strategies may use to make ordering
+/// decisions. Designed to be extended: future fields (e.g., historical step durations)
+/// can be added here without changing the <see cref="ISchedulingStrategy"/> signature.
+/// </remarks>
+/// <param name="Dependents">
+/// Reverse adjacency map: for each step, the list of steps that depend on it.
+/// A step with an empty list is a sink (no descendants).
+/// </param>
+public sealed record SchedulingContext(
+  IReadOnlyDictionary<FlowStep, IReadOnlyList<FlowStep>> Dependents
+);

src/core/Flowthru.Core/Graph/TaskGraphExecutor.cs

@@ -1,5 +1,6 @@
 using System.Collections.Concurrent;
 using Flowthru.Core.Flows;
+using Flowthru.Core.Graph.Scheduling;
 using Microsoft.Extensions.Logging;
 
 namespace Flowthru.Core.Graph;
@@ -29,6 +30,7 @@ internal sealed class TaskGraphExecutor
 {
   private readonly IReadOnlyList<FlowStep> _steps;
   private readonly int _maxDegreeOfParallelism;
+  private readonly ISchedulingStrategy _strategy;
   private readonly ILogger? _logger;
   private readonly Func<FlowStep, CancellationToken, Task<StepResult>> _executeStep;
 
@@ -42,11 +44,13 @@ internal sealed class TaskGraphExecutor
   /// Pass <see cref="int.MaxValue"/> for unbounded parallelism.
   /// </param>
   /// <param name="executeStep">Per-step execution delegate (matches <c>ExecuteStepWithTrackingAsync</c>).</param>
+  /// <param name="strategy">Priority strategy used to order ready steps on each dispatch cycle.</param>
   /// <param name="logger">Optional logger.</param>
   internal TaskGraphExecutor(
     IReadOnlyList<FlowStep> steps,
     int maxDegreeOfParallelism,
     Func<FlowStep, CancellationToken, Task<StepResult>> executeStep,
+    ISchedulingStrategy strategy,
     ILogger? logger = null
   )
   {
@@ -61,6 +65,7 @@ internal TaskGraphExecutor(
     _steps = steps;
     _maxDegreeOfParallelism = maxDegreeOfParallelism == -1 ? int.MaxValue : maxDegreeOfParallelism;
     _executeStep = executeStep;
+    _strategy = strategy;
     _logger = logger;
   }
 
@@ -96,7 +101,10 @@ CancellationToken cancellationToken
     );
 
     // Reverse adjacency: for each step, which steps depend on it?
-    var dependents = _steps.ToDictionary(s => s, _ => new List<FlowStep>());
+    var dependents = _steps.ToDictionary(
+      s => s,
+      s => (IReadOnlyList<FlowStep>)new List<FlowStep>()
+    );
     foreach (var step in _steps)
     {
       foreach (var dep in step.Dependencies)
@@ -105,14 +113,17 @@ CancellationToken cancellationToken
         // steps that are in _steps (sliced or full set).
         if (dependents.TryGetValue(dep, out var list))
         {
-          list.Add(step);
+          ((List<FlowStep>)list).Add(step);
         }
       }
     }
 
-    // Steps whose upstream is fully satisfied (or has no dependencies).
-    // Channel is unbounded write / bounded dispatch (controlled by the semaphore).
-    var readyQueue = new ConcurrentQueue<FlowStep>(_steps.Where(s => s.Dependencies.Count == 0));
+    var schedulingContext = new SchedulingContext(dependents);
+
+    // Newly-ready steps are added here by concurrent task completions; a ConcurrentBag
+    // is safe for multi-producer, single-consumer access. On each dispatch cycle the
+    // main loop drains it into a List and passes it to the strategy for ordering.
+    var readyBag = new ConcurrentBag<FlowStep>(_steps.Where(s => s.Dependencies.Count == 0));
 
     // Tracks which steps were skipped because an upstream dependency failed.
     var skipped = new HashSet<FlowStep>();
@@ -139,7 +150,16 @@ CancellationToken cancellationToken
     while (results.Count + skipped.Count < totalSteps)
     {
       // Drain all currently runnable steps into in-flight tasks.
-      while (readyQueue.TryDequeue(out var step))
+      // Collect from the concurrent bag into a snapshot, then ask the strategy to order them.
+      var readySnapshot = new List<FlowStep>();
+      while (readyBag.TryTake(out var taken))
+      {
+        readySnapshot.Add(taken);
+      }
+
+      var prioritised = _strategy.Prioritize(readySnapshot, schedulingContext);
+
+      foreach (var step in prioritised)
       {
         if (skipped.Contains(step))
         {
@@ -208,7 +228,7 @@ CancellationToken cancellationToken
                 );
 
                 // Notify dependents — decrement their pending count.
-                EnqueueReadyDependents(capturedStep, pendingDeps, dependents, skipped, readyQueue);
+                EnqueueReadyDependents(capturedStep, pendingDeps, dependents, skipped, readyBag);
               }
             }
             finally
@@ -223,7 +243,7 @@ CancellationToken cancellationToken
         inFlight.Add(task);
       }
 
-      if (inFlight.Count == 0)
+      if (inFlight.Count == 0 && readyBag.IsEmpty)
       {
         // Nothing dispatched and nothing running. If not all steps accounted for,
         // the dependency graph has a cycle that AssignLayers should have caught.
@@ -290,9 +310,9 @@ CancellationToken cancellationToken
   private static void EnqueueReadyDependents(
     FlowStep completedStep,
     ConcurrentDictionary<FlowStep, int> pendingDeps,
-    Dictionary<FlowStep, List<FlowStep>> dependents,
+    IReadOnlyDictionary<FlowStep, IReadOnlyList<FlowStep>> dependents,
     HashSet<FlowStep> skipped,
-    ConcurrentQueue<FlowStep> readyQueue
+    ConcurrentBag<FlowStep> readyBag
   )
   {
     foreach (var dependent in dependents[completedStep])
@@ -305,15 +325,15 @@ ConcurrentQueue<FlowStep> readyQueue
       var remaining = pendingDeps.AddOrUpdate(dependent, 0, (_, current) => current - 1);
       if (remaining == 0)
       {
-        readyQueue.Enqueue(dependent);
+        readyBag.Add(dependent);
       }
     }
   }
 
   private static void SkipDownstream(
     FlowStep failedStep,
     HashSet<FlowStep> skipped,
-    Dictionary<FlowStep, List<FlowStep>> dependents
+    IReadOnlyDictionary<FlowStep, IReadOnlyList<FlowStep>> dependents
   )
   {
     // BFS over the dependents graph.