.NET: Fix Concurrency Support for Orchestrations

dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs

@@ -4,6 +4,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading.Tasks;
 using Microsoft.Agents.AI.Workflows.Specialized;
 using Microsoft.Extensions.AI;
 using Microsoft.Shared.Diagnostics;
@@ -136,7 +137,12 @@ private static Workflow BuildConcurrentCore(
         // each agent's accumulator to it. If no aggregation function was provided, we default to returning
         // the last message from each agent
         aggregator ??= static lists => (from list in lists where list.Count > 0 select list.Last()).ToList();
-        ConcurrentEndExecutor end = new(agentExecutors.Length, aggregator);
+
+        Func<string, string, ValueTask<ConcurrentEndExecutor>> endFactory =
+            (string _, string __) => new(new ConcurrentEndExecutor(agentExecutors.Length, aggregator));
+
+        ExecutorIsh end = endFactory.ConfigureFactory(ConcurrentEndExecutor.ExecutorId);
+
         builder.AddFanInEdge(end, sources: accumulators);
 
         builder = builder.WithOutputFrom(end);

dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorRegistration.cs

@@ -13,24 +13,29 @@ internal sealed class ExecutorRegistration(string id, Type executorType, Executo
     public string Id { get; } = Throw.IfNullOrEmpty(id);
     public Type ExecutorType { get; } = Throw.IfNull(executorType);
     private ExecutorFactoryF ProviderAsync { get; } = Throw.IfNull(provider);
-    public bool IsNotExecutorInstance { get; } = rawData is not Executor;
-    public bool IsUnresettableSharedInstance { get; } = rawData is Executor executor &&
-                                                        // Cross-Run Shareable executors are "trivially" resettable, since they
-                                                        // have no on-object state.
-                                                        !executor.IsCrossRunShareable &&
-                                                        rawData is not IResettableExecutor;
-    public bool SupportsConcurrent { get; } = (rawData is not Executor executor || executor.IsCrossRunShareable) &&
-                                              (rawData is not Workflow workflow || workflow.AllowConcurrent);
+
+    public bool IsSharedInstance { get; } = rawData is Executor;
+
+    /// <summary>
+    /// Gets a value whether instances of the executor created from this registration can be reset between subsequent
+    /// runs from the same <see cref="Workflow"/> instance. This value is not relevant for executors that <see
+    /// cref="SupportsConcurrent"/>.
+    /// </summary>
+    public bool SupportsResetting { get; } = rawData is Executor &&
+                                             // Cross-Run Shareable executors are "trivially" resettable, since they
+                                             // have no on-object state.
+                                             rawData is IResettableExecutor;
+
+    /// <summary>
+    /// Gets a value whether instances of the executor created from this registration can be used in concurrent runs
+    /// from the same <see cref="Workflow"/> instance.
+    /// </summary>
+    public bool SupportsConcurrent { get; } = rawData is not Executor executor || executor.IsCrossRunShareable;
 
     internal async ValueTask<bool> TryResetAsync()
     {
-        if (this.IsUnresettableSharedInstance)
-        {
-            return false;
-        }
-
-        // If the executor supports concurrent use, then resetting is a no-op.
-        if (this.SupportsConcurrent)
+        // Non-shared instances do not need resetting
+        if (!this.IsSharedInstance)
         {
             return true;
         }

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/CollectChatMessagesExecutor.cs

@@ -11,7 +11,7 @@ namespace Microsoft.Agents.AI.Workflows.Specialized;
 /// Provides an executor that batches received chat messages that it then releases when
 /// receiving a <see cref="TurnToken"/>.
 /// </summary>
-internal sealed class CollectChatMessagesExecutor(string id) : ChatProtocolExecutor(id), IResettableExecutor
+internal sealed class CollectChatMessagesExecutor(string id) : ChatProtocolExecutor(id, declareCrossRunShareable: true), IResettableExecutor
 {
     /// <inheritdoc/>
     protected override ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default)

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/ConcurrentEndExecutor.cs

@@ -14,12 +14,14 @@ namespace Microsoft.Agents.AI.Workflows.Specialized;
 /// </summary>
 internal sealed class ConcurrentEndExecutor : Executor, IResettableExecutor
 {
+    public const string ExecutorId = "ConcurrentEnd";
+
     private readonly int _expectedInputs;
     private readonly Func<IList<List<ChatMessage>>, List<ChatMessage>> _aggregator;
     private List<List<ChatMessage>> _allResults;
     private int _remaining;
 
-    public ConcurrentEndExecutor(int expectedInputs, Func<IList<List<ChatMessage>>, List<ChatMessage>> aggregator) : base("ConcurrentEnd")
+    public ConcurrentEndExecutor(int expectedInputs, Func<IList<List<ChatMessage>>, List<ChatMessage>> aggregator) : base(ExecutorId)
     {
         this._expectedInputs = expectedInputs;
         this._aggregator = Throw.IfNull(aggregator);

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffAgentExecutor.cs

@@ -14,7 +14,7 @@ namespace Microsoft.Agents.AI.Workflows.Specialized;
 /// <summary>Executor used to represent an agent in a handoffs workflow, responding to <see cref="HandoffState"/> events.</summary>
 internal sealed class HandoffAgentExecutor(
     AIAgent agent,
-    string? handoffInstructions) : Executor(agent.GetDescriptiveId()), IResettableExecutor
+    string? handoffInstructions) : Executor(agent.GetDescriptiveId(), declareCrossRunShareable: true), IResettableExecutor
 {
     private static readonly JsonElement s_handoffSchema = AIFunctionFactory.Create(
         ([Description("The reason for the handoff")] string? reasonForHandoff) => { }).JsonSchema;
@@ -70,9 +70,9 @@ protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
             List<ChatMessage>? roleChanges = allMessages.ChangeAssistantToUserForOtherParticipants(this._agent.DisplayName);
 
             await foreach (var update in this._agent.RunStreamingAsync(allMessages,
-                                                                           options: this._agentOptions,
-                                                                           cancellationToken: cancellationToken)
-                                                        .ConfigureAwait(false))
+                                                                       options: this._agentOptions,
+                                                                       cancellationToken: cancellationToken)
+                                                     .ConfigureAwait(false))
             {
                 await AddUpdateAsync(update, cancellationToken).ConfigureAwait(false);
 

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffsEndExecutor.cs

@@ -5,8 +5,10 @@
 namespace Microsoft.Agents.AI.Workflows.Specialized;
 
 /// <summary>Executor used at the end of a handoff workflow to raise a final completed event.</summary>
-internal sealed class HandoffsEndExecutor() : Executor("HandoffEnd"), IResettableExecutor
+internal sealed class HandoffsEndExecutor() : Executor(ExecutorId, declareCrossRunShareable: true), IResettableExecutor
 {
+    public const string ExecutorId = "HandoffEnd";
+
     protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
         routeBuilder.AddHandler<HandoffState>((handoff, context, cancellationToken) =>
             context.YieldOutputAsync(handoff.Messages, cancellationToken));

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffsStartExecutor.cs

@@ -8,8 +8,10 @@
 namespace Microsoft.Agents.AI.Workflows.Specialized;
 
 /// <summary>Executor used at the start of a handoffs workflow to accumulate messages and emit them as HandoffState upon receiving a turn token.</summary>
-internal sealed class HandoffsStartExecutor() : ChatProtocolExecutor("HandoffStart", DefaultOptions), IResettableExecutor
+internal sealed class HandoffsStartExecutor() : ChatProtocolExecutor(ExecutorId, DefaultOptions, declareCrossRunShareable: true), IResettableExecutor
 {
+    internal const string ExecutorId = "HandoffStart";
+
     private static ChatProtocolExecutorOptions DefaultOptions => new()
     {
         StringMessageChatRole = ChatRole.User

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/OutputMessagesExecutor.cs

@@ -13,7 +13,7 @@ public static partial class AgentWorkflowBuilder
     /// Provides an executor that batches received chat messages that it then publishes as the final result
     /// when receiving a <see cref="TurnToken"/>.
     /// </summary>
-    internal sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages"), IResettableExecutor
+    internal sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages", declareCrossRunShareable: true), IResettableExecutor
     {
         protected override ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default)
             => context.YieldOutputAsync(messages, cancellationToken);

dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs

@@ -86,14 +86,14 @@ internal Workflow(string startExecutorId, string? name = null, string? descripti
     }
 
     private bool _needsReset;
-    private bool IsResettable => this.Registrations.Values.All(registration => !registration.IsUnresettableSharedInstance);
-
+    private bool HasResettable => this.Registrations.Values.Any(registration => registration.SupportsResetting);
     private async ValueTask<bool> TryResetExecutorRegistrationsAsync()
     {
-        if (this.IsResettable)
+        if (this.HasResettable)
         {
             foreach (ExecutorRegistration registration in this.Registrations.Values)
             {
+                // TryResetAsync returns true if the executor does not need resetting
                 if (!await registration.TryResetAsync().ConfigureAwait(false))
                 {
                     return false;
@@ -158,7 +158,7 @@ internal void TakeOwnership(object ownerToken, bool subworkflow = false, object?
                 });
         }
 
-        this._needsReset = true;
+        this._needsReset = this.HasResettable;
         this._ownedAsSubworkflow = subworkflow;
     }
 

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs

@@ -382,11 +382,12 @@ public async Task BuildGroupChat_AgentsRunInOrderAsync(int maxIterations)
     }
 
     private static async Task<(string UpdateText, List<ChatMessage>? Result)> RunWorkflowAsync(
-        Workflow workflow, List<ChatMessage> input)
+        Workflow workflow, List<ChatMessage> input, ExecutionEnvironment executionEnvironment = ExecutionEnvironment.InProcess_Lockstep)
     {
         StringBuilder sb = new();
 
-        await using StreamingRun run = await InProcessExecution.Lockstep.StreamAsync(workflow, input);
+        IWorkflowExecutionEnvironment environment = executionEnvironment.ToWorkflowExecutionEnvironment();
+        await using StreamingRun run = await environment.StreamAsync(workflow, input);
         await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
 
         WorkflowOutputEvent? output = null;

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs

@@ -6,20 +6,23 @@
 using System.IO;
 using System.Linq;
 using System.Runtime.CompilerServices;
-using System.Text;
 using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Agents.AI.Workflows.UnitTests;
 using Microsoft.Extensions.AI;
 
 namespace Microsoft.Agents.AI.Workflows.Sample;
 
 internal static class Step6EntryPoint
 {
+    public const string EchoAgentId = "echo";
+    public const string EchoPrefix = "You said: ";
+
     public static Workflow CreateWorkflow(int maxTurns) =>
         AgentWorkflowBuilder
             .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = maxTurns })
-            .AddParticipants(new HelloAgent(), new EchoAgent())
+            .AddParticipants(new HelloAgent(), new TestEchoAgent(id: EchoAgentId, prefix: EchoPrefix))
             .Build();
 
     public static async ValueTask RunAsync(TextWriter writer, IWorkflowExecutionEnvironment environment, int maxSteps = 2)
@@ -85,54 +88,3 @@ public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync
 }
 
 internal sealed class HelloAgentThread() : InMemoryAgentThread();
-
-internal sealed class EchoAgent(string id = nameof(EchoAgent)) : AIAgent
-{
-    public const string Prefix = "You said: ";
-    public const string DefaultId = nameof(EchoAgent);
-
-    public override string Id => id;
-    public override string? Name => id;
-
-    public override AgentThread GetNewThread()
-        => new EchoAgentThread();
-
-    public override AgentThread DeserializeThread(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null)
-        => new EchoAgentThread();
-
-    public override async Task<AgentRunResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
-    {
-        IEnumerable<AgentRunResponseUpdate> update = [
-            await this.RunStreamingAsync(messages, thread, options, cancellationToken)
-                      .SingleAsync(cancellationToken)
-                      .ConfigureAwait(false)];
-
-        return update.ToAgentRunResponse();
-    }
-
-    public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
-    {
-        var messagesList = messages as IReadOnlyCollection<ChatMessage> ?? messages.ToList();
-
-        if (messagesList.Count == 0)
-        {
-            throw new ArgumentException("No messages provided to echo.", nameof(messages));
-        }
-
-        StringBuilder collectedText = new(Prefix);
-        foreach (string messageText in messagesList.Select(message => message.Text)
-                                               .Where(text => !string.IsNullOrEmpty(text)))
-        {
-            collectedText.AppendLine(messageText);
-        }
-
-        yield return new(ChatRole.Assistant, collectedText.ToString())
-        {
-            AgentId = this.Id,
-            AuthorName = this.Name,
-            MessageId = Guid.NewGuid().ToString("N"),
-        };
-    }
-}
-
-internal sealed class EchoAgentThread() : InMemoryAgentThread();

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs

@@ -8,7 +8,10 @@ namespace Microsoft.Agents.AI.Workflows.Sample;
 
 internal static class Step7EntryPoint
 {
-    public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2)
+    public static string EchoAgentId => Step6EntryPoint.EchoAgentId;
+    public static string EchoPrefix => Step6EntryPoint.EchoPrefix;
+
+    public static async ValueTask RunAsync(TextWriter writer, IWorkflowExecutionEnvironment environment, int maxSteps = 2)
     {
         Workflow workflow = Step6EntryPoint.CreateWorkflow(maxSteps);
 

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/10_Sequential_HostAsAgent.cs

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.IO;
+using System.Threading.Tasks;
+using Microsoft.Agents.AI.Workflows.UnitTests;
+using Microsoft.Extensions.AI;
+
+namespace Microsoft.Agents.AI.Workflows.Sample;
+
+internal static class Step10EntryPoint
+{
+    public static Workflow CreateWorkflow()
+    {
+        TestEchoAgent echoAgent = new("echo", "Echo");
+        return AgentWorkflowBuilder.BuildSequential(echoAgent);
+    }
+    public static Workflow WorkflowInstance => CreateWorkflow();
+
+    public static async ValueTask RunAsync(TextWriter writer, IWorkflowExecutionEnvironment executionEnvironment, IEnumerable<string> inputs)
+    {
+        AIAgent hostAgent = WorkflowInstance.AsAgent("echo-workflow", "EchoW", executionEnvironment: executionEnvironment);
+
+        AgentThread thread = hostAgent.GetNewThread();
+        foreach (string input in inputs)
+        {
+            AgentRunResponse response;
+            object? continuationToken = null;
+            do
+            {
+                response = await hostAgent.RunAsync(input, thread, new AgentRunOptions { ContinuationToken = continuationToken });
+            } while ((continuationToken = response.ContinuationToken) is { });
+
+            foreach (ChatMessage message in response.Messages)
+            {
+                writer.WriteLine($"{message.AuthorName}: {message.Text}");
+            }
+        }
+    }
+}

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/11_Concurrent_HostAsAgent.cs

@@ -0,0 +1,52 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Agents.AI.Workflows.UnitTests;
+using Microsoft.Extensions.AI;
+
+namespace Microsoft.Agents.AI.Workflows.Sample;
+
+internal static class Step11EntryPoint
+{
+    public const int AgentCount = 2;
+
+    public const string EchoAgentIdPrefix = "echo-";
+    public const string EchoAgentNamePrefix = "Echo";
+
+    public static string ExpectedOutputForInput(string input, int agentNumber)
+        => $"{EchoAgentNamePrefix}{agentNumber}: {input}";
+
+    public static Workflow CreateWorkflow()
+    {
+        TestEchoAgent[] echoAgents = Enumerable.Range(1, AgentCount)
+            .Select(i => new TestEchoAgent($"{EchoAgentIdPrefix}{i}", $"{EchoAgentNamePrefix}{i}"))
+            .ToArray();
+
+        return AgentWorkflowBuilder.BuildConcurrent(echoAgents);
+    }
+    public static Workflow WorkflowInstance => CreateWorkflow();
+
+    public static async ValueTask RunAsync(TextWriter writer, IWorkflowExecutionEnvironment executionEnvironment, IEnumerable<string> inputs)
+    {
+        AIAgent hostAgent = WorkflowInstance.AsAgent("echo-workflow", "EchoW", executionEnvironment: executionEnvironment);
+
+        AgentThread thread = hostAgent.GetNewThread();
+        foreach (string input in inputs)
+        {
+            AgentRunResponse response;
+            object? continuationToken = null;
+            do
+            {
+                response = await hostAgent.RunAsync(input, thread, new AgentRunOptions { ContinuationToken = continuationToken });
+            } while ((continuationToken = response.ContinuationToken) is { });
+
+            foreach (ChatMessage message in response.Messages)
+            {
+                writer.WriteLine($"{message.AuthorName}: {message.Text}");
+            }
+        }
+    }
+}