From a069f8e256f8fcf222b3d871933ac17f78558914 Mon Sep 17 00:00:00 2001 From: KirschQAQ <114209152+KirschBluteX@users.noreply.github.com> Date: Sat, 15 Aug 2026 07:44:59 -0700 Subject: [PATCH 1/4] feat: add parallel declarative foreach execution --- .../Interpreter/WorkflowActionVisitor.cs | 8 +- .../Interpreter/WorkflowElementWalker.cs | 6 + .../ObjectModel/ForeachExecutionOptions.cs | 115 +++ .../ObjectModel/ForeachExecutor.cs | 171 ++++- .../ParallelForeachIterationRunner.cs | 206 ++++++ .../PowerFx/WorkflowFormulaState.cs | 80 ++- .../README.md | 33 + .../ParallelForeachWorkflowTests.cs | 661 ++++++++++++++++++ 8 files changed, 1268 insertions(+), 12 deletions(-) create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs create mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowActionVisitor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowActionVisitor.cs index 1cd1b2bc94..ba57492eda 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowActionVisitor.cs +++ b/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 diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowElementWalker.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowElementWalker.cs index 4be8bb8892..7b9d36b969 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/WorkflowElementWalker.cs +++ b/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; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs new file mode 100644 index 0000000000..8545838dde --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs @@ -0,0 +1,115 @@ +// 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, +} + +/// +/// Strongly typed adapter for Foreach execution fields preserved by the external ObjectModel package. +/// +/// +/// Once exposes generated properties for these fields, only this adapter needs to change. +/// +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 || + !Enum.TryParse(stringValue.Value, ignoreCase: true, out ForeachExecutionMode mode)) + { + throw InvalidConfiguration(model, $"'{ModePropertyName}' must be 'Sequential' or 'Parallel'."); + } + + return mode; + } + + 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 parallel Foreach configuration for '{model.Id.Value}': {message}", innerException); +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs index f154ad7f97..db85cbed6e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs +++ b/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 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 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 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 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(), + ]; + if (failures.Length > 0) + { + throw new AggregateException($"Parallel Foreach '{this.Id}' failed.", failures); + } + + foreach (ParallelForeachIterationResult iterationResult in iterationResults.Cast()) + { + 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 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 /// 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(ValuesStateKey, cancellationToken: cancellationToken).ConfigureAwait(false); if (savedValues is null) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs new file mode 100644 index 0000000000..34b3595f9e --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs @@ -0,0 +1,206 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Declarative.Extensions; +using Microsoft.Agents.AI.Workflows.Declarative.Interpreter; +using Microsoft.Agents.AI.Workflows.Declarative.Kit; +using Microsoft.Agents.AI.Workflows.Declarative.PowerFx; +using Microsoft.Agents.ObjectModel; +using Microsoft.PowerFx.Types; + +namespace Microsoft.Agents.AI.Workflows.Declarative.ObjectModel; + +internal sealed record ParallelForeachIterationResult( + int Index, + WorkflowStateChange[] StateChanges, + WorkflowEvent[] Events); + +/// +/// Runs one Foreach body through the existing workflow runtime with isolated formula state. +/// +internal static class ParallelForeachIterationRunner +{ + public static void ValidateBody(Foreach model) + { + foreach (DialogAction action in model.Descendants().OfType()) + { + if (action is Question or RequestExternalInput) + { + throw new DeclarativeModelException( + $"Parallel Foreach '{model.Id.Value}' cannot safely checkpoint while action " + + $"'{action.Id.Value}' ({action.GetType().Name}) is awaiting external input."); + } + + if (action is BreakLoop or ContinueLoop && TargetsLoop(action, model)) + { + throw new DeclarativeModelException( + $"Parallel Foreach '{model.Id.Value}' does not support {action.GetType().Name} targeting the parallel loop."); + } + } + } + + public static async Task RunAsync( + Foreach model, + FormulaValue value, + int index, + WorkflowStateSnapshot stateSnapshot, + DeclarativeWorkflowOptions workflowOptions, + TimeSpan? timeout, + CancellationToken cancellationToken) + { + using CancellationTokenSource timeoutSource = new(); + using CancellationTokenSource iterationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutSource.Token); + if (timeout.HasValue) + { + timeoutSource.CancelAfter(timeout.Value); + } + + WorkflowFormulaState branchState = WorkflowFormulaState.CreateBranch(workflowOptions.CreateRecalcEngine(), stateSnapshot); + SetLoopVariable(branchState, model.Value!.Path, new PortableValue(value.AsPortable()).ToFormula()); + if (model.Index is not null) + { + SetLoopVariable(branchState, model.Index.Path, FormulaValue.New(index)); + } + branchState.Bind(); + branchState.BeginTrackingChanges(); + + Workflow workflow = BuildIterationWorkflow(model, branchState, workflowOptions); + + StreamingRun? run = null; + bool runCanceled = false; + try + { + run = await InProcessExecution.RunStreamingAsync( + workflow, + new ActionExecutorResult(model.Id.Value), + cancellationToken: iterationSource.Token).ConfigureAwait(false); + + List eventList = []; + await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync(blockOnPendingRequest: false, iterationSource.Token).ConfigureAwait(false)) + { + eventList.Add(workflowEvent); + } + + if (iterationSource.IsCancellationRequested) + { + await run.CancelRunAsync().ConfigureAwait(false); + runCanceled = true; + } + + if (timeoutSource.IsCancellationRequested) + { + throw new TimeoutException( + $"Parallel Foreach '{model.Id.Value}' iteration {index} exceeded its timeout of {timeout.GetValueOrDefault().TotalMilliseconds} ms."); + } + + cancellationToken.ThrowIfCancellationRequested(); + + RunStatus status = await run.GetStatusAsync(iterationSource.Token).ConfigureAwait(false); + WorkflowEvent[] events = [.. eventList]; + + Exception[] failures = + [ + .. events + .OfType() + .Select( + workflowError => + workflowError.Data as Exception ?? + new DeclarativeActionException( + $"Parallel Foreach '{model.Id.Value}' iteration {index} failed without exception data.")), + ]; + if (failures.Length > 0) + { + Exception innerException = failures.Length == 1 ? failures[0] : new AggregateException(failures); + throw new DeclarativeActionException( + $"Parallel Foreach '{model.Id.Value}' iteration {index} failed.", + innerException); + } + + if (status == RunStatus.PendingRequests || events.Any(workflowEvent => workflowEvent is RequestInfoEvent)) + { + throw new DeclarativeActionException( + $"Parallel Foreach '{model.Id.Value}' iteration {index} requested external input. " + + "Checkpointing an in-flight parallel iteration is not supported."); + } + + if (status != RunStatus.Idle) + { + throw new DeclarativeActionException( + $"Parallel Foreach '{model.Id.Value}' iteration {index} ended with unsupported status '{status}'."); + } + + WorkflowStateChange[] stateChanges = + [ + .. branchState + .CaptureChanges() + .Where(change => !Matches(change, model.Value.Path) && (model.Index is null || !Matches(change, model.Index.Path))), + ]; + WorkflowEvent[] bufferedEvents = [.. events.Where(workflowEvent => ShouldReplay(workflowEvent, model.Id.Value))]; + + return new(index, stateChanges, bufferedEvents); + } + catch (OperationCanceledException) when (timeoutSource.IsCancellationRequested) + { + throw new TimeoutException( + $"Parallel Foreach '{model.Id.Value}' iteration {index} exceeded its timeout of {timeout!.Value.TotalMilliseconds} ms."); + } + finally + { + if (run is not null) + { + if (iterationSource.IsCancellationRequested && !runCanceled) + { + await run.CancelRunAsync().ConfigureAwait(false); + } + + await run.DisposeAsync().ConfigureAwait(false); + } + } + } + + private static Workflow BuildIterationWorkflow( + Foreach model, + WorkflowFormulaState state, + DeclarativeWorkflowOptions workflowOptions) + { + DelegateActionExecutor root = new(model.Id.Value, state); + WorkflowActionVisitor visitor = new(root, state, workflowOptions); + WorkflowElementWalker walker = new(visitor); + foreach (DialogAction action in model.Actions) + { + walker.Visit(action); + } + + return visitor.Complete(); + } + + private static void SetLoopVariable(WorkflowFormulaState state, PropertyPath path, FormulaValue value) => + state.Set(path.VariableName!, value, path.NamespaceAlias); + + private static bool Matches(WorkflowStateChange change, PropertyPath path) => + string.Equals(change.VariableName, path.VariableName, StringComparison.Ordinal) && + string.Equals(change.ScopeName, WorkflowFormulaState.GetScopeName(path.NamespaceAlias), StringComparison.Ordinal); + + private static bool ShouldReplay(WorkflowEvent workflowEvent, string rootExecutorId) => + (workflowEvent is not WorkflowStartedEvent + and not SuperStepEvent + and not WorkflowErrorEvent + and not RequestInfoEvent + and not ExecutorFailedEvent) && + (workflowEvent is not ExecutorEvent executorEvent || executorEvent.ExecutorId != rootExecutorId); + + private static bool TargetsLoop(DialogAction action, Foreach loop) + { + BotElement? ancestor = action.Parent; + while (ancestor is not null && ancestor is not Foreach) + { + ancestor = ancestor.Parent; + } + + return ancestor is Foreach ancestorLoop && ancestorLoop.Id.Equals(loop.Id); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs index c739cb3bf9..8f5ca5cad7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs @@ -29,6 +29,8 @@ internal sealed class WorkflowFormulaState private readonly Dictionary _scopes; + private HashSet<(string ScopeName, string VariableName)>? _trackedChanges; + private int _isInitialized; public RecalcEngine Engine { get; } @@ -56,8 +58,76 @@ public FormulaValue Get(string variableName, string? scopeName = null) return FormulaValue.NewBlank(); } - public void Set(string variableName, FormulaValue value, string? scopeName = null) => - this.GetScope(scopeName ?? DefaultScopeName)[variableName] = value; + public void Set(string variableName, FormulaValue value, string? scopeName = null) + { + string normalizedScopeName = GetScopeName(scopeName); + this.GetScope(normalizedScopeName)[variableName] = value; + this._trackedChanges?.Add((normalizedScopeName, variableName)); + } + + /// + /// Captures a portable deep snapshot of every workflow variable scope. + /// + public WorkflowStateSnapshot CaptureSnapshot() + { + WorkflowStateEntry[] entries = + [ + .. VariableScopeNames.AllScopes + .Select(GetScopeName) + .Distinct() + .SelectMany( + scopeName => + this.Keys(scopeName) + .OrderBy(variableName => variableName, System.StringComparer.Ordinal) + .Select( + variableName => + new WorkflowStateEntry( + scopeName, + variableName, + new PortableValue(this.Get(variableName, scopeName).AsPortable())))), + ]; + + return new(entries); + } + + /// + /// Creates an isolated state instance from a previously captured snapshot. + /// + public static WorkflowFormulaState CreateBranch(RecalcEngine engine, WorkflowStateSnapshot snapshot) + { + WorkflowFormulaState branch = new(engine); + foreach (WorkflowStateEntry entry in snapshot.Entries) + { + branch.Set(entry.VariableName, entry.Value.ToFormula(), entry.ScopeName); + } + + branch.Bind(); + return branch; + } + + /// + /// Starts recording the variables written after branch initialization. + /// + public void BeginTrackingChanges() => this._trackedChanges = []; + + /// + /// Captures the current values of all variables written since change tracking began. + /// + public WorkflowStateChange[] CaptureChanges() => + this._trackedChanges is null + ? [] + : + [ + .. this._trackedChanges + .OrderBy(change => change.ScopeName, System.StringComparer.Ordinal) + .ThenBy(change => change.VariableName, System.StringComparer.Ordinal) + .Select( + change => + new WorkflowStateChange( + change.ScopeName, + change.VariableName, + new PortableValue(this.Get(change.VariableName, change.ScopeName).AsPortable()))), + ]; public bool SetInitialized() => Interlocked.CompareExchange(ref this._isInitialized, 1, 0) == 0; @@ -145,3 +215,9 @@ public static string GetScopeName(string? scopeName) /// private sealed class WorkflowScope : Dictionary; } + +internal sealed record WorkflowStateEntry(string ScopeName, string VariableName, PortableValue Value); + +internal sealed record WorkflowStateChange(string ScopeName, string VariableName, PortableValue Value); + +internal sealed record WorkflowStateSnapshot(IReadOnlyList Entries); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md index 2a50b6045d..4a8e47d9af 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md @@ -57,4 +57,37 @@ we've provided a console application that is able to execute any declarative wor |**Foreach**|Iterates through a collection of items, executing a set of actions for each. Ideal for processing lists or batch operations. |**GotoAction**|Jumps directly to a specified action within the workflow. Enables non-linear navigation in the logic flow. +## Parallel Foreach + +`Foreach` remains sequential when execution options are omitted. Set `mode` to `Parallel` to opt in, and use +`maxParallelism` to bound the number of active iterations. The default maximum in parallel mode is 4. +`timeoutInMilliseconds`, when present, applies to each iteration. + +```yaml +- kind: Foreach + id: translate_languages + items: =Local.TargetLanguages + value: Local.TargetLanguage + index: Local.TargetLanguageIndex + mode: Parallel + maxParallelism: 4 + timeoutInMilliseconds: 30000 + actions: + - kind: InvokeAzureAgent + id: translate + agent: + name: TranslatorAgent + input: + arguments: + language: =Local.TargetLanguage +``` + +Each iteration receives an isolated copy of workflow state. After all iterations succeed, state writes and emitted +events are applied in collection order; writes to the same variable therefore use deterministic last-index-wins +semantics. A failure or timeout cancels outstanding iterations and does not commit buffered state or events. + +Parallel iterations cannot currently suspend for external input because an in-flight parallel set cannot be safely +represented by a parent workflow checkpoint. `Question` and `RequestExternalInput` actions are rejected when the +workflow is built. An action that conditionally requests external input fails the parallel Foreach at runtime instead +of creating an unsafe checkpoint. `BreakLoop` and `ContinueLoop` cannot target a parallel Foreach. diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs new file mode 100644 index 0000000000..674f36519e --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs @@ -0,0 +1,661 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Globalization; +using System.IO; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Workflows.Declarative.UnitTests; + +/// +/// End-to-end behavioral tests for parallel declarative Foreach execution. +/// +public sealed class ParallelForeachWorkflowTests +{ + [Fact] + public async Task ForeachRemainsSequentialByDefaultAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: TimeSpan.FromMilliseconds(75)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\"]", + executionOptions: null, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(3, provider.InvocationCount); + Assert.Equal(1, provider.PeakConcurrency); + } + + [Fact] + public async Task ParallelForeachExecutesConcurrentlyAndHonorsLimitAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: TimeSpan.FromMilliseconds(250)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\", \"e\", \"f\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + timeoutInMilliseconds: 5000 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(6, provider.InvocationCount); + Assert.Equal(2, provider.PeakConcurrency); + } + + [Fact] + public async Task ParallelForeachUsesBoundedDefaultLimitAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 4, barrierTimeout: TimeSpan.FromMilliseconds(250)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\", \"e\", \"f\"]", + executionOptions: """ + mode: Parallel + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(6, provider.InvocationCount); + Assert.Equal(4, provider.PeakConcurrency); + } + + [Fact] + public async Task ParallelForeachIsolatesStateAndCommitsInSourceOrderAsync() + { + // Arrange + ControlledAgentProvider provider = new(delayByIndex: index => TimeSpan.FromMilliseconds((3 - index) * 30)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: $$""" + - kind: SetVariable + id: capture_last + variable: Local.LastValue + value: =Local.Item + + {{InvokeAgentAction}} + """, + afterActions: """ + - kind: SendActivity + id: report_final + activity: Final {Local.LastValue} + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(s_orderedResponses, GetAgentResponses(observation)); + Assert.Contains(observation.Events.OfType(), evt => evt.Message.Trim() == "Final d"); + Assert.Equal( + s_orderedResponses, + provider.Invocations.OrderBy(invocation => invocation.Index).Select(invocation => $"{invocation.Index}:{invocation.Value}")); + } + + [Fact] + public async Task ParallelForeachDeepCopiesComplexLoopValuesAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: TimeSpan.FromMilliseconds(250)); + string yaml = CreateWorkflowYaml( + items: "=[Local.Shared, Local.Shared]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: EditTable + id: mutate_item + itemsVariable: Local.Item + changeType: Add + value: ={ id: Local.Index } + + - kind: InvokeAzureAgent + id: invoke_agent + agent: + name: TestAgent + input: + arguments: + value: =Text(CountRows(Local.Item)) + index: =Local.Index + output: + autoSend: true + """, + afterActions: """ + - kind: SendActivity + id: report_shared_count + activity: Shared {CountRows(Local.Shared)} + """, + beforeActions: """ + - kind: SetVariable + id: initialize_shared + variable: Local.Shared + value: =[{ id: -1 }] + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:2", "1:2"], GetAgentResponses(observation)); + Assert.Contains(observation.Events.OfType(), evt => evt.Message.Trim() == "Shared 1"); + } + + [Fact] + public async Task ParallelForeachAggregatesBranchFailureAsync() + { + // Arrange + ControlledAgentProvider provider = new( + barrierParticipants: 4, + barrierTimeout: TimeSpan.FromMilliseconds(250), + failureIndexes: [1]); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + Exception error = AssertWorkflowError(observation); + Assert.Contains(Flatten(error), exception => exception is AggregateException); + Assert.Contains(Flatten(error), exception => exception.Message.Contains("iteration 1", StringComparison.OrdinalIgnoreCase)); + Assert.Empty(GetAgentResponses(observation)); + } + + [Fact] + public async Task ParallelForeachPropagatesCancellationAsync() + { + // Arrange + ControlledAgentProvider provider = new(waitUntilCanceled: true); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: InvokeAgentAction); + Workflow workflow = BuildWorkflow(yaml, provider); + await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, "input"); + Task watchTask = CollectEventsAsync(run); + await AwaitWithTimeoutAsync(provider.FirstInvocationStarted, TimeSpan.FromSeconds(5)); + + // Act + await run.CancelRunAsync(); + WorkflowEvent[] events = await AwaitWithTimeoutAsync(watchTask, TimeSpan.FromSeconds(5)); + RunStatus status = await run.GetStatusAsync(); + + // Assert + Assert.Equal(RunStatus.Ended, status); + await AwaitWithTimeoutAsync(provider.CancellationObserved, TimeSpan.FromSeconds(5)); + Assert.DoesNotContain(events.OfType(), evt => evt.ActionId == "parallel_loop"); + } + + [Fact] + public async Task ParallelForeachTimesOutIterationAndCancelsPeersAsync() + { + // Arrange + ControlledAgentProvider provider = new(waitUntilCanceled: true); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + timeoutInMilliseconds: 75 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + Exception error = AssertWorkflowError(observation); + Assert.Contains(Flatten(error), exception => exception is TimeoutException); + await AwaitWithTimeoutAsync(provider.CancellationObserved, TimeSpan.FromSeconds(5)); + } + + [Fact] + public async Task ParallelForeachHandlesEmptyCollectionAsync() + { + // Arrange + ControlledAgentProvider provider = new(); + string yaml = CreateWorkflowYaml( + items: "=[]", + executionOptions: """ + mode: Parallel + maxParallelism: 3 + """, + bodyActions: InvokeAgentAction, + afterActions: """ + - kind: SendActivity + id: empty_complete + activity: empty complete + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(0, provider.InvocationCount); + Assert.Contains(observation.Events.OfType(), evt => evt.ActionId == "parallel_loop"); + Assert.Contains(observation.Events.OfType(), evt => evt.Message.Trim() == "empty complete"); + } + + [Theory] + [InlineData("Parallel", 0)] + [InlineData("Parallel", -1)] + [InlineData("unsupported", 2)] + public void ParallelForeachRejectsInvalidConfiguration(string mode, int maxParallelism) + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: $$""" + mode: {{mode}} + maxParallelism: {{maxParallelism}} + """, + bodyActions: InvokeAgentAction); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("parallel", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void ParallelForeachRejectsInvalidTimeout(int timeoutInMilliseconds) + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: $$""" + mode: Parallel + maxParallelism: 2 + timeoutInMilliseconds: {{timeoutInMilliseconds}} + """, + bodyActions: InvokeAgentAction); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("timeout", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public void ParallelForeachRejectsCheckpointingBody() + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: Question + id: prompt_in_loop + alwaysPrompt: true + autoSend: false + property: Local.Answer + prompt: + kind: Message + text: + - answer + entity: + kind: StringPrebuiltEntity + """); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("checkpoint", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Theory] + [InlineData("BreakLoop")] + [InlineData("ContinueLoop")] + public void ParallelForeachRejectsOuterLoopControl(string actionKind) + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: $$""" + - kind: {{actionKind}} + id: unsupported_loop_control + """); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains(actionKind, exception.Message, StringComparison.Ordinal); + } + + [Fact] + public async Task ParallelForeachRejectsConditionalRequestAtRuntimeAsync() + { + // Arrange + ControlledAgentProvider provider = new(requestApproval: true); + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + Exception error = AssertWorkflowError(observation); + Assert.Contains(Flatten(error), exception => exception.Message.Contains("checkpoint", StringComparison.OrdinalIgnoreCase)); + Assert.DoesNotContain(observation.Events, workflowEvent => workflowEvent is RequestInfoEvent); + } + + private static readonly string[] s_orderedResponses = ["0:a", "1:b", "2:c", "3:d"]; + + private const string InvokeAgentAction = """ + - kind: InvokeAzureAgent + id: invoke_agent + agent: + name: TestAgent + input: + arguments: + value: =Local.Item + index: =Local.Index + output: + autoSend: true + """; + + private static string CreateWorkflowYaml( + string items, + string? executionOptions, + string bodyActions, + string? afterActions = null, + string? beforeActions = null) => + $$""" + kind: Workflow + trigger: + kind: OnConversationStart + id: workflow + actions: + {{beforeActions}} + - kind: Foreach + id: parallel_loop + items: {{items}} + value: Local.Item + index: Local.Index + {{executionOptions}} + actions: + {{bodyActions}} + {{afterActions}} + """; + + private static Workflow BuildWorkflow(string yaml, ResponseAgentProvider provider) + { + using StringReader reader = new(yaml); + return DeclarativeWorkflowBuilder.Build(reader, new DeclarativeWorkflowOptions(provider)); + } + + private static async Task RunWorkflowAsync(string yaml, ResponseAgentProvider provider) + { + using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(10)); + Workflow workflow = BuildWorkflow(yaml, provider); + await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, "input", cancellationToken: timeout.Token).ConfigureAwait(false); + WorkflowEvent[] events = await CollectEventsAsync(run, timeout.Token).ConfigureAwait(false); + RunStatus status = await run.GetStatusAsync(timeout.Token).ConfigureAwait(false); + return new(events, status); + } + + private static async Task CollectEventsAsync(StreamingRun run, CancellationToken cancellationToken = default) + { + List events = []; + await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync(cancellationToken).ConfigureAwait(false)) + { + events.Add(workflowEvent); + } + + return [.. events]; + } + + private static void AssertNoWorkflowError(WorkflowObservation observation) + { + Assert.DoesNotContain(observation.Events, workflowEvent => workflowEvent is WorkflowErrorEvent); + Assert.Equal(RunStatus.Idle, observation.Status); + } + + private static Exception AssertWorkflowError(WorkflowObservation observation) => + Assert.IsAssignableFrom(Assert.Single(observation.Events.OfType()).Data); + + private static string[] GetAgentResponses(WorkflowObservation observation) => + [.. observation.Events + .OfType() + .Where(evt => evt.ExecutorId == "invoke_agent") + .Select(evt => evt.Response.Messages.Single().Text)]; + + private static IEnumerable Flatten(Exception exception) + { + yield return exception; + + if (exception is AggregateException aggregateException) + { + foreach (Exception innerException in aggregateException.InnerExceptions.SelectMany(Flatten)) + { + yield return innerException; + } + } + else if (exception.InnerException is not null) + { + foreach (Exception innerException in Flatten(exception.InnerException)) + { + yield return innerException; + } + } + } + + private static async Task AwaitWithTimeoutAsync(Task task, TimeSpan timeout) + { + Task completedTask = await Task.WhenAny(task, Task.Delay(timeout)).ConfigureAwait(false); + Assert.Same(task, completedTask); + await task.ConfigureAwait(false); + } + + private static async Task AwaitWithTimeoutAsync(Task task, TimeSpan timeout) + { + await AwaitWithTimeoutAsync((Task)task, timeout).ConfigureAwait(false); + return await task.ConfigureAwait(false); + } + + private sealed record WorkflowObservation(WorkflowEvent[] Events, RunStatus Status); + + private sealed record Invocation(int Index, string Value); + + private sealed class ControlledAgentProvider : ResponseAgentProvider + { + private readonly TaskCompletionSource _barrier = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _cancellationObserved = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _firstInvocationStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly int _barrierParticipants; + private readonly TimeSpan _barrierTimeout; + private readonly Func? _delayByIndex; + private readonly HashSet _failureIndexes; + private readonly bool _waitUntilCanceled; + private readonly bool _requestApproval; + private int _activeCount; + private int _barrierArrivals; + private int _invocationCount; + private int _peakConcurrency; + + public ControlledAgentProvider( + int barrierParticipants = 1, + TimeSpan? barrierTimeout = null, + Func? delayByIndex = null, + IEnumerable? failureIndexes = null, + bool waitUntilCanceled = false, + bool requestApproval = false) + { + this._barrierParticipants = barrierParticipants; + this._barrierTimeout = barrierTimeout ?? TimeSpan.Zero; + this._delayByIndex = delayByIndex; + this._failureIndexes = failureIndexes is null ? [] : [.. failureIndexes]; + this._waitUntilCanceled = waitUntilCanceled; + this._requestApproval = requestApproval; + } + + public int InvocationCount => Volatile.Read(ref this._invocationCount); + + public int PeakConcurrency => Volatile.Read(ref this._peakConcurrency); + + public ConcurrentBag Invocations { get; } = []; + + public Task CancellationObserved => this._cancellationObserved.Task; + + public Task FirstInvocationStarted => this._firstInvocationStarted.Task; + + public override Task CreateConversationAsync(CancellationToken cancellationToken = default) => + Task.FromResult(Guid.NewGuid().ToString("N")); + + public override Task CreateMessageAsync(string conversationId, ChatMessage conversationMessage, CancellationToken cancellationToken = default) => + Task.FromResult(conversationMessage); + + public override Task GetMessageAsync(string conversationId, string messageId, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public override async IAsyncEnumerable InvokeAgentAsync( + string agentId, + string? agentVersion, + string? conversationId, + IEnumerable? messages, + IDictionary? inputArguments, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + int index = Convert.ToInt32(inputArguments!["index"], CultureInfo.InvariantCulture); + string value = Convert.ToString(inputArguments["value"], CultureInfo.InvariantCulture)!; + this.Invocations.Add(new(index, value)); + Interlocked.Increment(ref this._invocationCount); + this._firstInvocationStarted.TrySetResult(true); + + int activeCount = Interlocked.Increment(ref this._activeCount); + UpdatePeak(ref this._peakConcurrency, activeCount); + using CancellationTokenRegistration registration = cancellationToken.Register(() => this._cancellationObserved.TrySetResult(true)); + + try + { + if (Interlocked.Increment(ref this._barrierArrivals) == this._barrierParticipants) + { + this._barrier.TrySetResult(true); + } + + if (this._barrierParticipants > 1) + { + await Task.WhenAny(this._barrier.Task, Task.Delay(this._barrierTimeout, cancellationToken)).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); + } + + if (this._waitUntilCanceled) + { + await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false); + } + + if (this._delayByIndex is not null) + { + await Task.Delay(this._delayByIndex(index), cancellationToken).ConfigureAwait(false); + } + + if (this._failureIndexes.Contains(index)) + { + throw new InvalidOperationException($"Failure for iteration {index}."); + } + + if (this._requestApproval) + { + yield return new AgentResponseUpdate( + ChatRole.Assistant, + [new ToolApprovalRequestContent("approval", new FunctionCallContent("approval", "test"))]); + } + else + { + yield return new AgentResponseUpdate(ChatRole.Assistant, $"{index}:{value}"); + } + } + finally + { + if (cancellationToken.IsCancellationRequested) + { + this._cancellationObserved.TrySetResult(true); + } + + Interlocked.Decrement(ref this._activeCount); + } + } + + public override async IAsyncEnumerable GetMessagesAsync( + string conversationId, + int? limit = null, + string? after = null, + string? before = null, + bool newestFirst = false, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.CompletedTask.ConfigureAwait(false); + yield break; + } + + private static void UpdatePeak(ref int peak, int candidate) + { + int observed; + do + { + observed = Volatile.Read(ref peak); + if (candidate <= observed) + { + return; + } + } + while (Interlocked.CompareExchange(ref peak, candidate, observed) != observed); + } + } +} From 3da94a7cecb96878c4acd164ab0d69ca59d9e7ca Mon Sep 17 00:00:00 2001 From: KirschQAQ <114209152+KirschBluteX@users.noreply.github.com> Date: Sat, 15 Aug 2026 08:02:49 -0700 Subject: [PATCH 2/4] fix: address parallel foreach review feedback --- .../ObjectModel/ForeachExecutionOptions.cs | 2 +- .../ObjectModel/ParallelForeachIterationRunner.cs | 15 ++++++--------- .../PowerFx/WorkflowFormulaState.cs | 1 - .../ParallelForeachWorkflowTests.cs | 14 ++++++++------ 4 files changed, 15 insertions(+), 17 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs index 8545838dde..322ace82ed 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs @@ -111,5 +111,5 @@ private static ForeachExecutionMode ParseMode(Foreach model, DataValue? value) 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 parallel Foreach configuration for '{model.Id.Value}': {message}", innerException); + new($"Invalid Foreach configuration for '{model.Id.Value}': {message}", innerException); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs index 34b3595f9e..0a4714b912 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs @@ -94,7 +94,7 @@ public static async Task RunAsync( if (timeoutSource.IsCancellationRequested) { throw new TimeoutException( - $"Parallel Foreach '{model.Id.Value}' iteration {index} exceeded its timeout of {timeout.GetValueOrDefault().TotalMilliseconds} ms."); + $"The iteration exceeded its timeout of {timeout.GetValueOrDefault().TotalMilliseconds} ms."); } cancellationToken.ThrowIfCancellationRequested(); @@ -110,27 +110,24 @@ .. events workflowError => workflowError.Data as Exception ?? new DeclarativeActionException( - $"Parallel Foreach '{model.Id.Value}' iteration {index} failed without exception data.")), + "The iteration failed without exception data.")), ]; if (failures.Length > 0) { - Exception innerException = failures.Length == 1 ? failures[0] : new AggregateException(failures); - throw new DeclarativeActionException( - $"Parallel Foreach '{model.Id.Value}' iteration {index} failed.", - innerException); + throw failures.Length == 1 ? failures[0] : new AggregateException(failures); } if (status == RunStatus.PendingRequests || events.Any(workflowEvent => workflowEvent is RequestInfoEvent)) { throw new DeclarativeActionException( - $"Parallel Foreach '{model.Id.Value}' iteration {index} requested external input. " + + "The iteration requested external input. " + "Checkpointing an in-flight parallel iteration is not supported."); } if (status != RunStatus.Idle) { throw new DeclarativeActionException( - $"Parallel Foreach '{model.Id.Value}' iteration {index} ended with unsupported status '{status}'."); + $"The iteration ended with unsupported status '{status}'."); } WorkflowStateChange[] stateChanges = @@ -146,7 +143,7 @@ .. branchState catch (OperationCanceledException) when (timeoutSource.IsCancellationRequested) { throw new TimeoutException( - $"Parallel Foreach '{model.Id.Value}' iteration {index} exceeded its timeout of {timeout!.Value.TotalMilliseconds} ms."); + $"The iteration exceeded its timeout of {timeout!.Value.TotalMilliseconds} ms."); } finally { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs index 8f5ca5cad7..a39aef3911 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs @@ -101,7 +101,6 @@ public static WorkflowFormulaState CreateBranch(RecalcEngine engine, WorkflowSta branch.Set(entry.VariableName, entry.Value.ToFormula(), entry.ScopeName); } - branch.Bind(); return branch; } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs index 674f36519e..a6229c4424 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs @@ -23,7 +23,7 @@ public sealed class ParallelForeachWorkflowTests public async Task ForeachRemainsSequentialByDefaultAsync() { // Arrange - ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: TimeSpan.FromMilliseconds(75)); + ControlledAgentProvider provider = new(); string yaml = CreateWorkflowYaml( items: "=[\"a\", \"b\", \"c\"]", executionOptions: null, @@ -42,7 +42,7 @@ public async Task ForeachRemainsSequentialByDefaultAsync() public async Task ParallelForeachExecutesConcurrentlyAndHonorsLimitAsync() { // Arrange - ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: TimeSpan.FromMilliseconds(250)); + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: s_barrierTimeout); string yaml = CreateWorkflowYaml( items: "=[\"a\", \"b\", \"c\", \"d\", \"e\", \"f\"]", executionOptions: """ @@ -65,7 +65,7 @@ public async Task ParallelForeachExecutesConcurrentlyAndHonorsLimitAsync() public async Task ParallelForeachUsesBoundedDefaultLimitAsync() { // Arrange - ControlledAgentProvider provider = new(barrierParticipants: 4, barrierTimeout: TimeSpan.FromMilliseconds(250)); + ControlledAgentProvider provider = new(barrierParticipants: 4, barrierTimeout: s_barrierTimeout); string yaml = CreateWorkflowYaml( items: "=[\"a\", \"b\", \"c\", \"d\", \"e\", \"f\"]", executionOptions: """ @@ -123,7 +123,7 @@ public async Task ParallelForeachIsolatesStateAndCommitsInSourceOrderAsync() public async Task ParallelForeachDeepCopiesComplexLoopValuesAsync() { // Arrange - ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: TimeSpan.FromMilliseconds(250)); + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: s_barrierTimeout); string yaml = CreateWorkflowYaml( items: "=[Local.Shared, Local.Shared]", executionOptions: """ @@ -175,7 +175,7 @@ public async Task ParallelForeachAggregatesBranchFailureAsync() // Arrange ControlledAgentProvider provider = new( barrierParticipants: 4, - barrierTimeout: TimeSpan.FromMilliseconds(250), + barrierTimeout: s_barrierTimeout, failureIndexes: [1]); string yaml = CreateWorkflowYaml( items: "=[\"a\", \"b\", \"c\", \"d\"]", @@ -233,7 +233,7 @@ public async Task ParallelForeachTimesOutIterationAndCancelsPeersAsync() executionOptions: """ mode: Parallel maxParallelism: 2 - timeoutInMilliseconds: 75 + timeoutInMilliseconds: 1000 """, bodyActions: InvokeAgentAction); @@ -397,6 +397,8 @@ public async Task ParallelForeachRejectsConditionalRequestAtRuntimeAsync() private static readonly string[] s_orderedResponses = ["0:a", "1:b", "2:c", "3:d"]; + private static readonly TimeSpan s_barrierTimeout = TimeSpan.FromSeconds(5); + private const string InvokeAgentAction = """ - kind: InvokeAzureAgent id: invoke_agent From f0db1564f650c3743ed0594403b3e3a7b07103b4 Mon Sep 17 00:00:00 2001 From: KirschQAQ <114209152+KirschBluteX@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:31:18 -0700 Subject: [PATCH 3/4] fix: validate parallel foreach execution options --- .../ObjectModel/ForeachExecutionOptions.cs | 15 +- .../ParallelForeachWorkflowTests.cs | 268 +++++++++++++++++- 2 files changed, 276 insertions(+), 7 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs index 322ace82ed..401139adf0 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutionOptions.cs @@ -67,13 +67,22 @@ private static ForeachExecutionMode ParseMode(Foreach model, DataValue? value) return ForeachExecutionMode.Sequential; } - if (value is not StringDataValue stringValue || - !Enum.TryParse(stringValue.Value, ignoreCase: true, out ForeachExecutionMode mode)) + if (value is not StringDataValue stringValue) { throw InvalidConfiguration(model, $"'{ModePropertyName}' must be 'Sequential' or 'Parallel'."); } - return mode; + 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) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs index a6229c4424..54f2ed0b69 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs @@ -82,11 +82,58 @@ public async Task ParallelForeachUsesBoundedDefaultLimitAsync() Assert.Equal(4, provider.PeakConcurrency); } + [Fact] + public async Task ParallelForeachCapsWorkerCountAtItemCountForExtremeLimitAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: s_barrierTimeout); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2147483647 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(2, provider.InvocationCount); + Assert.Equal(2, provider.PeakConcurrency); + } + + [Fact] + public async Task ParallelForeachKeepsDuplicateValuesSeparatedByIndexAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 3, barrierTimeout: s_barrierTimeout); + string yaml = CreateWorkflowYaml( + items: "=[\"same\", \"same\", \"same\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 3 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:same", "1:same", "2:same"], GetAgentResponses(observation)); + Assert.Equal(3, provider.PeakConcurrency); + } + [Fact] public async Task ParallelForeachIsolatesStateAndCommitsInSourceOrderAsync() { // Arrange - ControlledAgentProvider provider = new(delayByIndex: index => TimeSpan.FromMilliseconds((3 - index) * 30)); + ControlledAgentProvider provider = new( + barrierParticipants: 4, + barrierTimeout: s_barrierTimeout, + delayByIndex: index => TimeSpan.FromMilliseconds((3 - index) * 100)); string yaml = CreateWorkflowYaml( items: "=[\"a\", \"b\", \"c\", \"d\"]", executionOptions: """ @@ -114,6 +161,7 @@ public async Task ParallelForeachIsolatesStateAndCommitsInSourceOrderAsync() AssertNoWorkflowError(observation); Assert.Equal(s_orderedResponses, GetAgentResponses(observation)); Assert.Contains(observation.Events.OfType(), evt => evt.Message.Trim() == "Final d"); + Assert.Equal([3, 2, 1, 0], provider.Completions); Assert.Equal( s_orderedResponses, provider.Invocations.OrderBy(invocation => invocation.Index).Select(invocation => $"{invocation.Index}:{invocation.Value}")); @@ -183,7 +231,13 @@ public async Task ParallelForeachAggregatesBranchFailureAsync() mode: Parallel maxParallelism: 4 """, - bodyActions: InvokeAgentAction); + bodyActions: $$""" + - kind: SendActivity + id: buffered_before_failure + activity: Branch {Local.Index} + + {{InvokeAgentAction}} + """); // Act WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); @@ -193,6 +247,36 @@ public async Task ParallelForeachAggregatesBranchFailureAsync() Assert.Contains(Flatten(error), exception => exception is AggregateException); Assert.Contains(Flatten(error), exception => exception.Message.Contains("iteration 1", StringComparison.OrdinalIgnoreCase)); Assert.Empty(GetAgentResponses(observation)); + Assert.DoesNotContain(observation.Events.OfType(), evt => evt.Message.Trim().StartsWith("Branch ", StringComparison.Ordinal)); + } + + [Fact] + public async Task ParallelForeachFailureCancelsActivePeersAsync() + { + // Arrange + ControlledAgentProvider provider = new( + barrierParticipants: 4, + barrierTimeout: s_barrierTimeout, + failureIndexes: [0], + waitUntilCanceledIndexes: [1, 2, 3]); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + Exception error = AssertWorkflowError(observation); + Assert.Contains(Flatten(error), exception => exception.Message.Contains("iteration 0", StringComparison.OrdinalIgnoreCase)); + await AwaitWithTimeoutAsync(provider.CancellationObserved, TimeSpan.FromSeconds(5)); + await AwaitWithTimeoutAsync(provider.NoActiveInvocations, TimeSpan.FromSeconds(5)); + Assert.Equal(0, provider.ActiveCount); + Assert.Empty(GetAgentResponses(observation)); } [Fact] @@ -220,7 +304,10 @@ public async Task ParallelForeachPropagatesCancellationAsync() // Assert Assert.Equal(RunStatus.Ended, status); await AwaitWithTimeoutAsync(provider.CancellationObserved, TimeSpan.FromSeconds(5)); + await AwaitWithTimeoutAsync(provider.NoActiveInvocations, TimeSpan.FromSeconds(5)); + Assert.Equal(0, provider.ActiveCount); Assert.DoesNotContain(events.OfType(), evt => evt.ActionId == "parallel_loop"); + Assert.DoesNotContain(events.OfType(), evt => evt.ActionId == "invoke_agent"); } [Fact] @@ -244,6 +331,10 @@ public async Task ParallelForeachTimesOutIterationAndCancelsPeersAsync() Exception error = AssertWorkflowError(observation); Assert.Contains(Flatten(error), exception => exception is TimeoutException); await AwaitWithTimeoutAsync(provider.CancellationObserved, TimeSpan.FromSeconds(5)); + await AwaitWithTimeoutAsync(provider.NoActiveInvocations, TimeSpan.FromSeconds(5)); + Assert.Equal(0, provider.ActiveCount); + Assert.Empty(GetAgentResponses(observation)); + Assert.DoesNotContain(observation.Events.OfType(), evt => evt.ActionId == "invoke_agent"); } [Fact] @@ -274,10 +365,79 @@ public async Task ParallelForeachHandlesEmptyCollectionAsync() Assert.Contains(observation.Events.OfType(), evt => evt.Message.Trim() == "empty complete"); } + [Fact] + public async Task ParallelForeachCheckpointResumesAfterCommittedIterationsAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: s_barrierTimeout); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: $$""" + - kind: SetVariable + id: capture_checkpoint_value + variable: Local.LastValue + value: =Local.Item + + {{InvokeAgentAction}} + """, + afterActions: """ + - kind: SendActivity + id: after_parallel + activity: after {Local.LastValue} + """); + CheckpointManager checkpointManager = CheckpointManager.CreateInMemory(); + CheckpointInfo? checkpointAfterLoop = null; + bool loopCompleted = false; + + // Act + await using (StreamingRun firstRun = await InProcessExecution.RunStreamingAsync( + BuildWorkflow(yaml, provider), + "input", + checkpointManager)) + { + await foreach (WorkflowEvent workflowEvent in firstRun.WatchStreamAsync()) + { + if (workflowEvent is DeclarativeActionCompletedEvent { ActionId: "parallel_loop" }) + { + loopCompleted = true; + } + + if (loopCompleted && + checkpointAfterLoop is null && + workflowEvent is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } checkpoint }) + { + checkpointAfterLoop = checkpoint; + } + } + } + + int invocationCountBeforeResume = provider.InvocationCount; + await using StreamingRun resumedRun = await InProcessExecution.ResumeStreamingAsync( + BuildWorkflow(yaml, provider), + Assert.IsType(checkpointAfterLoop), + checkpointManager); + WorkflowEvent[] resumedEvents = await CollectEventsAsync(resumedRun); + + // Assert + Assert.Equal(2, invocationCountBeforeResume); + Assert.Equal(invocationCountBeforeResume, provider.InvocationCount); + Assert.DoesNotContain(resumedEvents.OfType(), evt => evt.ActionId == "parallel_loop"); + Assert.DoesNotContain(resumedEvents, workflowEvent => workflowEvent is AgentResponseEvent { ExecutorId: "invoke_agent" }); + Assert.Contains(resumedEvents.OfType(), evt => evt.Message.Trim() == "after b"); + } + [Theory] [InlineData("Parallel", 0)] [InlineData("Parallel", -1)] [InlineData("unsupported", 2)] + [InlineData("\"1\"", 2)] + [InlineData("Sequential, Parallel", 2)] + [InlineData("null", 2)] + [InlineData("true", 2)] public void ParallelForeachRejectsInvalidConfiguration(string mode, int maxParallelism) { // Arrange @@ -318,6 +478,56 @@ public void ParallelForeachRejectsInvalidTimeout(int timeoutInMilliseconds) Assert.Contains("timeout", exception.Message, StringComparison.OrdinalIgnoreCase); } + [Theory] + [InlineData("1.5")] + [InlineData("2147483648")] + [InlineData("\"2\"")] + [InlineData("-2147483649")] + [InlineData("null")] + [InlineData("true")] + public void ParallelForeachRejectsNonIntegerOrOutOfRangeMaxParallelism(string maxParallelism) + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: $$""" + mode: Parallel + maxParallelism: {{maxParallelism}} + """, + bodyActions: InvokeAgentAction); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("maxParallelism", exception.Message, StringComparison.Ordinal); + } + + [Theory] + [InlineData("1.5")] + [InlineData("2147483648")] + [InlineData("\"1000\"")] + [InlineData("-2147483649")] + [InlineData("null")] + [InlineData("true")] + public void ParallelForeachRejectsNonIntegerOrOutOfRangeTimeout(string timeoutInMilliseconds) + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: $$""" + mode: Parallel + timeoutInMilliseconds: {{timeoutInMilliseconds}} + """, + bodyActions: InvokeAgentAction); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("timeout", exception.Message, StringComparison.OrdinalIgnoreCase); + } + [Fact] public void ParallelForeachRejectsCheckpointingBody() { @@ -373,6 +583,41 @@ public void ParallelForeachRejectsOuterLoopControl(string actionKind) Assert.Contains(actionKind, exception.Message, StringComparison.Ordinal); } + [Theory] + [InlineData("BreakLoop")] + [InlineData("ContinueLoop")] + public async Task ParallelForeachAllowsNestedSequentialLoopControlAsync(string actionKind) + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 2, barrierTimeout: s_barrierTimeout); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: $$""" + - kind: Foreach + id: inner_loop + items: =[1, 2] + value: Local.InnerItem + index: Local.InnerIndex + actions: + - kind: {{actionKind}} + id: control_inner_loop + + {{InvokeAgentAction}} + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:a", "1:b"], GetAgentResponses(observation)); + Assert.Equal(2, provider.PeakConcurrency); + } + [Fact] public async Task ParallelForeachRejectsConditionalRequestAtRuntimeAsync() { @@ -520,10 +765,12 @@ private sealed class ControlledAgentProvider : ResponseAgentProvider private readonly TaskCompletionSource _barrier = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _cancellationObserved = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _firstInvocationStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _noActiveInvocations = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly int _barrierParticipants; private readonly TimeSpan _barrierTimeout; private readonly Func? _delayByIndex; private readonly HashSet _failureIndexes; + private readonly HashSet _waitUntilCanceledIndexes; private readonly bool _waitUntilCanceled; private readonly bool _requestApproval; private int _activeCount; @@ -536,6 +783,7 @@ public ControlledAgentProvider( TimeSpan? barrierTimeout = null, Func? delayByIndex = null, IEnumerable? failureIndexes = null, + IEnumerable? waitUntilCanceledIndexes = null, bool waitUntilCanceled = false, bool requestApproval = false) { @@ -543,6 +791,7 @@ public ControlledAgentProvider( this._barrierTimeout = barrierTimeout ?? TimeSpan.Zero; this._delayByIndex = delayByIndex; this._failureIndexes = failureIndexes is null ? [] : [.. failureIndexes]; + this._waitUntilCanceledIndexes = waitUntilCanceledIndexes is null ? [] : [.. waitUntilCanceledIndexes]; this._waitUntilCanceled = waitUntilCanceled; this._requestApproval = requestApproval; } @@ -551,12 +800,18 @@ public ControlledAgentProvider( public int PeakConcurrency => Volatile.Read(ref this._peakConcurrency); + public int ActiveCount => Volatile.Read(ref this._activeCount); + public ConcurrentBag Invocations { get; } = []; + public ConcurrentQueue Completions { get; } = []; + public Task CancellationObserved => this._cancellationObserved.Task; public Task FirstInvocationStarted => this._firstInvocationStarted.Task; + public Task NoActiveInvocations => this._noActiveInvocations.Task; + public override Task CreateConversationAsync(CancellationToken cancellationToken = default) => Task.FromResult(Guid.NewGuid().ToString("N")); @@ -597,7 +852,7 @@ public override async IAsyncEnumerable InvokeAgentAsync( cancellationToken.ThrowIfCancellationRequested(); } - if (this._waitUntilCanceled) + if (this._waitUntilCanceled || this._waitUntilCanceledIndexes.Contains(index)) { await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false); } @@ -612,6 +867,8 @@ public override async IAsyncEnumerable InvokeAgentAsync( throw new InvalidOperationException($"Failure for iteration {index}."); } + this.Completions.Enqueue(index); + if (this._requestApproval) { yield return new AgentResponseUpdate( @@ -630,7 +887,10 @@ public override async IAsyncEnumerable InvokeAgentAsync( this._cancellationObserved.TrySetResult(true); } - Interlocked.Decrement(ref this._activeCount); + if (Interlocked.Decrement(ref this._activeCount) == 0) + { + this._noActiveInvocations.TrySetResult(true); + } } } From 7cc349bf7210888e065a8ca45418805a0e20784f Mon Sep 17 00:00:00 2001 From: KirschQAQ <114209152+KirschBluteX@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:21:13 -0700 Subject: [PATCH 4/4] fix: harden parallel declarative foreach semantics --- .../Extensions/AgentProviderExtensions.cs | 54 +- .../Extensions/IWorkflowContextExtensions.cs | 3 + .../Interpreter/DeclarativeActionExecutor.cs | 30 +- .../Interpreter/DeclarativeWorkflowContext.cs | 1 + .../AddConversationMessageExecutor.cs | 6 + .../CopyConversationMessagesExecutor.cs | 6 + .../ObjectModel/ForeachExecutor.cs | 70 ++- .../ObjectModel/HttpRequestExecutor.cs | 6 + .../ObjectModel/InvokeAzureAgentExecutor.cs | 6 + .../ObjectModel/InvokeFunctionToolExecutor.cs | 6 + .../ObjectModel/InvokeMcpToolExecutor.cs | 6 + .../ParallelForeachIterationRunner.cs | 93 +++- .../PowerFx/WorkflowFormulaState.cs | 16 +- .../README.md | 26 +- .../ParallelForeachWorkflowTests.cs | 515 +++++++++++++++++- 15 files changed, 804 insertions(+), 40 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs index bd9d590987..a68c2e9abd 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs @@ -3,8 +3,11 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Text.Json; +using System.Text.Json.Serialization.Metadata; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Declarative.Interpreter; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows.Declarative.Extensions; @@ -29,6 +32,17 @@ public static async ValueTask InvokeAgentAsync( // contract here. Workflow.AsAIAgent separately removes matching streamed/completed // message duplicates at its hosting boundary. bool isWorkflowConversation = context.IsWorkflowConversation(conversationId, out string? workflowConversationId); + WorkflowConversationMessageBuffer? conversationMessageBuffer = + (context as DeclarativeWorkflowContext)?.ConversationMessageBuffer; + + // An agent provider may mutate the supplied conversation while it is invoked. A parallel + // branch cannot safely share the workflow conversation, so reject an explicit use before + // the provider is called. Responses targeting the workflow conversation are staged below. + if (conversationMessageBuffer is not null && isWorkflowConversation) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{executorId}' cannot invoke an agent with the shared workflow conversation."); + } autoSend |= isWorkflowConversation; // Assign stable IDs to content-bearing chat updates before emitting and aggregating them. @@ -100,7 +114,14 @@ public static async ValueTask InvokeAgentAsync( { foreach (ChatMessage message in response.Messages) { - await agentProvider.CreateMessageAsync(workflowConversationId, message, cancellationToken).ConfigureAwait(false); + if (conversationMessageBuffer is null) + { + await agentProvider.CreateMessageAsync(workflowConversationId, message, cancellationToken).ConfigureAwait(false); + } + else + { + conversationMessageBuffer.Add(message); + } } } @@ -154,3 +175,34 @@ private static void ThrowIfFailed(AgentResponse response, string agentName) throw new DeclarativeActionException($"Agent '{agentName}' failed [{errorCode}]: {errorMessage}"); } } + +/// +/// Owns workflow-conversation messages produced by one parallel Foreach branch until the +/// parent executor can replay them in source order. +/// +internal sealed class WorkflowConversationMessageBuffer +{ + private static readonly JsonTypeInfo s_messageTypeInfo = + (JsonTypeInfo)AgentAbstractionsJsonUtilities.DefaultOptions.GetTypeInfo(typeof(ChatMessage)); + + private readonly List _messages = []; + + public IReadOnlyList Messages => this._messages; + + public void Add(ChatMessage message) + { + try + { + JsonElement serialized = JsonSerializer.SerializeToElement(message, s_messageTypeInfo); + ChatMessage copy = serialized.Deserialize(s_messageTypeInfo) + ?? throw new JsonException("The staged conversation message deserialized to null."); + this._messages.Add(copy); + } + catch (Exception exception) when (exception is JsonException or NotSupportedException) + { + throw new DeclarativeActionException( + "A parallel Foreach response could not be isolated for deterministic conversation replay.", + exception); + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs index 1b92235eee..697a61457a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs @@ -85,6 +85,9 @@ public static bool IsWorkflowConversation( return workflowConversationId?.Equals(conversationId, StringComparison.Ordinal) ?? false; } + internal static bool IsParallelForeachBranch(this IWorkflowContext context) => + context is DeclarativeWorkflowContext { ConversationMessageBuffer: not null }; + private static DeclarativeWorkflowContext DeclarativeContext(IWorkflowContext context) { if (context is not DeclarativeWorkflowContext declarativeContext) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs index 776caebda6..6dbe80a939 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs @@ -98,13 +98,23 @@ public override async ValueTask HandleAsync(ActionExecutorResult message, IWorkf } catch (DeclarativeActionException exception) { + if (!IsCanceled(exception, cancellationToken)) + { + this._state.ParallelFailureReporter?.Invoke(exception); + } Debug.WriteLine($"ERROR [{this.Id}] {exception.GetType().Name}\n{exception.Message}"); throw; } catch (Exception exception) { + DeclarativeActionException wrappedException = + new($"Unhandled workflow failure - #{this.Id} ({this.Model.GetType().Name})", exception); + if (!IsCanceled(exception, cancellationToken)) + { + this._state.ParallelFailureReporter?.Invoke(wrappedException); + } Debug.WriteLine($"ERROR [{this.Id}] {exception.GetType().Name}\n{exception.Message}"); - throw new DeclarativeActionException($"Unhandled workflow failure - #{this.Id} ({this.Model.GetType().Name})", exception); + throw wrappedException; } finally { @@ -117,6 +127,24 @@ public override async ValueTask HandleAsync(ActionExecutorResult message, IWorkf protected abstract ValueTask ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken = default); + private static bool IsCanceled(Exception exception, CancellationToken cancellationToken) + { + if (!cancellationToken.IsCancellationRequested) + { + return false; + } + + for (Exception? current = exception; current is not null; current = current.InnerException) + { + if (current is OperationCanceledException) + { + return true; + } + } + + return false; + } + /// /// Restore the state of the executor from a checkpoint. /// This must be overridden to restore any state that was saved during checkpointing. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs index 6616aa5d00..262d0e84fb 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs @@ -32,6 +32,7 @@ public DeclarativeWorkflowContext(IWorkflowContext source, WorkflowFormulaState private IWorkflowContext Source { get; } public WorkflowFormulaState State { get; } + internal WorkflowConversationMessageBuffer? ConversationMessageBuffer => this.State.ConversationMessageBuffer; public IReadOnlyDictionary? TraceContext => this.Source.TraceContext; /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/AddConversationMessageExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/AddConversationMessageExecutor.cs index 21c14de546..3c3daa4bc8 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/AddConversationMessageExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/AddConversationMessageExecutor.cs @@ -21,6 +21,12 @@ internal sealed class AddConversationMessageExecutor(AddConversationMessage mode Throw.IfNull(this.Model.ConversationId, $"{nameof(this.Model)}.{nameof(this.Model.ConversationId)}"); string conversationId = this.Evaluator.GetValue(this.Model.ConversationId).Value; + if (context.IsParallelForeachBranch()) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{this.Id}' cannot mutate a conversation immediately."); + } + bool isWorkflowConversation = context.IsWorkflowConversation(conversationId, out string? _); ChatMessage newMessage = new(this.Model.Role.Value.ToChatRole(), [.. this.GetContent()]) { AdditionalProperties = this.GetMetadata() }; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/CopyConversationMessagesExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/CopyConversationMessagesExecutor.cs index 381abcb84d..d46b325015 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/CopyConversationMessagesExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/CopyConversationMessagesExecutor.cs @@ -20,6 +20,12 @@ internal sealed class CopyConversationMessagesExecutor(CopyConversationMessages { Throw.IfNull(this.Model.ConversationId, $"{nameof(this.Model)}.{nameof(this.Model.ConversationId)}"); string conversationId = this.Evaluator.GetValue(this.Model.ConversationId).Value; + if (context.IsParallelForeachBranch()) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{this.Id}' cannot mutate a conversation immediately."); + } + bool isWorkflowConversation = context.IsWorkflowConversation(conversationId, out string? _); IEnumerable? inputMessages = this.GetInputMessages(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs index db85cbed6e..722563a7ab 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft. All rights reserved. using System; +using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -9,6 +10,7 @@ using Microsoft.Agents.AI.Workflows.Declarative.PowerFx; using Microsoft.Agents.ObjectModel; using Microsoft.Agents.ObjectModel.Abstractions; +using Microsoft.Extensions.AI; using Microsoft.PowerFx.Types; using Microsoft.Shared.Diagnostics; @@ -98,6 +100,9 @@ public ForeachExecutor(Foreach model, WorkflowFormulaState state, DeclarativeWor Exception?[] iterationFailures = new Exception?[values.Length]; int nextIndex = -1; + void RecordIterationFailure(int index, Exception exception) => + Interlocked.CompareExchange(ref iterationFailures[index], exception, null); + using CancellationTokenSource groupCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); int workerCount = Math.Min(values.Length, this._executionOptions.MaxParallelism); Task[] workers = @@ -108,24 +113,27 @@ .. 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(), - ]; - if (failures.Length > 0) + List failures = []; + for (int index = 0; index < iterationFailures.Length; index++) + { + if (iterationFailures[index] is Exception exception) + { + failures.Add( + new DeclarativeActionException( + $"Parallel Foreach '{this.Id}' iteration {index} failed.", + exception)); + } + } + + if (failures.Count > 0) { throw new AggregateException($"Parallel Foreach '{this.Id}' failed.", failures); } + WorkflowConversationMessageBuffer? parentConversationBuffer = + (context as DeclarativeWorkflowContext)?.ConversationMessageBuffer; + string? workflowConversationId = parentConversationBuffer is null ? context.GetWorkflowConversation() : null; + foreach (ParallelForeachIterationResult iterationResult in iterationResults.Cast()) { foreach (WorkflowStateChange stateChange in iterationResult.StateChanges) @@ -137,6 +145,32 @@ exception is null { await context.AddEventAsync(workflowEvent, cancellationToken).ConfigureAwait(false); } + + if (iterationResult.ConversationMessages.Length > 0) + { + if (parentConversationBuffer is not null) + { + foreach (ChatMessage message in iterationResult.ConversationMessages) + { + parentConversationBuffer.Add(message); + } + } + else if (workflowConversationId is null) + { + throw new DeclarativeActionException( + $"Parallel Foreach '{this.Id}' produced workflow-conversation messages without a workflow conversation."); + } + else + { + foreach (ChatMessage message in iterationResult.ConversationMessages) + { + await workflowOptions.AgentProvider.CreateMessageAsync( + workflowConversationId, + message, + cancellationToken).ConfigureAwait(false); + } + } + } } return default; @@ -146,7 +180,10 @@ async Task RunWorkerAsync() while (!groupCancellation.IsCancellationRequested) { int iterationIndex = Interlocked.Increment(ref nextIndex); - if (iterationIndex >= values.Length) + // Use an unsigned comparison so an exhausted allocator cannot wrap around and + // address a negative array index when an extreme item count is combined with + // multiple workers. + if ((uint)iterationIndex >= (uint)values.Length) { return; } @@ -160,6 +197,7 @@ async Task RunWorkerAsync() stateSnapshot, workflowOptions, this._executionOptions.IterationTimeout, + exception => RecordIterationFailure(iterationIndex, exception), groupCancellation.Token).ConfigureAwait(false); } catch (OperationCanceledException) when (groupCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested) @@ -168,7 +206,7 @@ async Task RunWorkerAsync() } catch (Exception exception) { - iterationFailures[iterationIndex] = exception; + RecordIterationFailure(iterationIndex, exception); groupCancellation.Cancel(); return; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/HttpRequestExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/HttpRequestExecutor.cs index d2c575faad..f4e07e4e3e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/HttpRequestExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/HttpRequestExecutor.cs @@ -39,6 +39,12 @@ internal sealed class HttpRequestExecutor( string? conversationId = this.GetConversationId(); string? connectionName = this.GetConnectionName(); + if (context.IsParallelForeachBranch() && conversationId is not null) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{this.Id}' cannot copy an HTTP response into a conversation immediately."); + } + HttpRequestInfo requestInfo = new() { Method = method, diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs index 2b0378f999..d184df0452 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs @@ -66,6 +66,12 @@ public async ValueTask CompleteAsync(IWorkflowContext context, ActionExecutorRes private async ValueTask InvokeAgentAsync(IWorkflowContext context, IEnumerable? messages, CancellationToken cancellationToken) { string? conversationId = this.GetConversationId(); + if (context.IsParallelForeachBranch() && conversationId is not null) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{this.Id}' cannot invoke an agent with an explicit conversation target."); + } + string agentName = this.GetAgentName(); bool autoSend = this.GetAutoSendValue(); Dictionary? inputParameters = this.GetStructuredInputs(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeFunctionToolExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeFunctionToolExecutor.cs index 386c3712b9..e6cf08e947 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeFunctionToolExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeFunctionToolExecutor.cs @@ -72,6 +72,12 @@ public static class Steps [SendsMessage(typeof(ExternalInputRequest))] protected override async ValueTask ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { + if (context.IsParallelForeachBranch()) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{this.Id}' cannot await external function-tool input."); + } + string functionName = this.GetFunctionName(); bool requireApproval = this.GetRequireApproval(); Dictionary? arguments = this.GetArguments(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeMcpToolExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeMcpToolExecutor.cs index 46a5cae5fd..ff276cc20c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeMcpToolExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeMcpToolExecutor.cs @@ -77,6 +77,12 @@ public static bool RequiresNothing(object? message) => [SendsMessage(typeof(ExternalInputRequest))] protected override async ValueTask ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { + if (context.IsParallelForeachBranch()) + { + throw new DeclarativeActionException( + $"Parallel Foreach action '{this.Id}' cannot execute an MCP tool because its approval and conversation effects are not stageable."); + } + string serverUrl = this.GetServerUrl(); string? serverLabel = this.GetServerLabel(); string toolName = this.GetToolName(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs index 0a4714b912..3bd0d9bea3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ParallelForeachIterationRunner.cs @@ -10,6 +10,7 @@ using Microsoft.Agents.AI.Workflows.Declarative.Kit; using Microsoft.Agents.AI.Workflows.Declarative.PowerFx; using Microsoft.Agents.ObjectModel; +using Microsoft.Extensions.AI; using Microsoft.PowerFx.Types; namespace Microsoft.Agents.AI.Workflows.Declarative.ObjectModel; @@ -17,7 +18,8 @@ namespace Microsoft.Agents.AI.Workflows.Declarative.ObjectModel; internal sealed record ParallelForeachIterationResult( int Index, WorkflowStateChange[] StateChanges, - WorkflowEvent[] Events); + WorkflowEvent[] Events, + ChatMessage[] ConversationMessages); /// /// Runs one Foreach body through the existing workflow runtime with isolated formula state. @@ -26,23 +28,66 @@ internal static class ParallelForeachIterationRunner { public static void ValidateBody(Foreach model) { - foreach (DialogAction action in model.Descendants().OfType()) + DialogAction[] bodyActions = + [ + .. model.Descendants() + .OfType() + .Where(action => BelongsToLoopBody(action, model)), + ]; + HashSet bodyActionIds = + [ + .. bodyActions + .Select(action => action.Id.Value), + ]; + + foreach (DialogAction action in bodyActions) { if (action is Question or RequestExternalInput) { - throw new DeclarativeModelException( - $"Parallel Foreach '{model.Id.Value}' cannot safely checkpoint while action " + - $"'{action.Id.Value}' ({action.GetType().Name}) is awaiting external input."); + Reject(model, action, "it can await external input and cannot be checkpointed safely"); + } + + if (action is InvokeFunctionTool or InvokeMcpTool) + { + Reject(model, action, "it can suspend for external input and cannot be checkpointed safely"); + } + + if (action is AddConversationMessage or CopyConversationMessages) + { + Reject(model, action, "it mutates a conversation immediately and cannot be staged safely"); + } + + if (action is InvokeAzureAgent { ConversationId: not null } or HttpRequestAction { ConversationId: not null }) + { + Reject(model, action, "an explicit conversation target cannot be isolated per iteration"); + } + + if (action is EndDialog or EndConversation or CancelAllDialogs or CancelDialog) + { + Reject(model, action, "it terminates or cancels workflow-wide control flow"); } if (action is BreakLoop or ContinueLoop && TargetsLoop(action, model)) { - throw new DeclarativeModelException( - $"Parallel Foreach '{model.Id.Value}' does not support {action.GetType().Name} targeting the parallel loop."); + Reject(model, action, $"{action.GetType().Name} cannot target the parallel loop"); + } + + if (action is GotoAction gotoAction) + { + string targetId = gotoAction.ActionId.Value; + if (!bodyActionIds.Contains(targetId) || TargetsDifferentParallelLoop(gotoAction, model)) + { + Reject(model, action, $"GotoAction target '{targetId}' is outside the parallel body"); + } } } } + private static void Reject(Foreach model, DialogAction action, string reason) => + throw new DeclarativeModelException( + $"Parallel Foreach '{model.Id.Value}' cannot execute action '{action.Id.Value}' " + + $"({action.GetType().Name}): {reason}."); + public static async Task RunAsync( Foreach model, FormulaValue value, @@ -50,6 +95,7 @@ public static async Task RunAsync( WorkflowStateSnapshot stateSnapshot, DeclarativeWorkflowOptions workflowOptions, TimeSpan? timeout, + Action reportFailure, CancellationToken cancellationToken) { using CancellationTokenSource timeoutSource = new(); @@ -60,6 +106,9 @@ public static async Task RunAsync( } WorkflowFormulaState branchState = WorkflowFormulaState.CreateBranch(workflowOptions.CreateRecalcEngine(), stateSnapshot); + WorkflowConversationMessageBuffer conversationMessageBuffer = new(); + branchState.ConversationMessageBuffer = conversationMessageBuffer; + branchState.ParallelFailureReporter = reportFailure; SetLoopVariable(branchState, model.Value!.Path, new PortableValue(value.AsPortable()).ToFormula()); if (model.Index is not null) { @@ -138,7 +187,7 @@ .. branchState ]; WorkflowEvent[] bufferedEvents = [.. events.Where(workflowEvent => ShouldReplay(workflowEvent, model.Id.Value))]; - return new(index, stateChanges, bufferedEvents); + return new(index, stateChanges, bufferedEvents, [.. conversationMessageBuffer.Messages]); } catch (OperationCanceledException) when (timeoutSource.IsCancellationRequested) { @@ -190,6 +239,34 @@ and not RequestInfoEvent and not ExecutorFailedEvent) && (workflowEvent is not ExecutorEvent executorEvent || executorEvent.ExecutorId != rootExecutorId); + private static bool TargetsDifferentParallelLoop(DialogAction action, Foreach loop) + { + BotElement? ancestor = action.Parent; + while (ancestor is not null && ancestor is not Foreach) + { + ancestor = ancestor.Parent; + } + return ancestor is Foreach ancestorLoop + && !ancestorLoop.Id.Equals(loop.Id) + && ForeachExecutionOptions.Parse(ancestorLoop).IsParallel; + } + + private static bool BelongsToLoopBody(DialogAction action, Foreach loop) + { + BotElement? ancestor = action.Parent; + while (ancestor is not null) + { + if (ancestor is Foreach ancestorLoop && ForeachExecutionOptions.Parse(ancestorLoop).IsParallel) + { + return ancestorLoop.Id.Equals(loop.Id); + } + + ancestor = ancestor.Parent; + } + + return false; + } + private static bool TargetsLoop(DialogAction action, Foreach loop) { BotElement? ancestor = action.Parent; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs index a39aef3911..52f29f0c87 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Frozen; using System.Collections.Generic; using System.Diagnostics; @@ -37,6 +38,19 @@ internal sealed class WorkflowFormulaState public WorkflowExpressionEngine Evaluator { get; } + /// + /// Gets the branch-local workflow-conversation staging buffer, when this state is running + /// inside a parallel Foreach iteration. + /// + internal WorkflowConversationMessageBuffer? ConversationMessageBuffer { get; set; } + + /// + /// Receives action failures while this state is executing as an isolated parallel + /// Foreach branch. The runner uses this side channel so a peer cancellation cannot + /// hide an error that was already raised by the branch action. + /// + internal Action? ParallelFailureReporter { get; set; } + public WorkflowFormulaState(RecalcEngine engine) { this._scopes = VariableScopeNames.AllScopes.ToDictionary(scopeName => GetScopeName(scopeName), _ => new WorkflowScope()); @@ -78,7 +92,7 @@ .. VariableScopeNames.AllScopes .SelectMany( scopeName => this.Keys(scopeName) - .OrderBy(variableName => variableName, System.StringComparer.Ordinal) + .OrderBy(variableName => variableName, StringComparer.Ordinal) .Select( variableName => new WorkflowStateEntry( diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md index 4a8e47d9af..b50d798bb8 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/README.md @@ -83,11 +83,21 @@ we've provided a console application that is able to execute any declarative wor ``` Each iteration receives an isolated copy of workflow state. After all iterations succeed, state writes and emitted -events are applied in collection order; writes to the same variable therefore use deterministic last-index-wins -semantics. A failure or timeout cancels outstanding iterations and does not commit buffered state or events. - -Parallel iterations cannot currently suspend for external input because an in-flight parallel set cannot be safely -represented by a parent workflow checkpoint. `Question` and `RequestExternalInput` actions are rejected when the -workflow is built. An action that conditionally requests external input fails the parallel Foreach at runtime instead -of creating an unsafe checkpoint. `BreakLoop` and `ContinueLoop` cannot target a parallel Foreach. - +events are replayed in source-index order; writes to the same variable therefore use deterministic last-index-wins +semantics. Agent responses copied to the workflow conversation are staged per iteration and replayed serially in the +same order. A failure or timeout cancels outstanding iterations and discards staged state, events, and workflow +conversation copies. Provider-side effects that happen while an iteration is running are external to this staging +boundary and are not rolled back. + +`maxParallelism` is a per-`Foreach` limit. Nested parallel loops can therefore use the product of their individual +limits; the option is not a workflow-wide concurrency budget. Iteration timeouts use cooperative cancellation, so a +provider that ignores its cancellation token can delay completion beyond the configured duration. + +Parallel iterations cannot suspend for external input because an in-flight parallel set cannot be safely represented +by a parent workflow checkpoint. `Question`, `RequestExternalInput`, `InvokeFunctionTool`, and `InvokeMcpTool` are +rejected when the workflow is built. Direct conversation mutations (`AddConversationMessage` and +`CopyConversationMessages`), explicit conversation targets on `InvokeAzureAgent`, and HTTP response copies to a +conversation are also rejected in a parallel body. An agent response that conditionally requests external input is +rejected at runtime instead of creating an unsafe checkpoint. Global termination/cancellation actions and +cross-body `GotoAction` targets are rejected as well. `BreakLoop` and `ContinueLoop` cannot target a parallel +`Foreach`, although they remain available for nested sequential loops. diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs index 54f2ed0b69..16df06bc82 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ParallelForeachWorkflowTests.cs @@ -61,6 +61,29 @@ public async Task ParallelForeachExecutesConcurrentlyAndHonorsLimitAsync() Assert.Equal(2, provider.PeakConcurrency); } + [Fact] + public async Task ParallelForeachWithOneWorkerRemainsSerializedAsync() + { + // Arrange + ControlledAgentProvider provider = new(delayByIndex: index => TimeSpan.FromMilliseconds((2 - index) * 25)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 1 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:a", "1:b", "2:c"], GetAgentResponses(observation)); + Assert.Equal([0, 1, 2], provider.Completions); + Assert.Equal(1, provider.PeakConcurrency); + } + [Fact] public async Task ParallelForeachUsesBoundedDefaultLimitAsync() { @@ -104,6 +127,30 @@ public async Task ParallelForeachCapsWorkerCountAtItemCountForExtremeLimitAsync( Assert.Equal(2, provider.PeakConcurrency); } + [Fact] + public async Task ParallelForeachPreservesBlankAndExtremeItemValuesAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 4, barrierTimeout: s_barrierTimeout); + string yaml = CreateWorkflowYaml( + items: "=[Blank(), -2147483648, 0, 2147483647]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(4, provider.InvocationCount); + Assert.Equal( + ["0:", "1:-2147483648", "2:0", "3:2147483647"], + GetAgentResponses(observation)); + } + [Fact] public async Task ParallelForeachKeepsDuplicateValuesSeparatedByIndexAsync() { @@ -167,6 +214,167 @@ public async Task ParallelForeachIsolatesStateAndCommitsInSourceOrderAsync() provider.Invocations.OrderBy(invocation => invocation.Index).Select(invocation => $"{invocation.Index}:{invocation.Value}")); } + [Fact] + public async Task ParallelForeachStagesWorkflowConversationWritesInSourceOrderAsync() + { + // Arrange + ControlledAgentProvider provider = new( + barrierParticipants: 4, + barrierTimeout: s_barrierTimeout, + delayByIndex: index => TimeSpan.FromMilliseconds((3 - index) * 100), + conversationWriteDelay: TimeSpan.FromMilliseconds(25)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(s_orderedResponses, provider.WorkflowConversationWrites); + Assert.Equal(1, provider.PeakConversationWriteConcurrency); + Assert.Equal(provider.InvocationCount, provider.InvocationCompletionsAtFirstConversationWrite); + } + + [Fact] + public async Task ParallelForeachDoesNotWriteWorkflowConversationWhenAnIterationFailsAsync() + { + // Arrange + ControlledAgentProvider provider = new( + barrierParticipants: 4, + barrierTimeout: s_barrierTimeout, + delayByIndex: index => TimeSpan.FromMilliseconds((3 - index) * 50), + failureIndexes: [2], + conversationWriteDelay: TimeSpan.FromMilliseconds(10)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertWorkflowError(observation); + Assert.Empty(provider.WorkflowConversationWrites); + } + + [Fact] + public async Task ParallelForeachConversationReplayCannotMutateBufferedEventsAsync() + { + // Arrange + ControlledAgentProvider provider = new(mutateConversationWrites: true); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:a", "1:b"], GetAgentResponses(observation)); + Assert.Equal(["0:a", "1:b"], provider.WorkflowConversationWrites); + } + + [Fact] + public async Task NestedParallelForeachKeepsConversationWritesStagedUntilTheOuterCommitAsync() + { + // Arrange + ControlledAgentProvider provider = new( + barrierParticipants: 2, + barrierTimeout: s_barrierTimeout, + delayByIndex: index => TimeSpan.FromMilliseconds((1 - index) * 50), + conversationWriteDelay: TimeSpan.FromMilliseconds(15)); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: Foreach + id: inner_parallel_loop + items: =[1, 2] + value: Local.InnerItem + index: Local.InnerIndex + mode: Parallel + maxParallelism: 2 + actions: + - kind: InvokeAzureAgent + id: nested_invoke_agent + agent: + name: TestAgent + input: + arguments: + value: =Concatenate(Text(Local.Index), ":", Text(Local.InnerIndex), ":", Local.Item) + index: =Local.InnerIndex + output: + autoSend: true + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:0:0:a", "1:0:1:a", "0:1:0:b", "1:1:1:b"], provider.WorkflowConversationWrites); + Assert.Equal(1, provider.PeakConversationWriteConcurrency); + } + + [Fact] + public async Task NestedParallelForeachHonorsTheProductOfLimitsAsync() + { + // Arrange + ControlledAgentProvider provider = new(barrierParticipants: 4, barrierTimeout: s_barrierTimeout); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: Foreach + id: inner_parallel_loop + items: =[1, 2] + value: Local.InnerItem + index: Local.InnerIndex + mode: Parallel + maxParallelism: 2 + actions: + - kind: InvokeAzureAgent + id: nested_invoke_agent + agent: + name: TestAgent + input: + arguments: + value: =Local.Item + index: =Local.InnerIndex + output: + autoSend: false + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(4, provider.InvocationCount); + Assert.Equal(4, provider.PeakConcurrency); + } + [Fact] public async Task ParallelForeachDeepCopiesComplexLoopValuesAsync() { @@ -250,6 +458,40 @@ public async Task ParallelForeachAggregatesBranchFailureAsync() Assert.DoesNotContain(observation.Events.OfType(), evt => evt.Message.Trim().StartsWith("Branch ", StringComparison.Ordinal)); } + [Fact] + public async Task ParallelForeachRetainsSimultaneousBranchFailuresInSourceOrderAsync() + { + // Arrange + ControlledAgentProvider provider = new( + barrierParticipants: 4, + barrierTimeout: s_barrierTimeout, + failureIndexes: [1, 3], + failureBarrierParticipants: 2); + string yaml = CreateWorkflowYaml( + items: "=[\"a\", \"b\", \"c\", \"d\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 4 + """, + bodyActions: InvokeAgentAction); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + Exception error = AssertWorkflowError(observation); + AggregateException aggregate = Assert.Single( + Flatten(error).OfType(), + exception => exception.Message.StartsWith("Parallel Foreach 'parallel_loop' failed.", StringComparison.Ordinal)); + string[] iterationFailures = [.. aggregate.InnerExceptions.Select(exception => exception.Message)]; + Assert.Equal( + [ + "Parallel Foreach 'parallel_loop' iteration 1 failed.", + "Parallel Foreach 'parallel_loop' iteration 3 failed.", + ], + iterationFailures); + } + [Fact] public async Task ParallelForeachFailureCancelsActivePeersAsync() { @@ -559,6 +801,199 @@ public void ParallelForeachRejectsCheckpointingBody() Assert.Contains("checkpoint", exception.Message, StringComparison.OrdinalIgnoreCase); } + [Fact] + public void ParallelForeachRejectsConversationTermination() + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: EndConversation + id: end_conversation_in_parallel + """); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("terminates", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public void ParallelForeachRejectsGotoOutsideBody() + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: GotoAction + id: goto_after_parallel + actionId: after_parallel + """, + afterActions: """ + - kind: SendActivity + id: after_parallel + activity: after + """); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("GotoAction", exception.Message, StringComparison.Ordinal); + Assert.Contains("parallel", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public async Task ParallelForeachAllowsGotoWithinBodyAsync() + { + // Arrange + ControlledAgentProvider provider = new(); + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: $$""" + - kind: GotoAction + id: goto_agent + actionId: invoke_agent + + - kind: SendActivity + id: skipped_activity + activity: never + + {{InvokeAgentAction}} + """); + + // Act + WorkflowObservation observation = await RunWorkflowAsync(yaml, provider); + + // Assert + AssertNoWorkflowError(observation); + Assert.Equal(["0:a"], GetAgentResponses(observation)); + Assert.DoesNotContain(observation.Events.OfType(), evt => evt.Message.Trim() == "never"); + } + + [Fact] + public void ParallelForeachRejectsFunctionToolCheckpoint() + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: InvokeFunctionTool + id: invoke_function_tool + functionName: TestFunction + requireApproval: false + """); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("external input", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Theory] + [InlineData("EndWorkflow", "terminates")] + [InlineData("AddConversationMessage", "conversation")] + [InlineData("CopyConversationMessages", "conversation")] + [InlineData("InvokeMcpTool", "external input")] + [InlineData("HttpRequestAction", "conversation")] + public void ParallelForeachRejectsUnsafeBodyActions(string actionKind, string expectedMessagePart) + { + // Arrange + string action = actionKind switch + { + "EndWorkflow" => """ + - kind: EndWorkflow + id: end_workflow_in_parallel + """, + "AddConversationMessage" => """ + - kind: AddConversationMessage + id: add_message_in_parallel + message: Local.Message + role: User + conversationId: =System.ConversationId + content: + - type: Text + value: parallel + """, + "CopyConversationMessages" => """ + - kind: CopyConversationMessages + id: copy_messages_in_parallel + conversationId: =System.ConversationId + messages: =[UserMessage("parallel")] + """, + "InvokeMcpTool" => """ + - kind: InvokeMcpTool + id: invoke_mcp_in_parallel + serverUrl: https://example.test/mcp + toolName: test + """, + "HttpRequestAction" => """ + - kind: HttpRequestAction + id: http_in_parallel + method: GET + url: https://example.test + conversationId: =System.ConversationId + """, + _ => throw new ArgumentOutOfRangeException(nameof(actionKind), actionKind, null), + }; + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: action); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains(expectedMessagePart, exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public void ParallelForeachRejectsExplicitAgentConversationTarget() + { + // Arrange + string yaml = CreateWorkflowYaml( + items: "=[\"a\"]", + executionOptions: """ + mode: Parallel + maxParallelism: 2 + """, + bodyActions: """ + - kind: InvokeAzureAgent + id: explicit_conversation_agent + conversationId: =System.ConversationId + agent: + name: TestAgent + """); + + // Act + DeclarativeModelException exception = Assert.Throws(() => BuildWorkflow(yaml, new ControlledAgentProvider())); + + // Assert + Assert.Contains("conversation target", exception.Message, StringComparison.OrdinalIgnoreCase); + } + [Theory] [InlineData("BreakLoop")] [InlineData("ContinueLoop")] @@ -765,18 +1200,27 @@ private sealed class ControlledAgentProvider : ResponseAgentProvider private readonly TaskCompletionSource _barrier = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _cancellationObserved = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _firstInvocationStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _failureBarrier = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _noActiveInvocations = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly int _barrierParticipants; private readonly TimeSpan _barrierTimeout; private readonly Func? _delayByIndex; private readonly HashSet _failureIndexes; + private readonly int _failureBarrierParticipants; private readonly HashSet _waitUntilCanceledIndexes; private readonly bool _waitUntilCanceled; private readonly bool _requestApproval; + private readonly TimeSpan _conversationWriteDelay; + private readonly bool _mutateConversationWrites; private int _activeCount; + private int _activeConversationWrites; private int _barrierArrivals; + private int _completedInvocations; + private int _failureBarrierArrivals; + private int _invocationCompletionsAtFirstConversationWrite = -1; private int _invocationCount; private int _peakConcurrency; + private int _peakConversationWriteConcurrency; public ControlledAgentProvider( int barrierParticipants = 1, @@ -785,38 +1229,87 @@ public ControlledAgentProvider( IEnumerable? failureIndexes = null, IEnumerable? waitUntilCanceledIndexes = null, bool waitUntilCanceled = false, - bool requestApproval = false) + bool requestApproval = false, + TimeSpan? conversationWriteDelay = null, + int failureBarrierParticipants = 0, + bool mutateConversationWrites = false) { this._barrierParticipants = barrierParticipants; this._barrierTimeout = barrierTimeout ?? TimeSpan.Zero; this._delayByIndex = delayByIndex; this._failureIndexes = failureIndexes is null ? [] : [.. failureIndexes]; + this._failureBarrierParticipants = failureBarrierParticipants; this._waitUntilCanceledIndexes = waitUntilCanceledIndexes is null ? [] : [.. waitUntilCanceledIndexes]; this._waitUntilCanceled = waitUntilCanceled; this._requestApproval = requestApproval; + this._conversationWriteDelay = conversationWriteDelay ?? TimeSpan.Zero; + this._mutateConversationWrites = mutateConversationWrites; } public int InvocationCount => Volatile.Read(ref this._invocationCount); public int PeakConcurrency => Volatile.Read(ref this._peakConcurrency); + public int PeakConversationWriteConcurrency => Volatile.Read(ref this._peakConversationWriteConcurrency); + + public int InvocationCompletionsAtFirstConversationWrite => Volatile.Read(ref this._invocationCompletionsAtFirstConversationWrite); + public int ActiveCount => Volatile.Read(ref this._activeCount); public ConcurrentBag Invocations { get; } = []; public ConcurrentQueue Completions { get; } = []; + public ConcurrentQueue WorkflowConversationWrites { get; } = []; + + public string? WorkflowConversationId { get; private set; } + public Task CancellationObserved => this._cancellationObserved.Task; public Task FirstInvocationStarted => this._firstInvocationStarted.Task; public Task NoActiveInvocations => this._noActiveInvocations.Task; - public override Task CreateConversationAsync(CancellationToken cancellationToken = default) => - Task.FromResult(Guid.NewGuid().ToString("N")); + public override Task CreateConversationAsync(CancellationToken cancellationToken = default) + { + this.WorkflowConversationId = Guid.NewGuid().ToString("N"); + return Task.FromResult(this.WorkflowConversationId); + } - public override Task CreateMessageAsync(string conversationId, ChatMessage conversationMessage, CancellationToken cancellationToken = default) => - Task.FromResult(conversationMessage); + public override async Task CreateMessageAsync(string conversationId, ChatMessage conversationMessage, CancellationToken cancellationToken = default) + { + if (string.Equals(conversationId, this.WorkflowConversationId, StringComparison.Ordinal) + && conversationMessage.Text.IndexOf(':') >= 0) + { + int active = Interlocked.Increment(ref this._activeConversationWrites); + UpdatePeak(ref this._peakConversationWriteConcurrency, active); + Interlocked.CompareExchange( + ref this._invocationCompletionsAtFirstConversationWrite, + Volatile.Read(ref this._completedInvocations), + -1); + try + { + if (this._conversationWriteDelay > TimeSpan.Zero) + { + await Task.Delay(this._conversationWriteDelay, cancellationToken).ConfigureAwait(false); + } + + string originalText = conversationMessage.Text; + this.WorkflowConversationWrites.Enqueue(originalText); + if (this._mutateConversationWrites) + { + conversationMessage.Contents.Clear(); + conversationMessage.Contents.Add(new TextContent("mutated")); + } + } + finally + { + Interlocked.Decrement(ref this._activeConversationWrites); + } + } + + return conversationMessage; + } public override Task GetMessageAsync(string conversationId, string messageId, CancellationToken cancellationToken = default) => throw new NotSupportedException(); @@ -864,10 +1357,22 @@ public override async IAsyncEnumerable InvokeAgentAsync( if (this._failureIndexes.Contains(index)) { + if (this._failureBarrierParticipants > 0) + { + int arrivals = Interlocked.Increment(ref this._failureBarrierArrivals); + if (arrivals == this._failureBarrierParticipants) + { + this._failureBarrier.TrySetResult(true); + } + + await this._failureBarrier.Task.ConfigureAwait(false); + } + throw new InvalidOperationException($"Failure for iteration {index}."); } this.Completions.Enqueue(index); + Interlocked.Increment(ref this._completedInvocations); if (this._requestApproval) {