From e1147273da297092532a80d1db88d71c4660e018 Mon Sep 17 00:00:00 2001 From: Jacob Alber Date: Thu, 28 Aug 2025 18:25:57 -0400 Subject: [PATCH] .NET: fix: FanIn Edge does not work (#541) * fix: FanIn Edge does not work We were not creating the state for FanIn edge in EdgeMap correctly, leading to crashes. After fixing that, it turns out the logic in FanInEdgeRunner was only forwarding the last message, not all of them. * fix: Remove duplicate code and fix typo --- .../Execution/EdgeMap.cs | 5 + .../Execution/FanInEdgeRunner.cs | 23 ++- .../EdgeMapSmokeTests.cs | 36 ++++ .../EdgeRunnerTests.cs | 181 ++++++++++++++++++ .../ForwardMessageExecutor.cs | 2 +- .../MessageDeliveryValidation.cs | 39 ++++ .../MessagingTestHelpers.cs | 40 ++++ .../TestRunContext.cs | 76 ++++++++ 8 files changed, 392 insertions(+), 10 deletions(-) create mode 100644 dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs create mode 100644 dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs create mode 100644 dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs create mode 100644 dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessagingTestHelpers.cs create mode 100644 dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs index a8f86848dc..27bfe7e9ba 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs @@ -32,6 +32,11 @@ internal class EdgeMap _ => throw new NotSupportedException($"Unsupported edge type: {edge.EdgeType}") }; + if (edgeRunner is FanInEdgeRunner fanInRunner) + { + this._fanInState[edge.Data.Connection] = fanInRunner.CreateState(); + } + this._edgeRunners[edge.Data.Connection] = edgeRunner; } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs index 222620a847..f61d0cfe2c 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs @@ -12,12 +12,12 @@ internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData public FanInEdgeState CreateState() => new(this.EdgeData); - public async ValueTask ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer) + public async ValueTask> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer) { if (envelope.TargetId != null && this.EdgeData.SinkId != envelope.TargetId) { // This message is not for us. - return null; + return []; } object message = envelope.Message; @@ -25,18 +25,23 @@ internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData if (releasedMessages is null) { // Not ready to process yet. - return null; + return []; } Executor target = await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer) - .ConfigureAwait(false); + .ConfigureAwait(false); - if (target.CanHandle(message.GetType())) + List> messageTasks = []; + + foreach (var messageTask in releasedMessages) { - tracer?.TraceActivated(target.Id); - return await target.ExecuteAsync(message, envelope.MessageType, this.BoundContext) - .ConfigureAwait(false); + if (target.CanHandle(messageTask.GetType())) + { + tracer?.TraceActivated(target.Id); + messageTasks.Add(target.ExecuteAsync(messageTask, envelope.MessageType, this.BoundContext).AsTask()); + } } - return null; + + return await Task.WhenAll(messageTasks.ToArray()).ConfigureAwait(false); } } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs new file mode 100644 index 0000000000..da00a17554 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs @@ -0,0 +1,36 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.Agents.Workflows.Execution; + +namespace Microsoft.Agents.Workflows.UnitTests; + +public class EdgeMapSmokeTests +{ + [Fact] + public async Task Test_EdgeMap_MaintainsFanInEdgeStateAsync() + { + TestRunContext runContext = new(); + + runContext.Executors["executor1"] = new ForwardMessageExecutor("executor1"); + runContext.Executors["executor2"] = new ForwardMessageExecutor("executor2"); + runContext.Executors["executor3"] = new ForwardMessageExecutor("executor3"); + + Dictionary> workflowEdges = new(); + + FanInEdgeData edgeData = new(["executor1", "executor2"], "executor3"); + Edge fanInEdge = new(edgeData); + + workflowEdges["executor1"] = [fanInEdge]; + workflowEdges["executor2"] = [fanInEdge]; + + EdgeMap edgeMap = new(runContext, workflowEdges, [], "executor1", null); + + await edgeMap.InvokeEdgeAsync(fanInEdge, "executor1", new("part1")); + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + + await edgeMap.InvokeEdgeAsync(fanInEdge, "executor2", new("part2")); + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, ("executor3", ["part1", "part2"])); + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs new file mode 100644 index 0000000000..9e0281e744 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs @@ -0,0 +1,181 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.Agents.Workflows.Execution; + +namespace Microsoft.Agents.Workflows.UnitTests; + +public class EdgeRunnerTests +{ + private async Task CreateAndRunDirectedEdgeTestAsync(bool? conditionMatch = null, bool? targetMatch = null) + { + const string MessageVariant1 = "test"; + const string MessageVariant2 = "something else"; + + Func? condition + = conditionMatch.HasValue + ? message => message is string value && value.Equals(conditionMatch.Value + ? MessageVariant1 + : MessageVariant2, StringComparison.Ordinal) + : null; + + string? targetId + = targetMatch.HasValue + ? (targetMatch.Value ? "executor2" : "executor1") + : null; + + TestRunContext runContext = new(); + + runContext.Executors["executor1"] = new ForwardMessageExecutor("executor1"); + runContext.Executors["executor2"] = new ForwardMessageExecutor("executor2"); + + DirectEdgeData edgeData = new("executor1", "executor2", condition); + DirectEdgeRunner runner = new(runContext, edgeData); + + MessageEnvelope envelope = new(MessageVariant1, targetId: targetId); + + await runner.ChaseAsync(envelope, tracer: null); + + bool expectMessage = (!conditionMatch.HasValue || conditionMatch.Value) + && (!targetMatch.HasValue || targetMatch.Value); + + if (expectMessage) + { + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, ("executor2", [MessageVariant1])); + } + else + { + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + } + } + + [Fact] + public async Task Test_DirectEdgeRunnerAsync() + { + // Test matrix: + // NoCondition vs Condition(=> true) vs Condition(=> false) + // Untargeted vs Targeted(matching) vs Targeted(not matching) + + await this.CreateAndRunDirectedEdgeTestAsync(); // NoCondition, Untargeted + + await this.CreateAndRunDirectedEdgeTestAsync(targetMatch: true); // NoCondition, Targeted + await this.CreateAndRunDirectedEdgeTestAsync(targetMatch: false); // NoCondition, Targeted(not matching) + + await this.CreateAndRunDirectedEdgeTestAsync(conditionMatch: true); // Condition(=> true), Untargeted + await this.CreateAndRunDirectedEdgeTestAsync(conditionMatch: false); // Condition(=> false), Untargeted + + await this.CreateAndRunDirectedEdgeTestAsync(conditionMatch: true, targetMatch: true); // Condition(=> true), Targeted(matching) + await this.CreateAndRunDirectedEdgeTestAsync(conditionMatch: true, targetMatch: false); // Condition(=> true), Targeted(not matching) + await this.CreateAndRunDirectedEdgeTestAsync(conditionMatch: false, targetMatch: true); // Condition(=> false), Targeted(matching) + await this.CreateAndRunDirectedEdgeTestAsync(conditionMatch: false, targetMatch: false); // Condition(=> false), Targeted(not matching) + } + + private async Task CreateAndRunFanOutEdgeTestAsync(bool? assignerSelectsEmpty = null, bool? targetMatch = null) + { + TestRunContext runContext = new(); + + runContext.Executors["executor1"] = new ForwardMessageExecutor("executor1"); + runContext.Executors["executor2"] = new ForwardMessageExecutor("executor2"); + runContext.Executors["executor3"] = new ForwardMessageExecutor("executor3"); + + Func>? assigner + = assignerSelectsEmpty.HasValue + ? (message, count) => assignerSelectsEmpty.Value ? [] : [0] + : null; + + string? targetId + = targetMatch.HasValue + ? (targetMatch.Value ? "executor2" : "executor1") + : null; + + FanOutEdgeData edgeData = new("executor1", ["executor2", "executor3"], assigner); + FanOutEdgeRunner runner = new(runContext, edgeData); + + MessageEnvelope envelope = new("test", targetId: targetId); + + await runner.ChaseAsync(envelope, tracer: null); + + bool expectForwardFrom2 = (!assignerSelectsEmpty.HasValue || !assignerSelectsEmpty.Value) + && (!targetMatch.HasValue || targetMatch.Value); + bool expectForwardFrom3 = !assignerSelectsEmpty.HasValue && !targetMatch.HasValue; // if there is a target, it is never executor3 + + List<(string expectedSender, List expectedMessages)> expectedForwards = []; + if (expectForwardFrom2) + { + expectedForwards.Add(("executor2", ["test"])); + } + + if (expectForwardFrom3) + { + expectedForwards.Add(("executor3", ["test"])); + } + + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, expectedForwards.ToArray()); + } + + [Fact] + public async Task Test_FanOutEdgeRunnerAsync() + { + // Test matrix: + // NoAssigned vs Assigner(includes output) vs Assigner(does not include output) + // Untargeted vs Targeted(matching) vs Targeted(not matching) + + await this.CreateAndRunFanOutEdgeTestAsync(); // NoAssigner, Untargeted + + await this.CreateAndRunFanOutEdgeTestAsync(targetMatch: true); // NoAssigner, Targeted(matching) + await this.CreateAndRunFanOutEdgeTestAsync(targetMatch: false); // NoAssigner, Targeted(not matching) + + await this.CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: false); // Assigner(includes output), Untargeted + await this.CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: true); // Assigner(does not include output), Untargeted + + await this.CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: false, targetMatch: true); // Assigner(includes output), Targeted(matching) + await this.CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: false, targetMatch: false); // Assigner(includes output), Targeted(not matching) + await this.CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: true, targetMatch: true); // Assigner(does not include output), Targeted(matching) + await this.CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: true, targetMatch: false); // Assigner(does not include output), Targeted(not matching) + } + + [Fact] + public async Task Test_FanInEdgeRunnerAsync() + { + TestRunContext runContext = new(); + + runContext.Executors["executor1"] = new ForwardMessageExecutor("executor1"); + runContext.Executors["executor2"] = new ForwardMessageExecutor("executor2"); + runContext.Executors["executor3"] = new ForwardMessageExecutor("executor3"); + + FanInEdgeData edgeData = new(["executor1", "executor2"], "executor3"); + FanInEdgeRunner runner = new(runContext, edgeData); + + // Step 1: Send message from executor1, should not forward yet. + // Step 2: Send targeted message to executor1 from executor2, should not forward + // Step 3: Send message from executor1, should not forward yet. + // Step 4: Send message from executor2, should forward now. + + FanInEdgeState state = runner.CreateState(); + await RunIteration(); + + // Repeat the same sequence, to ensure state is properly reset inside of FanInEdgeState. + runContext.QueuedMessages.Clear(); + await RunIteration(); + + async ValueTask RunIteration() + { + await runner.ChaseAsync("executor1", new("part1"), state, tracer: null); + + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + + await runner.ChaseAsync("executor2", new("part-for-1", targetId: "executor1"), state, tracer: null); + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + + await runner.ChaseAsync("executor1", new("part2", targetId: "executor3"), state, tracer: null); + + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + + await runner.ChaseAsync("executor2", new("final part"), state, tracer: null); + + MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, ("executor3", ["part1", "part2", "final part"])); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/ForwardMessageExecutor.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/ForwardMessageExecutor.cs index 7724f73630..256660d0ba 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/ForwardMessageExecutor.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/ForwardMessageExecutor.cs @@ -2,7 +2,7 @@ namespace Microsoft.Agents.Workflows.UnitTests; -internal sealed class ForwardMessageExecutor : Executor where TMessage : notnull +internal sealed class ForwardMessageExecutor(string? id = null) : Executor(id) where TMessage : notnull { protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) { diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs new file mode 100644 index 0000000000..47cf00172e --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using FluentAssertions; +using Microsoft.Agents.Workflows.Execution; + +namespace Microsoft.Agents.Workflows.UnitTests; + +internal static class MessageDeliveryValidation +{ + public static void CheckForwarded(Dictionary> queuedMessages, params (string expectedSender, List expectedMessages)[] expectedForwards) + { + queuedMessages.Should().HaveCount(expectedForwards.Length); + + IEnumerable> perSenderValidations = expectedForwards.Select( + (forward) => + { + (string expectedSender, List expectedMessages) = forward; + + return (Action)( + (string senderId) => + { + senderId.Should().Be(expectedSender); + queuedMessages[senderId].Should().HaveCount(expectedMessages.Count); + + Action[] validations + = expectedMessages.Select(message => (Action)(envelope => envelope!.Message.Should().Be(message))) + .ToArray(); + + Assert.Collection(queuedMessages[senderId], validations); + }); + } + ); + + Assert.Collection(queuedMessages.Keys, perSenderValidations.ToArray()); + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessagingTestHelpers.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessagingTestHelpers.cs new file mode 100644 index 0000000000..07660a7506 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessagingTestHelpers.cs @@ -0,0 +1,40 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using FluentAssertions; +using Microsoft.Agents.Workflows.Execution; +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.Workflows.UnitTests; + +internal static class MessagingTestHelpers +{ + private static void CheckForwarded(Dictionary> queuedMessages, params (string expectedSender, List expectedMessages)[] expectedForwards) + { + queuedMessages.Should().HaveCount(expectedForwards.Length); + + IEnumerable> perSenderValidations = expectedForwards.Select( + (forward) => + { + (string expectedSender, List expectedMessages) = forward; + + return (Action)( + (string senderId) => + { + senderId.Should().Be(expectedSender); + queuedMessages[senderId].Should().HaveCount(expectedMessages.Count); + + Action[] validations + = expectedMessages.Select(message => (Action)(envelope => envelope!.Message.Should().Be(message))) + .ToArray(); + + Assert.Collection(queuedMessages[senderId], validations); + }); + } + ); + + Assert.Collection(queuedMessages.Keys, perSenderValidations.ToArray()); + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs new file mode 100644 index 0000000000..4e19bf38b1 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs @@ -0,0 +1,76 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.Agents.Workflows.Execution; + +namespace Microsoft.Agents.Workflows.UnitTests; + +public class TestRunContext : IRunnerContext +{ + private sealed class BoundContext(string executorId, TestRunContext runnerContext) : IWorkflowContext + { + public ValueTask AddEventAsync(WorkflowEvent workflowEvent) + => runnerContext.AddEventAsync(workflowEvent); + + public ValueTask QueueClearScopeAsync(string? scopeName = null) + => default; + + public ValueTask QueueStateUpdateAsync(string key, T? value, string? scopeName = null) + => default; + + public ValueTask ReadStateAsync(string key, string? scopeName = null) + => new(default(T?)); + + public ValueTask> ReadStateKeysAsync(string? scopeName = null) + => new([]); + + public ValueTask SendMessageAsync(object message, string? targetId = null) + => runnerContext.SendMessageAsync(executorId, message, targetId); + } + + public List Events { get; } = new(); + + public ValueTask AddEventAsync(WorkflowEvent workflowEvent) + { + this.Events.Add(workflowEvent); + return default; + } + + public IWorkflowContext Bind(string executorId) + { + return new BoundContext(executorId, this); + } + + public List ExternalRequests { get; } = new(); + public ValueTask PostAsync(ExternalRequest request) + { + this.ExternalRequests.Add(request); + return default; + } + + internal Dictionary> QueuedMessages { get; } = new(); + public ValueTask SendMessageAsync(string sourceId, object message, string? targetId = null) + { + if (!this.QueuedMessages.TryGetValue(sourceId, out List? deliveryQueue)) + { + this.QueuedMessages[sourceId] = deliveryQueue = new(); + } + + deliveryQueue.Add(new(message, targetId: targetId)); + return default; + } + + StepContext IRunnerContext.Advance() + { + throw new NotImplementedException(); + } + + public Dictionary Executors { get; } = new(); + + ValueTask IRunnerContext.EnsureExecutorAsync(string executorId, IStepTracer? tracer) + { + return new(this.Executors[executorId]); + } +}