diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/FanOutEdgeRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/FanOutEdgeRunner.cs index 3ff3469f1ff..e4bac5f3fb6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/FanOutEdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/FanOutEdgeRunner.cs @@ -29,6 +29,10 @@ this.EdgeData.EdgeAssigner is null : this.EdgeData.EdgeAssigner(message, this.EdgeData.SinkIds.Count) .Select(i => this.EdgeData.SinkIds[i]); + // The assigner, or the target list itself, can name a target more than once, for example when two + // tags on a message route to the same executor. Deliver to each target once, in the order first named. + targetIds = targetIds.Distinct(StringComparer.Ordinal); + Executor[] result = await Task.WhenAll(targetIds.Where(IsValidTarget) .Select(tid => this.RunContext.EnsureExecutorAsync(tid, stepTracer) .AsTask())) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/EdgeRunnerTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/EdgeRunnerTests.cs index 22ae41c7010..47ccf4ffa80 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/EdgeRunnerTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/EdgeRunnerTests.cs @@ -151,6 +151,85 @@ public async Task Test_FanOutEdgeRunnerAsync() await CreateAndRunFanOutEdgeTestAsync(assignerSelectsEmpty: true, targetMatch: false); // Assigner(does not include output), Targeted(not matching) } + [Fact] + public async Task Test_FanOutEdgeRunner_DeduplicatesAssignedTargetsAsync() + { + // Arrange + TestRunContext runContext = new(); + runContext.ConfigureExecutors([ + new ForwardMessageExecutor("executor1"), + new ForwardMessageExecutor("executor2"), + new ForwardMessageExecutor("executor3") + ]); + + // The assigner names executor3 twice, and names it before executor2. + FanOutEdgeData edgeData = new("executor1", ["executor2", "executor3"], new EdgeId(0), (message, count) => [1, 0, 1]); + FanOutEdgeRunner runner = new(runContext, edgeData); + + // Act + DeliveryMapping? mapping = await runner.ChaseEdgeAsync(new("test", "executor1"), stepTracer: null, CancellationToken.None); + + // Assert + Assert.NotNull(mapping); + Assert.Equal(["executor3", "executor2"], mapping.Deliveries.Select(delivery => delivery.TargetId)); + } + + [Fact] + public async Task Test_FanOutEdgeRunner_DeduplicatesRepeatedSinksAsync() + { + // Arrange + TestRunContext runContext = new(); + runContext.ConfigureExecutors([ + new ForwardMessageExecutor("executor1"), + new ForwardMessageExecutor("executor2") + ]); + + // No assigner, and the target list names executor2 twice. + FanOutEdgeData edgeData = new("executor1", ["executor2", "executor2"], new EdgeId(0)); + FanOutEdgeRunner runner = new(runContext, edgeData); + + // Act + DeliveryMapping? mapping = await runner.ChaseEdgeAsync(new("test", "executor1"), stepTracer: null, CancellationToken.None); + + // Assert + Assert.NotNull(mapping); + Assert.Equal(["executor2"], mapping.Deliveries.Select(delivery => delivery.TargetId)); + } + + [Fact] + public async Task Test_FanOutEdge_TargetAssignedTwiceRunsOnceAsync() + { + // Arrange + int billingCalls = 0; + int techCalls = 0; + + ForwardMessageExecutor triage = new("triage"); + Func billingHandler = _ => + { + Interlocked.Increment(ref billingCalls); + return default; + }; + Func techHandler = _ => + { + Interlocked.Increment(ref techCalls); + return default; + }; + ExecutorBinding billing = billingHandler.BindAsExecutor("billing"); + ExecutorBinding tech = techHandler.BindAsExecutor("tech"); + + // The selector names the billing executor twice, as when two tags on a message route to it. + Workflow workflow = new WorkflowBuilder(triage) + .AddFanOutEdge(triage, [billing, tech], (message, count) => [0, 0]) + .Build(); + + // Act + await using Run run = await InProcessExecution.RunAsync(workflow, "charged twice, want my money back"); + + // Assert + Assert.Equal(1, billingCalls); + Assert.Equal(0, techCalls); + } + [Fact] public async Task Test_FanInEdgeRunnerAsync() {