diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs index 7623252a63..52a7cee234 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -6,6 +6,9 @@ using Microsoft.Agents.AI.Workflows; using Microsoft.DurableTask; using Microsoft.Extensions.Logging; +#pragma warning disable CA1848 // Use LoggerMessage delegates +#pragma warning disable CA1873 // Expensive evaluation + namespace Microsoft.Agents.AI.DurableTask; /// @@ -247,10 +250,11 @@ public class DurableWorkflowRunner break; // No more work } -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates, expensive evaluation - logger.LogDebug("Superstep {Step}: {Count} active executor(s): {Executors}", - superstep, activeExecutors.Count, string.Join(", ", activeExecutors)); -#pragma warning restore CA1848, CA1873 + logger.LogDebug( + "Superstep {Step}: {Count} active executor(s): {Executors}", + superstep, + activeExecutors.Count, + string.Join(", ", activeExecutors)); // Process each active executor foreach (string executorId in activeExecutors) @@ -277,9 +281,7 @@ public class DurableWorkflowRunner { haltRequested = true; finalOutput = result; -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates logger.LogDebug("Halt requested by executor {ExecutorId}", executorId); -#pragma warning restore CA1848, CA1873 break; } @@ -315,9 +317,7 @@ public class DurableWorkflowRunner if (superstep >= MaxSupersteps) { -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates logger.LogWarning("Workflow reached maximum superstep limit ({MaxSteps})", MaxSupersteps); -#pragma warning restore CA1848, CA1873 } // Return final output or last result from output executors @@ -388,6 +388,11 @@ public class DurableWorkflowRunner /// /// Routes a message through edges to successor executors. /// + /// The source executor ID. + /// The serialized message to route. + /// The workflow execution plan. + /// The message queues for each executor. + /// The logger instance. [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] private static void RouteMessageToSuccessors( @@ -397,55 +402,12 @@ public class DurableWorkflowRunner Dictionary> messageQueues, ILogger logger) { - if (!plan.Successors.TryGetValue(sourceId, out List? successors)) - { - return; // No outgoing edges - } - - // Get the output type of the source executor to pass as input type to successors plan.ExecutorOutputTypes.TryGetValue(sourceId, out Type? sourceOutputType); - string? inputTypeName = sourceOutputType?.FullName; - - foreach (string sinkId in successors) - { - // Check edge condition - if (plan.EdgeConditions.TryGetValue((sourceId, sinkId), out Func? condition) - && condition is not null) - { - try - { - // Deserialize the message for condition evaluation - object? messageObj = DeserializeForCondition(message, sourceOutputType); - - if (!condition(messageObj)) - { -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates - logger.LogDebug("Edge {Source} -> {Sink}: condition returned false, skipping", - sourceId, sinkId); -#pragma warning restore CA1848, CA1873 - continue; - } - } - catch (Exception ex) - { -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates - logger.LogWarning(ex, "Failed to evaluate condition for edge {Source} -> {Sink}, skipping", - sourceId, sinkId); -#pragma warning restore CA1848, CA1873 - continue; - } - } - - // Queue message to successor with type information -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates - logger.LogDebug("Edge {Source} -> {Sink}: routing message", sourceId, sinkId); -#pragma warning restore CA1848, CA1873 - EnqueueMessage(messageQueues, sinkId, message, inputTypeName); - } + RouteMessageToSuccessorsCore(sourceId, message, sourceOutputType?.FullName, sourceOutputType, plan, messageQueues, logger); } /// - /// Routes a message through edges to successor executors, with an explicit type name for the message. + /// Routes a message through edges to successor executors, with an explicit type name. /// Used for messages sent via SendMessageAsync. /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] @@ -459,19 +421,8 @@ public class DurableWorkflowRunner Dictionary> messageQueues, ILogger logger) { - if (!plan.Successors.TryGetValue(sourceId, out List? successors)) - { - return; // No outgoing edges - } - - // Use explicit type name if provided, otherwise fall back to executor output type + Type? messageType = !string.IsNullOrEmpty(explicitTypeName) ? Type.GetType(explicitTypeName) : null; string? inputTypeName = explicitTypeName; - Type? messageType = null; - - if (!string.IsNullOrEmpty(explicitTypeName)) - { - messageType = Type.GetType(explicitTypeName); - } if (messageType is null && plan.ExecutorOutputTypes.TryGetValue(sourceId, out Type? sourceOutputType)) { @@ -479,44 +430,77 @@ public class DurableWorkflowRunner inputTypeName = sourceOutputType?.FullName; } + RouteMessageToSuccessorsCore(sourceId, message, inputTypeName, messageType, plan, messageQueues, logger); + } + + /// + /// Core implementation for routing messages to successors. + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] + private static void RouteMessageToSuccessorsCore( + string sourceId, + string message, + string? inputTypeName, + Type? messageType, + WorkflowExecutionPlan plan, + Dictionary> messageQueues, + ILogger logger) + { + if (!plan.Successors.TryGetValue(sourceId, out List? successors)) + { + return; + } + foreach (string sinkId in successors) { - // Check edge condition - if (plan.EdgeConditions.TryGetValue((sourceId, sinkId), out Func? condition) - && condition is not null) + if (!TryEvaluateEdgeCondition(sourceId, sinkId, message, messageType, plan, logger)) { - try - { - // Deserialize the message for condition evaluation - object? messageObj = DeserializeForCondition(message, messageType); - - if (!condition(messageObj)) - { -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates - logger.LogDebug("Edge {Source} -> {Sink}: condition returned false, skipping", - sourceId, sinkId); -#pragma warning restore CA1848, CA1873 - continue; - } - } - catch (Exception ex) - { -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates - logger.LogWarning(ex, "Failed to evaluate condition for edge {Source} -> {Sink}, skipping", - sourceId, sinkId); -#pragma warning restore CA1848, CA1873 - continue; - } + continue; } - // Queue message to successor with type information -#pragma warning disable CA1848, CA1873 // Use LoggerMessage delegates - logger.LogDebug("Edge {Source} -> {Sink}: routing sent message (type: {TypeName})", sourceId, sinkId, inputTypeName); -#pragma warning restore CA1848, CA1873 + logger.LogDebug("Edge {Source} -> {Sink}: routing message", sourceId, sinkId); EnqueueMessage(messageQueues, sinkId, message, inputTypeName); } } + /// + /// Evaluates an edge condition if one exists. + /// + /// True if the message should be routed (no condition or condition passed); false otherwise. + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] + private static bool TryEvaluateEdgeCondition( + string sourceId, + string sinkId, + string message, + Type? messageType, + WorkflowExecutionPlan plan, + ILogger logger) + { + if (!plan.EdgeConditions.TryGetValue((sourceId, sinkId), out Func? condition) || condition is null) + { + return true; + } + + try + { + object? messageObj = DeserializeForCondition(message, messageType); + if (!condition(messageObj)) + { + logger.LogDebug("Edge {Source} -> {Sink}: condition returned false, skipping", sourceId, sinkId); + return false; + } + + return true; + } + catch (Exception ex) + { + logger.LogWarning(ex, "Failed to evaluate condition for edge {Source} -> {Sink}, skipping", sourceId, sinkId); + return false; + } + } + /// /// Gets the final result for a message-driven workflow execution. /// Checks yielded outputs first (from YieldOutputAsync calls), then falls back to executor results. @@ -623,55 +607,76 @@ public class DurableWorkflowRunner try { - // Try to deserialize as DurableActivityOutput DurableActivityOutput? output = JsonSerializer.Deserialize(rawResult); - // Check if this is actually a DurableActivityOutput (has Result property set or state updates or sent messages) - // This distinguishes it from other JSON objects that would deserialize with default/empty values - if (output is not null && (output.Result is not null || output.StateUpdates.Count > 0 || output.ClearedScopes.Count > 0 || output.Events.Count > 0 || output.SentMessages.Count > 0)) + if (output is null || !IsValidActivityOutput(output)) { - // Apply cleared scopes first - foreach (string clearedScope in output.ClearedScopes) - { - string scopePrefix = clearedScope == "__default__" ? "__default__:" : $"{clearedScope}:"; - List keysToRemove = sharedState.Keys - .Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal)) - .ToList(); - - foreach (string key in keysToRemove) - { - sharedState.Remove(key); - } - } - - // Apply state updates - foreach (KeyValuePair update in output.StateUpdates) - { - if (update.Value is null) - { - sharedState.Remove(update.Key); - } - else - { - sharedState[update.Key] = update.Value; - } - } - - // Add events to the accumulated list - if (output.Events.Count > 0) - { - customStatus.Events.AddRange(output.Events); - } - - return (output.Result ?? string.Empty, output.SentMessages); + return (rawResult, []); } + + ApplyClearedScopes(output.ClearedScopes, sharedState); + ApplyStateUpdates(output.StateUpdates, sharedState); + + if (output.Events.Count > 0) + { + customStatus.Events.AddRange(output.Events); + } + + return (output.Result ?? string.Empty, output.SentMessages); } catch (JsonException) { - // Not a wrapped result, return as-is + return (rawResult, []); } + } - return (rawResult, []); + /// + /// Checks if the deserialized output is a valid DurableActivityOutput (not just default values). + /// + private static bool IsValidActivityOutput(DurableActivityOutput output) + { + return output.Result is not null + || output.StateUpdates.Count > 0 + || output.ClearedScopes.Count > 0 + || output.Events.Count > 0 + || output.SentMessages.Count > 0; + } + + /// + /// Clears all state entries matching the specified scopes. + /// + private static void ApplyClearedScopes(List clearedScopes, Dictionary sharedState) + { + foreach (string clearedScope in clearedScopes) + { + string scopePrefix = clearedScope == "__default__" ? "__default__:" : $"{clearedScope}:"; + List keysToRemove = sharedState.Keys + .Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal)) + .ToList(); + + foreach (string key in keysToRemove) + { + sharedState.Remove(key); + } + } + } + + /// + /// Applies state updates to the shared state dictionary. + /// + private static void ApplyStateUpdates(Dictionary stateUpdates, Dictionary sharedState) + { + foreach (KeyValuePair update in stateUpdates) + { + if (update.Value is null) + { + sharedState.Remove(update.Key); + } + else + { + sharedState[update.Key] = update.Value; + } + } } /// @@ -686,32 +691,6 @@ public class DurableWorkflowRunner } } - /// - /// Wrapper for activity output that includes state updates and events. - /// - internal sealed class ActivityOutputWithState - { - /// - /// Gets or sets the serialized result of the activity. - /// - public string? Result { get; set; } - - /// - /// Gets or sets state updates made during activity execution. - /// - public Dictionary StateUpdates { get; set; } = []; - - /// - /// Gets or sets scopes that were cleared during activity execution. - /// - public List ClearedScopes { get; set; } = []; - - /// - /// Gets or sets the serialized workflow events emitted during activity execution. - /// - public List Events { get; set; } = []; - } - /// /// Wrapper for serialized workflow events that includes type information for proper deserialization. ///