.NET: Add parallel declarative Foreach execution

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowActionVisitor.cs

@@ -182,7 +182,13 @@ protected override void Visit(Foreach item)
         this.Trace(item);
 
         // Entry point for loop
-        ForeachExecutor action = new(item, this._workflowState);
+        ForeachExecutor action = new(item, this._workflowState, this._workflowOptions);
+        if (action.IsParallel)
+        {
+            this.ContinueWith(action);
+            return;
+        }
+
         string loopId = ForeachExecutor.Steps.Next(action.Id);
         this.ContinueWith(action, condition: null, CompletionHandler);
         // Transition to select the next item

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowElementWalker.cs

@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
+using Microsoft.Agents.AI.Workflows.Declarative.ObjectModel;
 using Microsoft.Agents.ObjectModel;
 
 namespace Microsoft.Agents.AI.Workflows.Declarative.Interpreter;
@@ -18,6 +19,11 @@ public override bool DefaultVisit(BotElement definition)
         if (definition is DialogAction action)
         {
             action.Accept(this._visitor);
+
+            if (action is Foreach foreachAction && ForeachExecutionOptions.Parse(foreachAction).IsParallel)
+            {
+                return false;
+            }
         }
 
         return true;

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs

@@ -0,0 +1,124 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Globalization;
+using Microsoft.Agents.AI.Workflows.Declarative.Extensions;
+using Microsoft.Agents.ObjectModel;
+
+namespace Microsoft.Agents.AI.Workflows.Declarative.ObjectModel;
+
+internal enum ForeachExecutionMode
+{
+    Sequential,
+    Parallel,
+}
+
+/// <summary>
+/// Strongly typed adapter for Foreach execution fields preserved by the external ObjectModel package.
+/// </summary>
+/// <remarks>
+/// Once <see cref="Foreach"/> exposes generated properties for these fields, only this adapter needs to change.
+/// </remarks>
+internal sealed record ForeachExecutionOptions(
+    ForeachExecutionMode Mode,
+    int MaxParallelism,
+    TimeSpan? IterationTimeout)
+{
+    internal const string ModePropertyName = "mode";
+    internal const string MaxParallelismPropertyName = "maxParallelism";
+    internal const string TimeoutPropertyName = "timeoutInMilliseconds";
+
+    private const int DefaultMaxParallelism = 4;
+
+    public bool IsParallel => this.Mode == ForeachExecutionMode.Parallel;
+
+    public static ForeachExecutionOptions Parse(Foreach model)
+    {
+        DataValue? modeValue = GetExtensionValue(model, ModePropertyName);
+        DataValue? maxParallelismValue = GetExtensionValue(model, MaxParallelismPropertyName);
+        DataValue? timeoutValue = GetExtensionValue(model, TimeoutPropertyName);
+
+        ForeachExecutionMode mode = ParseMode(model, modeValue);
+        int maxParallelism = ParseInteger(model, MaxParallelismPropertyName, maxParallelismValue) ?? DefaultMaxParallelism;
+        int? timeoutMilliseconds = ParseInteger(model, TimeoutPropertyName, timeoutValue);
+
+        if (mode == ForeachExecutionMode.Sequential && (maxParallelismValue is not null || timeoutValue is not null))
+        {
+            throw InvalidConfiguration(model, $"'{MaxParallelismPropertyName}' and '{TimeoutPropertyName}' require '{ModePropertyName}: Parallel'.");
+        }
+
+        if (maxParallelism <= 0)
+        {
+            throw InvalidConfiguration(model, $"'{MaxParallelismPropertyName}' must be greater than zero.");
+        }
+
+        if (timeoutMilliseconds <= 0)
+        {
+            throw InvalidConfiguration(model, $"'{TimeoutPropertyName}' must be greater than zero when specified.");
+        }
+
+        return new(mode, maxParallelism, timeoutMilliseconds.HasValue ? TimeSpan.FromMilliseconds(timeoutMilliseconds.Value) : null);
+    }
+
+    private static ForeachExecutionMode ParseMode(Foreach model, DataValue? value)
+    {
+        if (value is null)
+        {
+            return ForeachExecutionMode.Sequential;
+        }
+
+        if (value is not StringDataValue stringValue)
+        {
+            throw InvalidConfiguration(model, $"'{ModePropertyName}' must be 'Sequential' or 'Parallel'.");
+        }
+
+        if (string.Equals(stringValue.Value, nameof(ForeachExecutionMode.Sequential), StringComparison.OrdinalIgnoreCase))
+        {
+            return ForeachExecutionMode.Sequential;
+        }
+
+        if (string.Equals(stringValue.Value, nameof(ForeachExecutionMode.Parallel), StringComparison.OrdinalIgnoreCase))
+        {
+            return ForeachExecutionMode.Parallel;
+        }
+
+        throw InvalidConfiguration(model, $"'{ModePropertyName}' must be 'Sequential' or 'Parallel'.");
+    }
+
+    private static int? ParseInteger(Foreach model, string propertyName, DataValue? value)
+    {
+        if (value is null)
+        {
+            return null;
+        }
+
+        object? rawValue = value.ToFormula().ToObject();
+        if (rawValue is not (byte or sbyte or short or ushort or int or uint or long or ulong or float or double or decimal))
+        {
+            throw InvalidConfiguration(model, $"'{propertyName}' must be an integer.");
+        }
+
+        decimal decimalValue;
+        try
+        {
+            decimalValue = Convert.ToDecimal(rawValue, CultureInfo.InvariantCulture);
+        }
+        catch (Exception exception) when (exception is FormatException or InvalidCastException or OverflowException)
+        {
+            throw InvalidConfiguration(model, $"'{propertyName}' must be an integer.", exception);
+        }
+
+        if (decimalValue != decimal.Truncate(decimalValue) || decimalValue < int.MinValue || decimalValue > int.MaxValue)
+        {
+            throw InvalidConfiguration(model, $"'{propertyName}' must be an integer.");
+        }
+
+        return decimal.ToInt32(decimalValue);
+    }
+
+    private static DataValue? GetExtensionValue(Foreach model, string propertyName) =>
+        model.ExtensionData?.Properties.TryGetValue(propertyName, out DataValue? value) is true ? value : null;
+
+    private static DeclarativeModelException InvalidConfiguration(Foreach model, string message, Exception? innerException = null) =>
+        new($"Invalid Foreach configuration for '{model.Id.Value}': {message}", innerException);
+}

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs

@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
+using System;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -29,36 +30,177 @@ public static class Steps
 
     private int _index;
     private FormulaValue[] _values;
+    private readonly ForeachExecutionOptions _executionOptions;
+    private readonly DeclarativeWorkflowOptions? _workflowOptions;
+    private readonly WorkflowFormulaState _workflowState;
 
-    public ForeachExecutor(Foreach model, WorkflowFormulaState state)
+    public ForeachExecutor(Foreach model, WorkflowFormulaState state, DeclarativeWorkflowOptions? workflowOptions = null)
         : base(model, state)
     {
         this._values = [];
+        this._executionOptions = ForeachExecutionOptions.Parse(model);
+        this._workflowOptions = workflowOptions;
+        this._workflowState = state;
+
+        if (this._executionOptions.IsParallel)
+        {
+            if (workflowOptions is null)
+            {
+                throw new DeclarativeModelException($"Parallel Foreach '{model.Id.Value}' requires workflow execution options.");
+            }
+
+            ParallelForeachIterationRunner.ValidateBody(model);
+        }
     }
 
     public bool HasValue { get; private set; }
 
-    protected override bool IsDiscreteAction => false;
+    public bool IsParallel => this._executionOptions.IsParallel;
+
+    protected override bool IsDiscreteAction => this.IsParallel;
 
     protected override async ValueTask<object?> ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
+    {
+        return this.IsParallel
+            ? await this.ExecuteParallelAsync(context, cancellationToken).ConfigureAwait(false)
+            : await this.ExecuteSequentialAsync(context, cancellationToken).ConfigureAwait(false);
+    }
+
+    private async ValueTask<object?> ExecuteSequentialAsync(IWorkflowContext context, CancellationToken cancellationToken)
     {
         Throw.IfNull(this.Model.Items, $"{nameof(this.Model)}.{nameof(this.Model.Items)}");
 
         this._index = 0;
+        this._values = this.GetValues();
 
-        EvaluationResult<DataValue> expressionResult = this.Evaluator.GetValue(this.Model.Items);
-        if (expressionResult.Value is TableDataValue tableValue)
+        await this.ResetStateAsync(context, cancellationToken).ConfigureAwait(false);
+
+        return default;
+    }
+
+    private async ValueTask<object?> ExecuteParallelAsync(IWorkflowContext context, CancellationToken cancellationToken)
+    {
+        Throw.IfNull(this.Model.Items, $"{nameof(this.Model)}.{nameof(this.Model.Items)}");
+        DeclarativeWorkflowOptions workflowOptions = Throw.IfNull(this._workflowOptions);
+        FormulaValue[] values = this.GetValues();
+
+        await this.ResetStateAsync(context, cancellationToken).ConfigureAwait(false);
+
+        try
         {
-            this._values = [.. tableValue.Values.Select(ToLoopValue)];
+            if (values.Length == 0)
+            {
+                return default;
+            }
+
+            WorkflowStateSnapshot stateSnapshot = this._workflowState.CaptureSnapshot();
+            ParallelForeachIterationResult?[] iterationResults = new ParallelForeachIterationResult?[values.Length];
+            Exception?[] iterationFailures = new Exception?[values.Length];
+            int nextIndex = -1;
+
+            using CancellationTokenSource groupCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            int workerCount = Math.Min(values.Length, this._executionOptions.MaxParallelism);
+            Task[] workers =
+            [
+                .. Enumerable.Range(0, workerCount).Select(_ => RunWorkerAsync()),
+            ];
+
+            await Task.WhenAll(workers).ConfigureAwait(false);
+            cancellationToken.ThrowIfCancellationRequested();
+
+            Exception[] failures =
+            [
+                .. iterationFailures
+                    .Select(
+                        (exception, index) =>
+                            exception is null
+                                ? null
+                                : new DeclarativeActionException(
+                                    $"Parallel Foreach '{this.Id}' iteration {index} failed.",
+                                    exception))
+                    .Where(exception => exception is not null)
+                    .Cast<Exception>(),
+            ];
+            if (failures.Length > 0)
+            {
+                throw new AggregateException($"Parallel Foreach '{this.Id}' failed.", failures);
+            }
+
+            foreach (ParallelForeachIterationResult iterationResult in iterationResults.Cast<ParallelForeachIterationResult>())
+            {
+                foreach (WorkflowStateChange stateChange in iterationResult.StateChanges)
+                {
+                    await CommitStateChangeAsync(context, stateChange, cancellationToken).ConfigureAwait(false);
+                }
+
+                foreach (WorkflowEvent workflowEvent in iterationResult.Events)
+                {
+                    await context.AddEventAsync(workflowEvent, cancellationToken).ConfigureAwait(false);
+                }
+            }
+
+            return default;
+
+            async Task RunWorkerAsync()
+            {
+                while (!groupCancellation.IsCancellationRequested)
+                {
+                    int iterationIndex = Interlocked.Increment(ref nextIndex);
+                    if (iterationIndex >= values.Length)
+                    {
+                        return;
+                    }
+
+                    try
+                    {
+                        iterationResults[iterationIndex] = await ParallelForeachIterationRunner.RunAsync(
+                            this.Model,
+                            values[iterationIndex],
+                            iterationIndex,
+                            stateSnapshot,
+                            workflowOptions,
+                            this._executionOptions.IterationTimeout,
+                            groupCancellation.Token).ConfigureAwait(false);
+                    }
+                    catch (OperationCanceledException) when (groupCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+                    catch (Exception exception)
+                    {
+                        iterationFailures[iterationIndex] = exception;
+                        groupCancellation.Cancel();
+                        return;
+                    }
+                }
+            }
         }
-        else
+        finally
         {
-            this._values = [expressionResult.Value.ToFormula()];
+            await this.ResetStateAsync(context, CancellationToken.None).ConfigureAwait(false);
         }
+    }
 
-        await this.ResetStateAsync(context, cancellationToken).ConfigureAwait(false);
+    private static ValueTask CommitStateChangeAsync(
+        IWorkflowContext context,
+        WorkflowStateChange stateChange,
+        CancellationToken cancellationToken)
+    {
+        FormulaValue value = stateChange.Value.ToFormula();
+        return stateChange.ScopeName switch
+        {
+            VariableScopeNames.System => context.QueueSystemUpdateAsync(stateChange.VariableName, value, cancellationToken),
+            VariableScopeNames.Environment => context.QueueEnvironmentUpdateAsync(stateChange.VariableName, value, cancellationToken),
+            _ => context.QueueStateUpdateAsync(stateChange.VariableName, value, stateChange.ScopeName, cancellationToken),
+        };
+    }
 
-        return default;
+    private FormulaValue[] GetValues()
+    {
+        EvaluationResult<DataValue> expressionResult = this.Evaluator.GetValue(Throw.IfNull(this.Model.Items));
+        return expressionResult.Value is TableDataValue tableValue
+            ? [.. tableValue.Values.Select(ToLoopValue)]
+            : [expressionResult.Value.ToFormula()];
     }
 
     public async ValueTask TakeNextAsync(IWorkflowContext context, object? _, CancellationToken cancellationToken)
@@ -117,6 +259,12 @@ value is RecordDataValue record
     /// </remarks>
     protected override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
     {
+        if (this.IsParallel)
+        {
+            await base.OnCheckpointingAsync(context, cancellationToken).ConfigureAwait(false);
+            return;
+        }
+
         PortableValue[] portableValues = [.. this._values.Select(value => new PortableValue(value.AsPortable()))];
 
         await context.QueueStateUpdateAsync(IndexStateKey, this._index, cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -137,6 +285,11 @@ protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext co
     {
         await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false);
 
+        if (this.IsParallel)
+        {
+            return;
+        }
+
         PortableValue[]? savedValues =
             await context.ReadStateAsync<PortableValue[]>(ValuesStateKey, cancellationToken: cancellationToken).ConfigureAwait(false);
         if (savedValues is null)