diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs
index 82bb8a419f..4d606dd45a 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs
@@ -1,10 +1,13 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
+
namespace Microsoft.Agents.AI.DurableTask;
///
/// Provides configuration options for durable agents and workflows.
///
+[DebuggerDisplay("Workflows = {Workflows.Workflows.Count}, Agents = {Agents.AgentCount}")]
public sealed class DurableOptions
{
///
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRequestInfoEvent.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRequestInfoEvent.cs
index d816f41d78..727318463b 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRequestInfoEvent.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRequestInfoEvent.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Microsoft.Agents.AI.Workflows;
@@ -14,6 +15,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// The full type name of the request type.
/// The full type name of the expected response type.
/// The request port definition, if available.
+[DebuggerDisplay("RequestPort = {RequestPortId}")]
public sealed class DurableRequestInfoEvent(
string RequestPortId,
string Input,
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs
index f928fd8958..3cf92c3d29 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Microsoft.Agents.AI.Workflows;
@@ -14,6 +15,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// This class provides a similar API to but for workflows executed as durable orchestrations.
/// Events are received by raising external events to the orchestration and can be streamed to the caller.
///
+[DebuggerDisplay("{WorkflowName} ({InstanceId})")]
public sealed class DurableRun : IRun
{
private readonly DurableTaskClient _client;
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs
index 078c2a1904..a67beaf75c 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Text.Json;
@@ -16,6 +17,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// Events are detected by monitoring the orchestration status for executors that are waiting
/// for external input (human-in-the-loop scenarios).
///
+[DebuggerDisplay("{WorkflowName} ({InstanceId})")]
public sealed class DurableStreamingRun : IStreamingRun
{
private readonly DurableTaskClient _client;
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowCompletedEvent.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowCompletedEvent.cs
index 6a676da20f..64b0992e46 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowCompletedEvent.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowCompletedEvent.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -7,6 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask;
///
/// Event raised when a durable workflow completes successfully.
///
+[DebuggerDisplay("Completed: {Result}")]
public sealed class DurableWorkflowCompletedEvent : WorkflowEvent
{
///
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowFailedEvent.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowFailedEvent.cs
index e68b16eae8..787ce68419 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowFailedEvent.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowFailedEvent.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -7,6 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask;
///
/// Event raised when a durable workflow fails.
///
+[DebuggerDisplay("Failed: {ErrorMessage}")]
public sealed class DurableWorkflowFailedEvent : WorkflowEvent
{
///
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs
index 4a72aed3c5..e71cdbaa3a 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -7,6 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask;
///
/// Provides configuration options for managing durable workflows within an application.
///
+[DebuggerDisplay("Workflows = {Workflows.Count}")]
public sealed class DurableWorkflowOptions
{
private readonly Dictionary _workflows = new(StringComparer.OrdinalIgnoreCase);
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs
index 247b87c27b..832407f024 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs
@@ -366,28 +366,13 @@ internal class DurableWorkflowRunner
/// - Maximum superstep limit is reached (safety limit)
///
///
- /// Message Routing Rules:
+ /// Fan-In Handling:
+ ///
+ ///
+ /// For executors with multiple predecessors (Fan-In), the implementation waits for all predecessor
+ /// results before invoking the executor. If the executor accepts an array type, all messages are
+ /// aggregated into a JSON array.
///
- ///
- /// -
- ///
- /// Messages sent via SendMessageAsync take priority and include explicit type information.
- /// This is the primary mechanism for void-returning executors.
- ///
- ///
- /// -
- ///
- /// If no messages were sent explicitly, the executor's return value is routed to successors.
- /// The type information comes from .
- ///
- ///
- /// -
- ///
- /// Edge conditions are evaluated before routing. If a condition returns false, the message
- /// is not forwarded to that particular successor.
- ///
- ///
- ///
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
@@ -441,54 +426,93 @@ internal class DurableWorkflowRunner
activeExecutors.Count,
string.Join(", ", activeExecutors));
- // Process each active executor
+ // Prepare execution tasks for all active executors (for parallel dispatch)
+ List<(string ExecutorId, string Input, string? InputTypeName, WorkflowExecutorInfo Info)> executorInputs = [];
+
foreach (string executorId in activeExecutors)
{
Queue<(string Message, string? InputTypeName)> queue = messageQueues[executorId];
- // Process all messages for this executor in this superstep
- while (queue.Count > 0)
+ // Check if this is a Fan-In executor that expects aggregated input
+ bool isFanIn = plan.Predecessors.TryGetValue(executorId, out List? predecessors) && predecessors.Count > 1;
+
+ string input;
+ string? inputTypeName;
+
+ if (isFanIn && queue.Count > 1)
{
- (string input, string? inputTypeName) = queue.Dequeue();
-
- // Create executor info
- WorkflowExecutorInfo executorInfo = CreateExecutorInfo(executorId, executorBindings);
-
- // Execute the activity with type information
- string rawResult = await this.DispatchExecutorAsync(
- context, executorInfo, input, inputTypeName, logger, customStatus, sharedState).ConfigureAwait(true);
-
- (string result, List sentMessages) = UnwrapActivityResult(rawResult, customStatus, sharedState);
- lastResults[executorId] = result;
-
- // Check for explicit halt request (via RequestHaltAsync)
- if (HasHaltBeenRequested(customStatus, executorId))
+ // Fan-In: Aggregate all messages into a JSON array
+ List messages = [];
+ while (queue.Count > 0)
{
- haltRequested = true;
- finalOutput = result;
- logger.LogDebug("Halt requested by executor {ExecutorId}", executorId);
- break;
+ (string msg, _) = queue.Dequeue();
+ messages.Add(msg);
}
- // Route messages sent via SendMessageAsync (takes priority for void-returning executors)
- if (sentMessages.Count > 0)
+ input = AggregateMessagesToJsonArray(messages);
+ inputTypeName = typeof(string[]).FullName;
+
+ logger.LogDebug("Fan-In executor {ExecutorId}: aggregated {Count} messages", executorId, messages.Count);
+ }
+ else
+ {
+ // Normal case: process single message
+ (input, inputTypeName) = queue.Dequeue();
+ }
+
+ WorkflowExecutorInfo executorInfo = CreateExecutorInfo(executorId, executorBindings);
+ executorInputs.Add((executorId, input, inputTypeName, executorInfo));
+ }
+
+ // Dispatch all executors in parallel by starting all tasks first, then awaiting together
+ List> executorTasks = [];
+ foreach ((string executorId, string input, string? inputTypeName, WorkflowExecutorInfo executorInfo) in executorInputs)
+ {
+ // Start the task without awaiting - this enables parallel dispatch
+ Task task = this.DispatchExecutorAsync(
+ context, executorInfo, input, inputTypeName, logger, customStatus, sharedState);
+ executorTasks.Add(task);
+ }
+
+ // Wait for all executors to complete in parallel
+ string[] rawResults = await Task.WhenAll(executorTasks).ConfigureAwait(true);
+
+ // Process results and route messages
+ for (int i = 0; i < executorInputs.Count; i++)
+ {
+ string executorId = executorInputs[i].ExecutorId;
+ string rawResult = rawResults[i];
+
+ (string result, List sentMessages) = UnwrapActivityResult(rawResult, customStatus, sharedState);
+ lastResults[executorId] = result;
+
+ // Check for explicit halt request (via RequestHaltAsync)
+ if (HasHaltBeenRequested(customStatus, executorId))
+ {
+ haltRequested = true;
+ finalOutput = result;
+ logger.LogDebug("Halt requested by executor {ExecutorId}", executorId);
+ break;
+ }
+
+ // Route messages sent via SendMessageAsync (takes priority for void-returning executors)
+ if (sentMessages.Count > 0)
+ {
+ foreach (SentMessageInfo sentMessage in sentMessages)
{
- foreach (SentMessageInfo sentMessage in sentMessages)
+ if (!string.IsNullOrEmpty(sentMessage.Message))
{
- if (!string.IsNullOrEmpty(sentMessage.Message))
- {
- // Route to successors with the sent message's type
- RouteMessageToSuccessors(
- executorId, sentMessage.Message, sentMessage.TypeName, plan, messageQueues, logger);
- }
+ // Route to successors with the sent message's type
+ RouteMessageToSuccessors(
+ executorId, sentMessage.Message, sentMessage.TypeName, plan, messageQueues, logger);
}
}
- else if (!string.IsNullOrEmpty(result))
- {
- // Route executor's return value to successor executors via edges (for non-void executors)
- RouteMessageToSuccessors(
- executorId, result, plan, messageQueues, logger);
- }
+ }
+ else if (!string.IsNullOrEmpty(result))
+ {
+ // Route executor's return value to successor executors via edges (for non-void executors)
+ RouteMessageToSuccessors(
+ executorId, result, plan, messageQueues, logger);
}
if (haltRequested)
@@ -509,6 +533,18 @@ internal class DurableWorkflowRunner
return finalOutput ?? DetermineFinalResult(workflow, lastResults, customStatus);
}
+ ///
+ /// Aggregates multiple messages into a JSON array.
+ ///
+ /// The messages to aggregate.
+ /// A JSON array string containing all messages.
+ [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing string array.")]
+ [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing string array.")]
+ private static string AggregateMessagesToJsonArray(List messages)
+ {
+ return JsonSerializer.Serialize(messages);
+ }
+
///
/// Enqueues a message to an executor's message queue with type information.
///
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableYieldedOutputEvent.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableYieldedOutputEvent.cs
index db2149fde2..1d5a0799a2 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableYieldedOutputEvent.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableYieldedOutputEvent.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -11,6 +12,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// This is the durable equivalent of since that class has an internal
/// constructor not accessible from outside the Workflows assembly.
///
+[DebuggerDisplay("Yielded by {ExecutorId}: {Output}")]
public sealed class DurableYieldedOutputEvent : WorkflowEvent
{
///
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs
index 1312ba7a06..12aaa4d931 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs
@@ -1,10 +1,13 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
+
namespace Microsoft.Agents.AI.DurableTask;
///
/// Represents the complete execution plan for a workflow, including parallel execution levels.
///
+[DebuggerDisplay("Start = {StartExecutorId}, Levels = {Levels.Count}")]
internal sealed class WorkflowExecutionPlan
{
///
@@ -33,27 +36,6 @@ internal sealed class WorkflowExecutionPlan
///
public Dictionary ExecutorOutputTypes { get; } = [];
- ///
- /// Gets whether this workflow has any parallel execution opportunities.
- ///
- public bool HasParallelism => this.Levels.Any(l => l.Executors.Count > 1);
-
- ///
- /// Gets whether this workflow has any Fan-In points.
- ///
- public bool HasFanIn => this.Levels.Any(l => l.IsFanIn);
-
- ///
- /// Gets or sets whether this workflow contains cycles requiring iterative message-driven execution.
- ///
- public bool HasCycles { get; set; }
-
- ///
- /// Gets the back-edges that create cycles in the workflow graph.
- /// These edges are excluded from topological level computation but are followed during message-driven execution.
- ///
- public List<(string SourceId, string TargetId)> BackEdges { get; } = [];
-
///
/// Gets or sets the starting executor ID for the workflow.
///
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs
index 35db3052c3..4fbb5b2368 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
+using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -10,6 +11,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// The unique identifier of the executor.
/// Indicates whether this executor is an agentic executor.
/// The request port if this executor is a request port executor; otherwise, null.
+[DebuggerDisplay("{ExecutorId}, Agentic = {IsAgenticExecutor}, HITL = {IsRequestPortExecutor}")]
internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor, RequestPort? RequestPort = null)
{
///
@@ -25,6 +27,7 @@ internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExe
/// The level number (0-based, starting from the root executor).
/// The executors that can run in parallel at this level.
/// Indicates if this level is a Fan-In point (has executors with multiple predecessors).
+[DebuggerDisplay("Level {Level}: {Executors.Count} executor(s), FanIn = {IsFanIn}")]
internal sealed record WorkflowExecutionLevel(int Level, List Executors, bool IsFanIn);
///
@@ -118,13 +121,6 @@ internal static class WorkflowHelper
// Detect back-edges using DFS from start executor
HashSet<(string Source, string Target)> backEdges = DetectBackEdges(workflow.StartExecutorId, successors);
- // Mark cycles in plan
- plan.HasCycles = backEdges.Count > 0;
- foreach ((string source, string target) in backEdges)
- {
- plan.BackEdges.Add((source, target));
- }
-
// Calculate in-degrees, EXCLUDING back-edges
int[] inDegree = new int[executors.Count];
foreach (string executorId in executors.Keys)
@@ -145,7 +141,7 @@ internal static class WorkflowHelper
plan.EdgeConditions[condition.Key] = condition.Value;
}
- // Store the graph structure in the plan (reuse the built lists directly)
+ // Store the graph structure in the plan
foreach (string executorId in executors.Keys)
{
plan.Predecessors[executorId] = predecessors[executorId];