diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs index 0144ce83d5..a5284fc508 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -156,19 +156,15 @@ public class DurableWorkflowRunner } string json = input; - if (input.StartsWith('"') && input.EndsWith('"')) + + // Check for double-serialized strings (e.g., "\"actual json\"") + // A properly double-serialized string starts with \" and ends with \" + if (input.Length > 2 && input[0] == '"' && input[^1] == '"' && input[1] == '\\') { - try + string? innerJson = JsonSerializer.Deserialize(input); + if (innerJson is not null) { - string? innerJson = JsonSerializer.Deserialize(input); - if (innerJson is not null) - { - json = innerJson; - } - } - catch (JsonException) - { - // Not double-serialized, use original + json = innerJson; } } @@ -183,7 +179,7 @@ public class DurableWorkflowRunner ILogger logger) { WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow); - Dictionary results = []; + Dictionary results = new(plan.Levels.Sum(l => l.Executors.Count)); foreach (WorkflowExecutionLevel level in plan.Levels) { @@ -204,14 +200,16 @@ public class DurableWorkflowRunner } else { - List> tasks = []; - foreach (WorkflowExecutorInfo executorInfo in eligibleExecutors) + Task<(string Id, string Result)>[] tasks = new Task<(string Id, string Result)>[eligibleExecutors.Count]; + for (int i = 0; i < eligibleExecutors.Count; i++) { + WorkflowExecutorInfo executorInfo = eligibleExecutors[i]; string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); - tasks.Add(this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger)); + tasks[i] = this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger); } - foreach ((string id, string result) in await Task.WhenAll(tasks).ConfigureAwait(true)) + (string Id, string Result)[] completedTasks = await Task.WhenAll(tasks).ConfigureAwait(true); + foreach ((string id, string result) in completedTasks) { results[id] = result; } @@ -234,15 +232,13 @@ public class DurableWorkflowRunner WorkflowExecutionPlan plan, ILogger logger) { - List eligible = []; + List eligible = new(executors.Count); foreach (WorkflowExecutorInfo executorInfo in executors) { - List predecessors = plan.Predecessors[executorInfo.ExecutorId]; - - // Root executor (no predecessors) is always eligible - if (predecessors.Count == 0) + if (!plan.Predecessors.TryGetValue(executorInfo.ExecutorId, out List? predecessors) || predecessors.Count == 0) { + // Root executor (no predecessors) is always eligible eligible.Add(executorInfo); continue; } @@ -252,16 +248,9 @@ public class DurableWorkflowRunner foreach (string predecessorId in predecessors) { // Get the condition for this edge (predecessor -> current executor) - if (!plan.EdgeConditions.TryGetValue((predecessorId, executorInfo.ExecutorId), out Func? condition)) + if (!plan.EdgeConditions.TryGetValue((predecessorId, executorInfo.ExecutorId), out Func? condition) || condition is null) { - // No condition registered for this edge, assume it's eligible - isEligible = true; - break; - } - - if (condition is null) - { - // Edge has no condition, always eligible + // No condition registered for this edge or edge has no condition, always eligible isEligible = true; break; } @@ -272,7 +261,7 @@ public class DurableWorkflowRunner try { // Get the predecessor's output type for proper deserialization - Type? predecessorOutputType = plan.ExecutorOutputTypes.GetValueOrDefault(predecessorId); + plan.ExecutorOutputTypes.TryGetValue(predecessorId, out Type? predecessorOutputType); // Deserialize the predecessor result to the expected type for condition evaluation object? resultObject = DeserializeForCondition(predecessorResult, predecessorOutputType); @@ -382,9 +371,7 @@ public class DurableWorkflowRunner Dictionary results, WorkflowExecutionPlan plan) { - List predecessors = plan.Predecessors[executorId]; - - if (predecessors.Count == 0) + if (!plan.Predecessors.TryGetValue(executorId, out List? predecessors) || predecessors.Count == 0) { return initialInput; } @@ -394,7 +381,7 @@ public class DurableWorkflowRunner return results.TryGetValue(predecessors[0], out string? result) ? result : initialInput; } - List aggregated = []; + List aggregated = new(predecessors.Count); foreach (string predecessorId in predecessors) { if (results.TryGetValue(predecessorId, out string? result)) @@ -409,14 +396,15 @@ public class DurableWorkflowRunner private static string GetFinalResult(WorkflowExecutionPlan plan, Dictionary results) { WorkflowExecutionLevel lastLevel = plan.Levels[^1]; + List lastExecutors = lastLevel.Executors; - if (lastLevel.Executors.Count == 1) + if (lastExecutors.Count == 1) { - return results[lastLevel.Executors[0].ExecutorId]; + return results.TryGetValue(lastExecutors[0].ExecutorId, out string? singleResult) ? singleResult : string.Empty; } - List finalResults = []; - foreach (WorkflowExecutorInfo executor in lastLevel.Executors) + List finalResults = new(lastExecutors.Count); + foreach (WorkflowExecutorInfo executor in lastExecutors) { if (results.TryGetValue(executor.ExecutorId, out string? result)) { diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs index a08f314433..c9cb307723 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs @@ -105,16 +105,18 @@ public static class WorkflowHelper WorkflowExecutionPlan plan = new(); // Build adjacency lists (successors and predecessors) - Dictionary> successors = []; - Dictionary> predecessors = []; - Dictionary inDegree = []; + Dictionary> successors = new(executors.Count); + Dictionary> predecessors = new(executors.Count); + int[] inDegree = new int[executors.Count]; + Dictionary executorIndex = new(executors.Count); // Initialize all executors and extract their output types + int index = 0; foreach (KeyValuePair executor in executors) { + executorIndex[executor.Key] = index++; successors[executor.Key] = []; predecessors[executor.Key] = []; - inDegree[executor.Key] = 0; // Extract output type from executor type (e.g., Executor -> TOutput) plan.ExecutorOutputTypes[executor.Key] = GetExecutorOutputType(executor.Value.ExecutorType); @@ -124,16 +126,17 @@ public static class WorkflowHelper foreach (KeyValuePair> edgeGroup in edges) { string sourceId = edgeGroup.Key; + List sourceSuccessors = successors[sourceId]; foreach (EdgeInfo edge in edgeGroup.Value) { foreach (string sinkId in edge.Connection.SinkIds) { - if (executors.ContainsKey(sinkId)) + if (executorIndex.TryGetValue(sinkId, out int sinkIdx)) { - successors[sourceId].Add(sinkId); + sourceSuccessors.Add(sinkId); predecessors[sinkId].Add(sourceId); - inDegree[sinkId]++; + inDegree[sinkIdx]++; } } } @@ -145,74 +148,82 @@ public static class WorkflowHelper plan.EdgeConditions[condition.Key] = condition.Value; } - // Store the graph structure in the plan + // Store the graph structure in the plan (reuse the built lists directly) foreach (string executorId in executors.Keys) { - plan.Predecessors[executorId] = [.. predecessors[executorId]]; - plan.Successors[executorId] = [.. successors[executorId]]; + plan.Predecessors[executorId] = predecessors[executorId]; + plan.Successors[executorId] = successors[executorId]; } - // Build execution levels using modified Kahn's algorithm - // Instead of processing one at a time, we process all nodes with in-degree 0 at once (same level) - HashSet processed = []; - Dictionary currentInDegree = new(inDegree); - int levelNumber = 0; - - while (processed.Count < executors.Count) + // Build execution levels using queue-based Kahn's algorithm + // Process all nodes with in-degree 0 at once (same level) for parallel execution + Queue currentLevel = new(); + foreach (KeyValuePair kvp in executorIndex) { - // Find all executors that can be executed at this level (in-degree == 0 and not yet processed) - List currentLevelIds = []; - - foreach (KeyValuePair kvp in currentInDegree) + if (inDegree[kvp.Value] == 0) { - if (kvp.Value == 0 && !processed.Contains(kvp.Key)) - { - currentLevelIds.Add(kvp.Key); - } + currentLevel.Enqueue(kvp.Key); } + } - // If no executors found but not all processed, there might be a cycle - if (currentLevelIds.Count == 0) + int levelNumber = 0; + int processedCount = 0; + + while (currentLevel.Count > 0) + { + List levelExecutors = new(currentLevel.Count); + Queue nextLevel = new(); + bool isFanIn = false; + + while (currentLevel.Count > 0) { - // Add remaining unprocessed executors - foreach (string executorId in executors.Keys) + string executorId = currentLevel.Dequeue(); + processedCount++; + + ExecutorBinding executorBinding = executors[executorId]; + bool isAgentic = IsAgentExecutorType(executorBinding.ExecutorType); + levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic)); + + // Check Fan-In for this executor + if (predecessors[executorId].Count > 1) { - if (!processed.Contains(executorId)) - { - currentLevelIds.Add(executorId); - } + isFanIn = true; } - if (currentLevelIds.Count == 0) - { - break; - } - } - - // Check if this level is a Fan-In point (any executor has multiple predecessors) - bool isFanIn = currentLevelIds.Any(id => predecessors[id].Count > 1); - - // Convert to WorkflowExecutorInfo - List levelExecutors = []; - foreach (string executorId in currentLevelIds) - { - processed.Add(executorId); - - if (executors.TryGetValue(executorId, out ExecutorBinding? executorBinding)) - { - bool isAgentic = IsAgentExecutorType(executorBinding.ExecutorType); - levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic)); - } - - // Decrement in-degree of all successors + // Decrement in-degree of all successors and enqueue those ready for next level foreach (string successor in successors[executorId]) { - currentInDegree[successor]--; + int successorIdx = executorIndex[successor]; + if (--inDegree[successorIdx] == 0) + { + nextLevel.Enqueue(successor); + } } } plan.Levels.Add(new WorkflowExecutionLevel(levelNumber, levelExecutors, isFanIn)); levelNumber++; + currentLevel = nextLevel; + } + + // Handle cycle detection: if not all executors were processed, there's a cycle + if (processedCount < executors.Count) + { + List remainingExecutors = []; + foreach (KeyValuePair executor in executors) + { + if (inDegree[executorIndex[executor.Key]] > 0) + { + bool isAgentic = IsAgentExecutorType(executor.Value.ExecutorType); + remainingExecutors.Add(new WorkflowExecutorInfo(executor.Key, isAgentic)); + } + } + + if (remainingExecutors.Count > 0) + { + bool isFanIn = remainingExecutors.Exists(e => predecessors[e.ExecutorId].Count > 1); + plan.Levels.Add(new WorkflowExecutionLevel(levelNumber, remainingExecutors, isFanIn)); + } } return plan;