From d0ad92af7a4750f590f69379a7d6be48d07c9310 Mon Sep 17 00:00:00 2001 From: Shyju Krishnankutty Date: Tue, 27 Jan 2026 21:00:45 -0800 Subject: [PATCH] Minor cleanups --- .../DurableOptions.cs | 3 + .../DurableRequestInfoEvent.cs | 2 + .../DurableRun.cs | 2 + .../DurableStreamingRun.cs | 2 + .../DurableWorkflowCompletedEvent.cs | 2 + .../DurableWorkflowFailedEvent.cs | 2 + .../DurableWorkflowOptions.cs | 2 + .../DurableWorkflowRunner.cs | 150 +++++++++++------- .../DurableYieldedOutputEvent.cs | 2 + .../WorkflowExecutionPlan.cs | 24 +-- .../WorkflowHelper.cs | 12 +- 11 files changed, 117 insertions(+), 86 deletions(-) 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];