mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.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
This commit is contained in:
Jacob Alber
committed by
GitHub
Unverified
GitHub
parent
0c28935382
commit
e1147273da
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -12,12 +12,12 @@ internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData
|
||||
|
||||
public FanInEdgeState CreateState() => new(this.EdgeData);
|
||||
|
||||
public async ValueTask<object?> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer)
|
||||
public async ValueTask<IEnumerable<object?>> 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<Task<object?>> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string>("executor1");
|
||||
runContext.Executors["executor2"] = new ForwardMessageExecutor<string>("executor2");
|
||||
runContext.Executors["executor3"] = new ForwardMessageExecutor<string>("executor3");
|
||||
|
||||
Dictionary<string, HashSet<Edge>> 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"]));
|
||||
}
|
||||
}
|
||||
@@ -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<object?, bool>? 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<string>("executor1");
|
||||
runContext.Executors["executor2"] = new ForwardMessageExecutor<string>("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<string>("executor1");
|
||||
runContext.Executors["executor2"] = new ForwardMessageExecutor<string>("executor2");
|
||||
runContext.Executors["executor3"] = new ForwardMessageExecutor<string>("executor3");
|
||||
|
||||
Func<object?, int, IEnumerable<int>>? 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<string> 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<string>("executor1");
|
||||
runContext.Executors["executor2"] = new ForwardMessageExecutor<string>("executor2");
|
||||
runContext.Executors["executor3"] = new ForwardMessageExecutor<string>("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"]));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
namespace Microsoft.Agents.Workflows.UnitTests;
|
||||
|
||||
internal sealed class ForwardMessageExecutor<TMessage> : Executor where TMessage : notnull
|
||||
internal sealed class ForwardMessageExecutor<TMessage>(string? id = null) : Executor(id) where TMessage : notnull
|
||||
{
|
||||
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
|
||||
{
|
||||
|
||||
@@ -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<string, List<MessageEnvelope>> queuedMessages, params (string expectedSender, List<string> expectedMessages)[] expectedForwards)
|
||||
{
|
||||
queuedMessages.Should().HaveCount(expectedForwards.Length);
|
||||
|
||||
IEnumerable<Action<string>> perSenderValidations = expectedForwards.Select(
|
||||
(forward) =>
|
||||
{
|
||||
(string expectedSender, List<string> expectedMessages) = forward;
|
||||
|
||||
return (Action<string>)(
|
||||
(string senderId) =>
|
||||
{
|
||||
senderId.Should().Be(expectedSender);
|
||||
queuedMessages[senderId].Should().HaveCount(expectedMessages.Count);
|
||||
|
||||
Action<MessageEnvelope>[] validations
|
||||
= expectedMessages.Select(message => (Action<MessageEnvelope>)(envelope => envelope!.Message.Should().Be(message)))
|
||||
.ToArray();
|
||||
|
||||
Assert.Collection(queuedMessages[senderId], validations);
|
||||
});
|
||||
}
|
||||
);
|
||||
|
||||
Assert.Collection(queuedMessages.Keys, perSenderValidations.ToArray());
|
||||
}
|
||||
}
|
||||
@@ -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<string, List<MessageEnvelope>> queuedMessages, params (string expectedSender, List<string> expectedMessages)[] expectedForwards)
|
||||
{
|
||||
queuedMessages.Should().HaveCount(expectedForwards.Length);
|
||||
|
||||
IEnumerable<Action<string>> perSenderValidations = expectedForwards.Select(
|
||||
(forward) =>
|
||||
{
|
||||
(string expectedSender, List<string> expectedMessages) = forward;
|
||||
|
||||
return (Action<string>)(
|
||||
(string senderId) =>
|
||||
{
|
||||
senderId.Should().Be(expectedSender);
|
||||
queuedMessages[senderId].Should().HaveCount(expectedMessages.Count);
|
||||
|
||||
Action<MessageEnvelope>[] validations
|
||||
= expectedMessages.Select(message => (Action<MessageEnvelope>)(envelope => envelope!.Message.Should().Be(message)))
|
||||
.ToArray();
|
||||
|
||||
Assert.Collection(queuedMessages[senderId], validations);
|
||||
});
|
||||
}
|
||||
);
|
||||
|
||||
Assert.Collection(queuedMessages.Keys, perSenderValidations.ToArray());
|
||||
}
|
||||
}
|
||||
@@ -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<T>(string key, T? value, string? scopeName = null)
|
||||
=> default;
|
||||
|
||||
public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null)
|
||||
=> new(default(T?));
|
||||
|
||||
public ValueTask<HashSet<string>> ReadStateKeysAsync(string? scopeName = null)
|
||||
=> new([]);
|
||||
|
||||
public ValueTask SendMessageAsync(object message, string? targetId = null)
|
||||
=> runnerContext.SendMessageAsync(executorId, message, targetId);
|
||||
}
|
||||
|
||||
public List<WorkflowEvent> 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<ExternalRequest> ExternalRequests { get; } = new();
|
||||
public ValueTask PostAsync(ExternalRequest request)
|
||||
{
|
||||
this.ExternalRequests.Add(request);
|
||||
return default;
|
||||
}
|
||||
|
||||
internal Dictionary<string, List<MessageEnvelope>> QueuedMessages { get; } = new();
|
||||
public ValueTask SendMessageAsync(string sourceId, object message, string? targetId = null)
|
||||
{
|
||||
if (!this.QueuedMessages.TryGetValue(sourceId, out List<MessageEnvelope>? deliveryQueue))
|
||||
{
|
||||
this.QueuedMessages[sourceId] = deliveryQueue = new();
|
||||
}
|
||||
|
||||
deliveryQueue.Add(new(message, targetId: targetId));
|
||||
return default;
|
||||
}
|
||||
|
||||
StepContext IRunnerContext.Advance()
|
||||
{
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
|
||||
public Dictionary<string, Executor> Executors { get; } = new();
|
||||
|
||||
ValueTask<Executor> IRunnerContext.EnsureExecutorAsync(string executorId, IStepTracer? tracer)
|
||||
{
|
||||
return new(this.Executors[executorId]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user