diff --git a/dotnet/src/Microsoft.Agents.Workflows/AgentWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.Workflows/AgentWorkflowBuilder.cs index 85b9e1cef0..2983861c49 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/AgentWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/AgentWorkflowBuilder.cs @@ -145,7 +145,7 @@ public static partial class AgentWorkflowBuilder /// /// Executor that runs the agent and forwards all messages, input and output, to the next executor. /// - private sealed class AgentRunStreamingExecutor(AIAgent agent, bool includeInputInOutput) : Executor(GetDescriptiveIdFromAgent(agent)) + private sealed class AgentRunStreamingExecutor(AIAgent agent, bool includeInputInOutput) : Executor(GetDescriptiveIdFromAgent(agent)), IResettableExecutor { private readonly List _pendingMessages = []; @@ -185,20 +185,28 @@ public static partial class AgentWorkflowBuilder await context.SendMessageAsync(messages).ConfigureAwait(false); await context.SendMessageAsync(token).ConfigureAwait(false); }); + + public ValueTask ResetAsync() + { + this._pendingMessages.Clear(); + return default; + } } /// /// Provides an executor that batches received chat messages that it then publishes as the final result /// when receiving a . /// - private sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages") + private sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages"), IResettableExecutor { protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default) => context.YieldOutputAsync(messages); + + ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync(); } /// Executor that forwards all messages. - private sealed class ChatForwardingExecutor(string id) : Executor(id) + private sealed class ChatForwardingExecutor(string id) : Executor(id), IResettableExecutor { protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder @@ -206,23 +214,27 @@ public static partial class AgentWorkflowBuilder .AddHandler((message, context) => context.SendMessageAsync(message)) .AddHandler>((messages, context) => context.SendMessageAsync(messages)) .AddHandler((turnToken, context) => context.SendMessageAsync(turnToken)); + + public ValueTask ResetAsync() => default; } /// /// Provides an executor that batches received chat messages that it then releases when /// receiving a . /// - private sealed class BatchChatMessagesToListExecutor(string id) : ChatProtocolExecutor(id) + private sealed class BatchChatMessagesToListExecutor(string id) : ChatProtocolExecutor(id), IResettableExecutor { protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default) => context.SendMessageAsync(messages); + + ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync(); } /// /// Provides an executor that accepts the output messages from each of the concurrent agents /// and produces a result list containing the last message from each. /// - private sealed class ConcurrentEndExecutor : Executor + private sealed class ConcurrentEndExecutor : Executor, IResettableExecutor { private readonly int _expectedInputs; private readonly Func>, List> _aggregator; @@ -238,6 +250,12 @@ public static partial class AgentWorkflowBuilder this._remaining = expectedInputs; } + private void Reset() + { + this._allResults = new List>(this._expectedInputs); + this._remaining = this._expectedInputs; + } + protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder.AddHandler>(async (messages, context) => { @@ -259,6 +277,12 @@ public static partial class AgentWorkflowBuilder await context.YieldOutputAsync(this._aggregator(results)).ConfigureAwait(false); } }); + + public ValueTask ResetAsync() + { + this.Reset(); + return default; + } } /// @@ -428,7 +452,7 @@ public static partial class AgentWorkflowBuilder } /// Executor used at the start of a handoffs workflow to accumulate messages and emit them as HandoffState upon receiving a turn token. - private sealed class StartHandoffsExecutor() : Executor("HandoffStart") + private sealed class StartHandoffsExecutor() : Executor("HandoffStart"), IResettableExecutor { private readonly List _pendingMessages = []; @@ -445,20 +469,28 @@ public static partial class AgentWorkflowBuilder this._pendingMessages.Clear(); await context.SendMessageAsync(new HandoffState(token, null, messages)).ConfigureAwait(false); }); + + public ValueTask ResetAsync() + { + this._pendingMessages.Clear(); + return default; + } } /// Executor used at the end of a handoff workflow to raise a final completed event. - private sealed class EndHandoffsExecutor() : Executor("HandoffEnd") + private sealed class EndHandoffsExecutor() : Executor("HandoffEnd"), IResettableExecutor { protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder.AddHandler((handoff, context) => context.YieldOutputAsync(handoff.Messages)); + + public ValueTask ResetAsync() => default; } /// Executor used to represent an agent in a handoffs workflow, responding to events. private sealed class HandoffAgentExecutor( AIAgent agent, - string? handoffInstructions) : Executor(GetDescriptiveIdFromAgent(agent)) + string? handoffInstructions) : Executor(GetDescriptiveIdFromAgent(agent)), IResettableExecutor { private static readonly JsonElement s_handoffSchema = AIFunctionFactory.Create( ([Description("The reason for the handoff")] string? reasonForHandoff) => { }).JsonSchema; @@ -548,6 +580,8 @@ public static partial class AgentWorkflowBuilder } } }); + + public ValueTask ResetAsync() => default; } private record class HandoffState( @@ -746,7 +780,7 @@ public static partial class AgentWorkflowBuilder return builder.WithOutputFrom(host).Build(); } - private sealed class GroupChatHost(AIAgent[] agents, Dictionary agentMap, Func, GroupChatManager> managerFactory) : Executor("GroupChatHost") + private sealed class GroupChatHost(AIAgent[] agents, Dictionary agentMap, Func, GroupChatManager> managerFactory) : Executor("GroupChatHost"), IResettableExecutor { private readonly AIAgent[] _agents = agents; private readonly Dictionary _agentMap = agentMap; @@ -786,6 +820,14 @@ public static partial class AgentWorkflowBuilder this._manager = null; await context.YieldOutputAsync(messages).ConfigureAwait(false); }); + + public ValueTask ResetAsync() + { + this._pendingMessages.Clear(); + this._manager = null; + + return default; + } } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/ChatProtocolExecutor.cs b/dotnet/src/Microsoft.Agents.Workflows/ChatProtocolExecutor.cs index 3c86865d36..4582fd73d3 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/ChatProtocolExecutor.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/ChatProtocolExecutor.cs @@ -18,6 +18,14 @@ internal abstract class ChatProtocolExecutor(string id, ChatProtocolExecutorOpti private List _pendingMessages = []; private readonly ChatRole? _stringMessageChatRole = options?.StringMessageChatRole; + // Note that we explicitly do not implement IResettableExecutor here, as we want to allow derived classes to + // implement it if they want to be resettable, but do not want to opt them into it. + protected ValueTask ResetAsync() + { + this._pendingMessages = []; + return default; + } + protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) { if (this._stringMessageChatRole.HasValue) diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableMessageEnvelope.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableMessageEnvelope.cs index c2e73de73d..3df7a5479a 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableMessageEnvelope.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableMessageEnvelope.cs @@ -9,13 +9,15 @@ internal sealed class PortableMessageEnvelope { public TypeId MessageType { get; } public PortableValue Message { get; } + public ExecutorIdentity Source { get; } public string? TargetId { get; } [JsonConstructor] - internal PortableMessageEnvelope(TypeId messageType, PortableValue message, string? targetId) + internal PortableMessageEnvelope(TypeId messageType, ExecutorIdentity source, PortableValue message, string? targetId) { this.MessageType = messageType; this.Message = message; + this.Source = source; this.TargetId = targetId; } @@ -28,6 +30,6 @@ internal sealed class PortableMessageEnvelope public MessageEnvelope ToMessageEnvelope() { - return new MessageEnvelope(this.Message, this.MessageType, this.TargetId); + return new MessageEnvelope(this.Message, this.Source, this.MessageType, this.TargetId); } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableValueConverter.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableValueConverter.cs index 512a459216..12144bb4ba 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableValueConverter.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/PortableValueConverter.cs @@ -13,8 +13,8 @@ namespace Microsoft.Agents.Workflows.Checkpointing; /// at the time of initial deserialization, e.g. user-defined state types. /// /// This operates in conjuction with and to abstract -/// away the speicfics of a given serialization format in favor of and -/// . +/// away the speicfics of a given serialization format in favor of and +/// and related methods. /// /// internal sealed class PortableValueConverter(JsonMarshaller marshaller) : JsonConverter diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/DeliveryMapping.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/DeliveryMapping.cs new file mode 100644 index 0000000000..ecfaf0d79c --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/DeliveryMapping.cs @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Linq; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.Workflows.Execution; + +internal sealed class DeliveryMapping +{ + private readonly IEnumerable _envelopes; + private readonly IEnumerable _targets; + + public DeliveryMapping(IEnumerable envelopes, IEnumerable targets) + { + this._envelopes = Throw.IfNull(envelopes); + this._targets = Throw.IfNull(targets); + } + + public DeliveryMapping(MessageEnvelope envelope, Executor target) : this([envelope], [target]) { } + public DeliveryMapping(MessageEnvelope envelope, IEnumerable targets) : this([envelope], targets) { } + public DeliveryMapping(IEnumerable envelopes, Executor target) : this(envelopes, [target]) { } + + public IEnumerable Deliveries => from target in this._targets + from envelope in this._envelopes + select new MessageDelivery(envelope, target); + + public void MapInto(StepContext nextStep) + { + foreach (Executor target in this._targets) + { + nextStep.MessagesFor(target.Id).AddRange(this._envelopes.Select(envelope => envelope)); + } + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs index e39ebdeedf..e28fa9bdde 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs @@ -1,6 +1,5 @@ // Copyright (c) Microsoft. All rights reserved. -using System.Collections.Generic; using System.Threading.Tasks; namespace Microsoft.Agents.Workflows.Execution; @@ -13,26 +12,25 @@ internal sealed class DirectEdgeRunner(IRunnerContext runContext, DirectEdgeData private async ValueTask FindRouterAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer) .ConfigureAwait(false); - public async ValueTask> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer) + protected internal override async ValueTask ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer) { if (envelope.TargetId is not null && this.EdgeData.SinkId != envelope.TargetId) { - return []; + return null; } object message = envelope.Message; if (this.EdgeData.Condition is not null && !this.EdgeData.Condition(message)) { - return []; + return null; } - Executor target = await this.FindRouterAsync(tracer).ConfigureAwait(false); + Executor target = await this.FindRouterAsync(stepTracer).ConfigureAwait(false); if (target.CanHandle(envelope.MessageType)) { - tracer?.TraceActivated(target.Id); - return [await target.ExecuteAsync(message, envelope.MessageType, this.WorkflowContext).ConfigureAwait(false)]; + return new DeliveryMapping(envelope, target); } - return []; + return null; } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs index cc2c7309a0..6c0e111282 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs @@ -10,12 +10,23 @@ namespace Microsoft.Agents.Workflows.Execution; internal sealed class EdgeMap { - private readonly Dictionary _edgeRunners = []; - private readonly Dictionary _fanInState = []; + private readonly Dictionary _edgeRunners = []; + private readonly Dictionary _statefulRunners = new(); private readonly Dictionary _portEdgeRunners; + private readonly InputEdgeRunner _inputRunner; private readonly IStepTracer? _stepTracer; + public EdgeMap(IRunnerContext runContext, + Workflow workflow, + IStepTracer? stepTracer) + : this(runContext, + workflow.Edges, + workflow.Ports.Values, + workflow.StartExecutorId, + stepTracer) + { } + public EdgeMap(IRunnerContext runContext, Dictionary> workflowEdges, IEnumerable workflowPorts, @@ -24,7 +35,7 @@ internal sealed class EdgeMap { foreach (Edge edge in workflowEdges.Values.SelectMany(e => e)) { - object edgeRunner = edge.Kind switch + EdgeRunner edgeRunner = edge.Kind switch { EdgeKind.Direct => new DirectEdgeRunner(runContext, edge.DirectEdgeData!), EdgeKind.FanOut => new FanOutEdgeRunner(runContext, edge.FanOutEdgeData!), @@ -32,12 +43,12 @@ internal sealed class EdgeMap _ => throw new NotSupportedException($"Unsupported edge type: {edge.Kind}") }; - if (edgeRunner is FanInEdgeRunner fanInRunner) - { - this._fanInState[edge.Data.Id] = fanInRunner.CreateState(); - } - this._edgeRunners[edge.Data.Id] = edgeRunner; + + if (edgeRunner is IStatefulEdgeRunner statefulRunner) + { + this._statefulRunners[edge.Data.Id] = statefulRunner; + } } this._portEdgeRunners = workflowPorts.ToDictionary( @@ -49,103 +60,52 @@ internal sealed class EdgeMap this._stepTracer = stepTracer; } - public async ValueTask> InvokeEdgeAsync(Edge edge, string sourceId, MessageEnvelope message) + public ValueTask PrepareDeliveryForEdgeAsync(Edge edge, MessageEnvelope message) { EdgeId id = edge.Data.Id; - if (!this._edgeRunners.TryGetValue(id, out object? edgeRunner)) + if (!this._edgeRunners.TryGetValue(id, out EdgeRunner? edgeRunner)) { throw new InvalidOperationException($"Edge {edge} not found in the edge map."); } - IEnumerable edgeResults; - switch (edge.Kind) - { - // We know the corresponding EdgeRunner type given the FlowEdge EdgeType, as - // established in the EdgeMap() ctor; this avoid doing an as-cast inside of - // the depths of the message delivery loop for every edges (multiplicity N, - // in FanIn/Out cases) - // TODO: Once we have a fixed interface, if it is reasonably generalizable - // between the Runners, we can normalize it behind an IFace. - case EdgeKind.Direct: - { - DirectEdgeRunner runner = (DirectEdgeRunner)edgeRunner; - edgeResults = await runner.ChaseAsync(message, this._stepTracer).ConfigureAwait(false); - break; - } - - case EdgeKind.FanOut: - { - FanOutEdgeRunner runner = (FanOutEdgeRunner)edgeRunner; - edgeResults = await runner.ChaseAsync(message, this._stepTracer).ConfigureAwait(false); - break; - } - - case EdgeKind.FanIn: - { - FanInEdgeState state = this._fanInState[id]; - FanInEdgeRunner runner = (FanInEdgeRunner)edgeRunner; - edgeResults = [await runner.ChaseAsync(sourceId, message, state, this._stepTracer).ConfigureAwait(false)]; - break; - } - - default: - throw new InvalidOperationException("Unknown edge type"); - - } - - return edgeResults; + return edgeRunner.ChaseEdgeAsync(message, this._stepTracer); } - // TODO: Should we promote Input to a true "FlowEdge" type? - public async ValueTask> InvokeInputAsync(MessageEnvelope envelope) + public ValueTask PrepareDeliveryForInputAsync(MessageEnvelope message) { - return [await this._inputRunner.ChaseAsync(envelope, this._stepTracer).ConfigureAwait(false)]; + return this._inputRunner.ChaseEdgeAsync(message, this._stepTracer); } - public async ValueTask> InvokeResponseAsync(ExternalResponse response) + public ValueTask PrepareDeliveryForResponseAsync(ExternalResponse response) { if (!this._portEdgeRunners.TryGetValue(response.PortInfo.PortId, out InputEdgeRunner? portRunner)) { throw new InvalidOperationException($"Port {response.PortInfo.PortId} not found in the edge map."); } - return [await portRunner.ChaseAsync(new MessageEnvelope(response), this._stepTracer).ConfigureAwait(false)]; + return portRunner.ChaseEdgeAsync(new MessageEnvelope(response, ExecutorIdentity.None), this._stepTracer); } - internal ValueTask> ExportStateAsync() + internal async ValueTask> ExportStateAsync() { Dictionary exportedStates = []; - // Right now there is only fan-in state - foreach (EdgeId id in this._fanInState.Keys) + foreach (EdgeId id in this._statefulRunners.Keys) { - FanInEdgeState state = this._fanInState[id]; - exportedStates[id] = new PortableValue(state); + exportedStates[id] = await this._statefulRunners[id].ExportStateAsync().ConfigureAwait(false); } - return new(exportedStates); + return exportedStates; } - internal ValueTask ImportStateAsync(Checkpoint checkpoint) + internal async ValueTask ImportStateAsync(Checkpoint checkpoint) { Dictionary importedState = checkpoint.EdgeStateData; - this._fanInState.Clear(); foreach (EdgeId id in importedState.Keys) { PortableValue exportedState = importedState[id]; - - FanInEdgeState? fanInState = exportedState.As(); - if (fanInState is not null) - { - this._fanInState[id] = fanInState; - } - else - { - throw new InvalidOperationException($"Unsupported exported state type: {exportedState.GetType()}; {id}"); - } + await this._statefulRunners[id].ImportStateAsync(exportedState).ConfigureAwait(false); } - - return default; } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeRunner.cs index 8871f9d8bb..c3962bc9ad 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeRunner.cs @@ -1,11 +1,24 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Threading.Tasks; using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.Workflows.Execution; +internal interface IStatefulEdgeRunner +{ + ValueTask ExportStateAsync(); + ValueTask ImportStateAsync(PortableValue state); +} + +internal abstract class EdgeRunner +{ + // TODO: Can this be sync? + protected internal abstract ValueTask ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer); +} + internal abstract class EdgeRunner( - IRunnerContext runContext, TEdgeData edgeData) + IRunnerContext runContext, TEdgeData edgeData) : EdgeRunner() { protected IRunnerContext RunContext { get; } = Throw.IfNull(runContext); protected TEdgeData EdgeData { get; } = Throw.IfNull(edgeData); diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs index 386c580619..7e350dc13c 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs @@ -1,58 +1,56 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Generic; using System.Diagnostics; +using System.Linq; using System.Threading.Tasks; namespace Microsoft.Agents.Workflows.Execution; internal sealed class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData) : - EdgeRunner(runContext, edgeData) + EdgeRunner(runContext, edgeData), + IStatefulEdgeRunner { - private IWorkflowContext BoundContext { get; } = runContext.Bind(edgeData.SinkId); + private FanInEdgeState _state = new(edgeData); - public FanInEdgeState CreateState() => new(this.EdgeData); - - public ValueTask> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer) + protected internal override async ValueTask ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer) { + Debug.Assert(!envelope.IsExternal, "FanIn edges should never be chased from external input"); + if (envelope.TargetId is not null && this.EdgeData.SinkId != envelope.TargetId) { - // This message is not for us. - return new([]); + return null; } - IEnumerable? releasedMessages = state.ProcessMessage(sourceId, envelope); + // source.Id is guaranteed to be non-null here because source is not None. + IEnumerable? releasedMessages = this._state.ProcessMessage(envelope.SourceId, envelope); if (releasedMessages is null) { // Not ready to process yet. - return new([]); + return null; } - return this.ForwardReleasedMessagesAsync(releasedMessages, tracer); - } - - private async ValueTask> ForwardReleasedMessagesAsync(IEnumerable releasedMessages, IStepTracer? tracer) - { - Executor target = await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer) + // TODO: Filter messages based on accepted input types? + Executor target = await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, stepTracer) .ConfigureAwait(false); - List> messageTasks = []; + return new DeliveryMapping(releasedMessages.Where(envelope => target.CanHandle(envelope.MessageType)), target); + } - foreach (MessageEnvelope releasedEnvelope in releasedMessages) + public ValueTask ExportStateAsync() + { + return new(new PortableValue(this._state)); + } + + public ValueTask ImportStateAsync(PortableValue state) + { + if (state.Is(out FanInEdgeState? importedState)) { - object message = releasedEnvelope.Message; - Debug.Assert(message is PortableValue, "It should not be possible to get messages released without roundtripping them through" + - "PortableValue via PortableMessageEnvelope."); - - PortableValue portable = message as PortableValue ?? new PortableValue(releasedEnvelope.MessageType, message); - - if (target.CanHandle(portable.TypeId)) - { - tracer?.TraceActivated(target.Id); - messageTasks.Add(target.ExecuteAsync(portable, releasedEnvelope.MessageType, this.BoundContext).AsTask()); - } + this._state = importedState; + return default; } - return await Task.WhenAll(messageTasks.ToArray()).ConfigureAwait(false); + throw new InvalidOperationException($"Unsupported exported state type: {state.GetType()}; {this.EdgeData.Id}"); } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs index 3613fcee05..1e0af8c60c 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs @@ -14,38 +14,28 @@ internal sealed class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData sinkId => sinkId, runContext.Bind); - public async ValueTask> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer) + protected internal override async ValueTask ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer) { object message = envelope.Message; - List targets = + IEnumerable targetIds = this.EdgeData.EdgeAssigner is null ? this.EdgeData.SinkIds : this.EdgeData.EdgeAssigner(message, this.BoundContexts.Count) - .Select(i => this.EdgeData.SinkIds[i]).ToList(); + .Select(i => this.EdgeData.SinkIds[i]); - IEnumerable filteredTargets = - envelope.TargetId is not null - ? targets.Where(IsValidTarget) - : targets; + Executor[] result = await Task.WhenAll(targetIds.Where(IsValidTarget) + .Select(tid => this.RunContext.EnsureExecutorAsync(tid, stepTracer) + .AsTask())) + .ConfigureAwait(false); - object?[] result = await Task.WhenAll(filteredTargets.Select(ProcessTargetAsync)).ConfigureAwait(false); - return result.Where(r => r is not null); - - async Task ProcessTargetAsync(string targetId) + if (result.Length == 0) { - Executor executor = await this.RunContext.EnsureExecutorAsync(targetId, tracer) - .ConfigureAwait(false); - - if (executor.CanHandle(message.GetType())) - { - tracer?.TraceActivated(executor.Id); - return await executor.ExecuteAsync(message, envelope.MessageType, this.BoundContexts[targetId]) - .ConfigureAwait(false); - } - return null; } + IEnumerable validTargets = result.Where(t => t.CanHandle(envelope.MessageType)); + return new DeliveryMapping(envelope, validTargets); + bool IsValidTarget(string targetId) { return envelope.TargetId is null || targetId == envelope.TargetId; diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs index 6311c8b914..cfe45494af 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs @@ -9,7 +9,7 @@ internal interface IRunnerContext : IExternalRequestSink ValueTask AddEventAsync(WorkflowEvent workflowEvent); ValueTask SendMessageAsync(string sourceId, object message, string? targetId = null); - StepContext Advance(); + ValueTask AdvanceAsync(); IWorkflowContext Bind(string executorId); ValueTask EnsureExecutorAsync(string executorId, IStepTracer? tracer); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/ISuperStepRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/ISuperStepRunner.cs index d56d10e602..858114659b 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/ISuperStepRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/ISuperStepRunner.cs @@ -19,4 +19,6 @@ internal interface ISuperStepRunner event EventHandler? WorkflowEvent; ValueTask RunSuperStepAsync(CancellationToken cancellation); + + ValueTask RequestEndRunAsync(); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs index e78129c912..4923a7ea4a 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs @@ -19,21 +19,17 @@ internal sealed class InputEdgeRunner(IRunnerContext runContext, string sinkId) return new InputEdgeRunner(runContext, port.Id); } - private async ValueTask FindExecutorAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false); - - public async ValueTask ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer) + protected internal override async ValueTask ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer) { - Executor target = await this.FindExecutorAsync(tracer).ConfigureAwait(false); + Debug.Assert(envelope.IsExternal, "Input edges should only be chased from external input"); + Executor target = await this.FindExecutorAsync(stepTracer).ConfigureAwait(false); if (target.CanHandle(envelope.MessageType)) { - tracer?.TraceActivated(target.Id); - return await target.ExecuteAsync(envelope.Message, envelope.MessageType, this.WorkflowContext) - .ConfigureAwait(false); + return new DeliveryMapping(envelope, target); } - // TODO: Throw instead? / Log - Debug.WriteLine($"Executor {target.Id} cannot handle message of type {envelope.MessageType.TypeName}. Dropping."); - return null; } + + private async ValueTask FindExecutorAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageDelivery.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageDelivery.cs new file mode 100644 index 0000000000..ff8bd46ad9 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageDelivery.cs @@ -0,0 +1,28 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json.Serialization; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.Workflows.Execution; + +internal sealed class MessageDelivery +{ + [JsonConstructor] + internal MessageDelivery(MessageEnvelope envelope, string targetId) + { + this.Envelope = Throw.IfNull(envelope); + this.TargetId = Throw.IfNull(targetId); + } + + internal MessageDelivery(MessageEnvelope envelope, Executor target) + : this(envelope, target.Id) + { + this.TargetCache = Throw.IfNull(target); + } + + public string TargetId { get; } + public MessageEnvelope Envelope { get; } + + [JsonIgnore] + internal Executor? TargetCache { get; set; } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageEnvelope.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageEnvelope.cs index aac534d799..aa3734abcb 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageEnvelope.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/MessageEnvelope.cs @@ -1,18 +1,25 @@ // Copyright (c) Microsoft. All rights reserved. using System; +using System.Diagnostics.CodeAnalysis; using Microsoft.Agents.Workflows.Checkpointing; namespace Microsoft.Agents.Workflows.Execution; -internal sealed class MessageEnvelope(object message, TypeId? declaredType = null, string? targetId = null) +internal sealed class MessageEnvelope(object message, ExecutorIdentity source, TypeId? declaredType = null, string? targetId = null) { public TypeId MessageType => declaredType ?? new(message.GetType()); public object Message => message; + public ExecutorIdentity Source => source; public string? TargetId => targetId; - internal MessageEnvelope(object message, Type declaredType, string? targetId = null) - : this(message, new TypeId(declaredType), targetId) + [MemberNotNullWhen(false, nameof(SourceId))] + public bool IsExternal => this.Source == ExecutorIdentity.None; + + public string? SourceId => this.Source.Id; + + internal MessageEnvelope(object message, ExecutorIdentity source, Type declaredType, string? targetId = null) + : this(message, source, new TypeId(declaredType), targetId) { if (!declaredType.IsAssignableFrom(message.GetType())) { diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/RunnerStateData.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/RunnerStateData.cs index c29fac8557..114d1a6055 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/RunnerStateData.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/RunnerStateData.cs @@ -5,9 +5,9 @@ using Microsoft.Agents.Workflows.Checkpointing; namespace Microsoft.Agents.Workflows.Execution; -internal sealed class RunnerStateData(HashSet instantiatedExecutors, Dictionary> queuedMessages, List outstandingRequests) +internal sealed class RunnerStateData(HashSet instantiatedExecutors, Dictionary> queuedMessages, List outstandingRequests) { public HashSet InstantiatedExecutors { get; } = instantiatedExecutors; - public Dictionary> QueuedMessages { get; } = queuedMessages; + public Dictionary> QueuedMessages { get; } = queuedMessages; public List OutstandingRequests { get; } = outstandingRequests; } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs index 5ebdb9af52..86d5806f1f 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs @@ -8,15 +8,15 @@ namespace Microsoft.Agents.Workflows.Execution; internal sealed class StepContext { - public Dictionary> QueuedMessages { get; } = []; + public Dictionary> QueuedMessages { get; } = []; public bool HasMessages => this.QueuedMessages.Values.Any(messageList => messageList.Count > 0); - public List MessagesFor(string? executorId) + public List MessagesFor(string target) { - if (!this.QueuedMessages.TryGetValue(executorId, out var messages)) + if (!this.QueuedMessages.TryGetValue(target, out var messages)) { - this.QueuedMessages[executorId] = messages = []; + this.QueuedMessages[target] = messages = []; } return messages; @@ -24,7 +24,7 @@ internal sealed class StepContext // TODO: Create a MessageEnvelope class that extends from the ExportedState object (with appropriate rename) to avoid // unnecessary wrapping and unwrapping of messages during checkpointing. - internal Dictionary> ExportMessages() + internal Dictionary> ExportMessages() { return this.QueuedMessages.Keys.ToDictionary( keySelector: identity => identity, @@ -33,9 +33,9 @@ internal sealed class StepContext ); } - internal void ImportMessages(Dictionary> messages) + internal void ImportMessages(Dictionary> messages) { - foreach (ExecutorIdentity identity in messages.Keys) + foreach (string identity in messages.Keys) { this.QueuedMessages[identity] = messages[identity].ConvertAll(UnwrapExportedState); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/ExecutorRegistration.cs b/dotnet/src/Microsoft.Agents.Workflows/ExecutorRegistration.cs index 1b40c5792d..07676c8d85 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/ExecutorRegistration.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/ExecutorRegistration.cs @@ -13,6 +13,35 @@ internal sealed class ExecutorRegistration(string id, Type executorType, Executo public string Id { get; } = Throw.IfNullOrEmpty(id); public Type ExecutorType { get; } = Throw.IfNull(executorType); public ExecutorFactoryF ProviderAsync { get; } = Throw.IfNull(provider); + public bool IsNotExecutorInstance { get; } = rawData is not Executor; + public bool IsUnresettableSharedInstance { get; } = rawData is Executor && rawData is not IResettableExecutor; + + internal async ValueTask TryResetAsync() + { + if (this.IsUnresettableSharedInstance) + { + return false; + } + + // If this is not an executor instance, this is a factory, and the expectation is that the factory will + // create separate instances of executors. + if (this.IsNotExecutorInstance) + { + return true; + } + + // Technically we definitely know this is true, since if rawData is an Executor, if it was not resettable + // then we would have returned in the first condition, and if rawData is not an Executor, we would have + // returned in the second condition. That only leaves the possibility of rawData is Executor and also + // IResettableExecutor. + if (this.RawExecutorishData is IResettableExecutor resettableExecutor) + { + await resettableExecutor.ResetAsync().ConfigureAwait(false); + return true; + } + + return false; + } internal object? RawExecutorishData { get; } = rawData; diff --git a/dotnet/src/Microsoft.Agents.Workflows/IResettableExecutor.cs b/dotnet/src/Microsoft.Agents.Workflows/IResettableExecutor.cs new file mode 100644 index 0000000000..35f7760a0c --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/IResettableExecutor.cs @@ -0,0 +1,25 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading.Tasks; + +namespace Microsoft.Agents.Workflows; + +/// +/// Provides a mechanism to return an executor to a 'reset' state, allowing a workflow containing +/// shared instances of it to be resued after a run is disposed. +/// +public interface IResettableExecutor +{ + /// + /// Reset the executor + /// + /// A representing the completion of the reset operation. + ValueTask ResetAsync() +#if NET + { + return default; + } +#else + ; +#endif +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs index 57ef528890..d85b789c40 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs @@ -2,7 +2,6 @@ using System; using System.Collections.Generic; -using System.Diagnostics; using System.IO; using System.Linq; using System.Threading; @@ -23,11 +22,12 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner { public InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? runId = null, params Type[] knownValidInputTypes) { - this.Workflow = Throw.IfNull(workflow); - this.RunContext = new InProcessRunnerContext(workflow); - this.CheckpointManager = checkpointManager; this.RunId = runId ?? Guid.NewGuid().ToString("N"); + this.Workflow = Throw.IfNull(workflow); + this.RunContext = new InProcessRunnerContext(workflow, this.RunId, this.StepTracer); + this.CheckpointManager = checkpointManager; + this._knownValidInputTypes = [.. knownValidInputTypes]; // Initialize the runners for each of the edges, along with the state for edges that @@ -56,40 +56,37 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner return false; } - public async ValueTask EnqueueMessageAsync(T message) + private async ValueTask EnqueueMessageInternalAsync(object message, Type messageType) { + this.RunContext.CheckEnded(); Throw.IfNull(message); + if (message is ExternalResponse response) + { + await this.RunContext.AddExternalResponseAsync(response).ConfigureAwait(false); + } + // Check that the type of the incoming message is compatible with the starting executor's // input type. - if (!await this.IsValidInputTypeAsync(typeof(T)).ConfigureAwait(false)) + if (!await this.IsValidInputTypeAsync(messageType).ConfigureAwait(false)) { return false; } - await this.RunContext.AddExternalMessageAsync(message).ConfigureAwait(false); + await this.RunContext.AddExternalMessageAsync(message, messageType).ConfigureAwait(false); return true; } - public async ValueTask EnqueueMessageAsync(object message) - { - Throw.IfNull(message); + public ValueTask EnqueueMessageAsync(T message) + => this.EnqueueMessageInternalAsync(Throw.IfNull(message), typeof(T)); - // Check that the type of the incoming message is compatible with the starting executor's - // input type. - if (!await this.IsValidInputTypeAsync(message.GetType()).ConfigureAwait(false)) - { - return false; - } - - await this.RunContext.AddExternalMessageUntypedAsync(message).ConfigureAwait(false); - return true; - } + public ValueTask EnqueueMessageAsync(object message) + => this.EnqueueMessageInternalAsync(Throw.IfNull(message), message.GetType()); ValueTask ISuperStepRunner.EnqueueResponseAsync(ExternalResponse response) { // TODO: Check that there exists a corresponding input port? - return this.RunContext.AddExternalMessageAsync(response); + return this.RunContext.AddExternalResponseAsync(response); } private InProcStepTracer StepTracer { get; } = new(); @@ -111,28 +108,9 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner this.WorkflowEvent?.Invoke(this, workflowEvent); } - private ValueTask> RouteExternalMessageAsync(MessageEnvelope envelope) - { - Debug.Assert(envelope.TargetId is null, "External Messages cannot be targeted to a specific executor."); - - object message = envelope.Message; - return message is ExternalResponse response - ? this.CompleteExternalResponseAsync(response) - : this.EdgeMap.InvokeInputAsync(envelope); - } - - private ValueTask> CompleteExternalResponseAsync(ExternalResponse response) - { - if (!this.RunContext.CompleteRequest(response.RequestId)) - { - throw new InvalidOperationException($"No pending request with ID {response.RequestId} found in the workflow context."); - } - - return this.EdgeMap.InvokeResponseAsync(response); - } - public async ValueTask ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); Throw.IfNull(checkpoint); if (this.CheckpointManager is null) { @@ -146,6 +124,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner public async ValueTask StreamAsync(object input, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); await this.EnqueueMessageAsync(input).ConfigureAwait(false); return new StreamingRun(this); @@ -153,6 +132,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner public async ValueTask StreamAsync(TInput input, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); await this.EnqueueMessageAsync(input).ConfigureAwait(false); return new StreamingRun(this); @@ -160,6 +140,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner internal async ValueTask ResumeAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); StreamingRun streamingRun = await this.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false); cancellation.ThrowIfCancellationRequested(); @@ -168,6 +149,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner public async ValueTask RunAsync(object input, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false); cancellation.ThrowIfCancellationRequested(); @@ -176,6 +158,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner public async ValueTask RunAsync(TInput input, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false); cancellation.ThrowIfCancellationRequested(); @@ -189,12 +172,13 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner async ValueTask ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellation) { + this.RunContext.CheckEnded(); if (cancellation.IsCancellationRequested) { return false; } - StepContext currentStep = this.RunContext.Advance(); + StepContext currentStep = await this.RunContext.AdvanceAsync().ConfigureAwait(false); if (currentStep.HasMessages) { @@ -218,32 +202,32 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner } } + private async ValueTask DeliverMessagesAsync(string receiverId, List envelopes) + { + Executor executor = await this.RunContext.EnsureExecutorAsync(receiverId, this.StepTracer).ConfigureAwait(false); + + this.StepTracer.TraceActivated(receiverId); + foreach (MessageEnvelope envelope in envelopes) + { + await executor.ExecuteAsync(envelope.Message, envelope.MessageType, this.RunContext.Bind(receiverId)) + .ConfigureAwait(false); + } + } + private async ValueTask RunSuperstepAsync(StepContext currentStep) { this.RaiseWorkflowEvent(this.StepTracer.Advance(currentStep)); // Deliver the messages and queue the next step - List>> edgeTasks = []; - foreach (ExecutorIdentity sender in currentStep.QueuedMessages.Keys) - { - IEnumerable senderMessages = currentStep.QueuedMessages[sender]; - if (sender.Id is null) - { - edgeTasks.AddRange(senderMessages.Select(envelope => this.RouteExternalMessageAsync(envelope).AsTask())); - } - else if (this.Workflow.Edges.TryGetValue(sender.Id!, out HashSet? outgoingEdges)) - { - foreach (Edge outgoingEdge in outgoingEdges) - { - edgeTasks.AddRange(senderMessages.Select(envelope => this.EdgeMap.InvokeEdgeAsync(outgoingEdge, sender.Id, envelope).AsTask())); - } - } - } + List receiverTasks = + currentStep.QueuedMessages.Keys + .Select(receiverId => this.DeliverMessagesAsync(receiverId, currentStep.MessagesFor(receiverId)).AsTask()) + .ToList(); // TODO: Should we let the user specify that they want strictly turn-based execution of the edges, vs. concurrent? // (Simply substitute a strategy that replaces Task.WhenAll with a loop with an await in the middle. Difficulty is // that we would need to avoid firing the tasks when we call InvokeEdgeAsync, or RouteExternalMessageAsync. - IEnumerable results = (await Task.WhenAll(edgeTasks).ConfigureAwait(false)).SelectMany(r => r); + await Task.WhenAll(receiverTasks).ConfigureAwait(false); // After the message handler invocations, we may have some events to deliver this.EmitPendingEvents(); @@ -257,6 +241,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner private readonly List _checkpoints = []; internal async ValueTask CheckpointAsync(CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); if (this.CheckpointManager is null) { // Always publish the state updates, even in the absence of a CheckpointManager. @@ -286,6 +271,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default) { + this.RunContext.CheckEnded(); Throw.IfNull(checkpointInfo); if (this.CheckpointManager is null) { @@ -316,4 +302,6 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner private bool CheckWorkflowMatch(Checkpoint checkpoint) => checkpoint.Workflow.IsMatch(this.Workflow); + + ValueTask ISuperStepRunner.RequestEndRunAsync() => this.RunContext.EndRunAsync(); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs index 3804fb6a2a..546cd0bbdd 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft. All rights reserved. using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -16,81 +17,157 @@ namespace Microsoft.Agents.Workflows.InProc; internal sealed class InProcessRunnerContext : IRunnerContext { - private StepContext _nextStep = new(); - private readonly Dictionary _executorRegistrations; - private readonly Dictionary _executors = []; - private readonly Dictionary _externalRequests = []; + private int _runEnded; + private readonly string _runId; + private readonly Workflow _workflow; + + private readonly EdgeMap _edgeMap; private readonly OutputFilter _outputFilter; - public InProcessRunnerContext(Workflow workflow, ILogger? logger = null) + private StepContext _nextStep = new(); + + private readonly ConcurrentDictionary> _executors = new(); + private readonly ConcurrentQueue> _queuedExternalDeliveries = new(); + + private readonly Dictionary _externalRequests = new(); + + public InProcessRunnerContext(Workflow workflow, string runId, IStepTracer? stepTracer, ILogger? logger = null) { - this._executorRegistrations = Throw.IfNull(workflow).Registrations; + workflow.TakeOwnership(this); + this._workflow = workflow; + this._runId = runId; + + this._edgeMap = new(this, this._workflow, stepTracer); this._outputFilter = new(workflow); } public async ValueTask EnsureExecutorAsync(string executorId, IStepTracer? tracer) { - if (!this._executors.TryGetValue(executorId, out var executor)) + this.CheckEnded(); + Task executorTask = this._executors.GetOrAdd(executorId, CreateExecutorAsync); + + async Task CreateExecutorAsync(string id) { - if (!this._executorRegistrations.TryGetValue(executorId, out var registration)) + if (!this._workflow.Registrations.TryGetValue(executorId, out var registration)) { throw new InvalidOperationException($"Executor with ID '{executorId}' is not registered."); } - this._executors[executorId] = executor = await registration.ProviderAsync().ConfigureAwait(false); + Executor executor = await registration.ProviderAsync().ConfigureAwait(false); tracer?.TraceActivated(executorId); if (executor is RequestInfoExecutor requestInputExecutor) { requestInputExecutor.AttachRequestSink(this); } + + return executor; } - return executor; + return await executorTask.ConfigureAwait(false); } - public ValueTask AddExternalMessageUntypedAsync(object message) + public ValueTask AddExternalMessageAsync(object message, Type declaredType) { + this.CheckEnded(); Throw.IfNull(message); - this._nextStep.MessagesFor(ExecutorIdentity.None).Add(new MessageEnvelope(message)); + this._queuedExternalDeliveries.Enqueue(PrepareExternalDeliveryAsync); return default; + + async ValueTask PrepareExternalDeliveryAsync() + { + DeliveryMapping? maybeMapping = + await this._edgeMap.PrepareDeliveryForInputAsync(new(message, ExecutorIdentity.None, declaredType)) + .ConfigureAwait(false); + + maybeMapping?.MapInto(this._nextStep); + } } - public ValueTask AddExternalMessageAsync(T message) + public ValueTask AddExternalResponseAsync(ExternalResponse response) { - Throw.IfNull(message); + this.CheckEnded(); + Throw.IfNull(response); - this._nextStep.MessagesFor(ExecutorIdentity.None).Add(new MessageEnvelope(message, declaredType: typeof(T))); + this._queuedExternalDeliveries.Enqueue(PrepareExternalDeliveryAsync); return default; + + async ValueTask PrepareExternalDeliveryAsync() + { + if (!this.CompleteRequest(response.RequestId)) + { + throw new InvalidOperationException($"No pending request with ID {response.RequestId} found in the workflow context."); + } + + DeliveryMapping? maybeMapping = + await this._edgeMap.PrepareDeliveryForResponseAsync(response) + .ConfigureAwait(false); + + maybeMapping?.MapInto(this._nextStep); + } } - public bool NextStepHasActions => this._nextStep.HasMessages; + public bool NextStepHasActions => this._nextStep.HasMessages || !this._queuedExternalDeliveries.IsEmpty; public bool HasUnservicedRequests => this._externalRequests.Count > 0; - public StepContext Advance() => Interlocked.Exchange(ref this._nextStep, new StepContext()); + public async ValueTask AdvanceAsync() + { + this.CheckEnded(); + + while (this._queuedExternalDeliveries.TryDequeue(out var deliveryPrep)) + { + // It's important we do not try to run these in parallel, because they make be modifying + // inner edge state, etc. + await deliveryPrep().ConfigureAwait(false); + } + + return Interlocked.Exchange(ref this._nextStep, new StepContext()); + } public ValueTask AddEventAsync(WorkflowEvent workflowEvent) { + this.CheckEnded(); this.QueuedEvents.Add(workflowEvent); return default; } - public ValueTask SendMessageAsync(string sourceId, object message, string? targetId = null) + public async ValueTask SendMessageAsync(string sourceId, object message, string? targetId = null) { - this._nextStep.MessagesFor(sourceId).Add(new MessageEnvelope(message, targetId: targetId)); - return default; + this.CheckEnded(); + MessageEnvelope envelope = new(message, sourceId, targetId: targetId); + + if (this._workflow.Edges.TryGetValue(sourceId, out HashSet? edges)) + { + foreach (Edge edge in edges) + { + DeliveryMapping? maybeMapping = + await this._edgeMap.PrepareDeliveryForEdgeAsync(edge, envelope) + .ConfigureAwait(false); + + maybeMapping?.MapInto(this._nextStep); + } + } } - public IWorkflowContext Bind(string executorId) => new BoundContext(this, executorId, this._outputFilter); + public IWorkflowContext Bind(string executorId) + { + this.CheckEnded(); + return new BoundContext(this, executorId, this._outputFilter); + } public ValueTask PostAsync(ExternalRequest request) { + this.CheckEnded(); this._externalRequests.Add(request.RequestId, request); return this.AddEventAsync(new RequestInfoEvent(request)); } - public bool CompleteRequest(string requestId) => this._externalRequests.Remove(requestId); + public bool CompleteRequest(string requestId) + { + this.CheckEnded(); + return this._externalRequests.Remove(requestId); + } public readonly List QueuedEvents = []; @@ -103,6 +180,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext public async ValueTask YieldOutputAsync(object output) { + RunnerContext.CheckEnded(); Throw.IfNull(output); Executor sourceExecutor = await RunnerContext.EnsureExecutorAsync(ExecutorId, tracer: null).ConfigureAwait(false); @@ -132,18 +210,42 @@ internal sealed class InProcessRunnerContext : IRunnerContext => RunnerContext.StateManager.ClearStateAsync(ExecutorId, scopeName); } - internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default) => Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellation).AsTask())); + internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default) + { + this.CheckEnded(); - internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellationToken = default) => Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellationToken).AsTask())); + return Task.WhenAll(this._executors.Values.Select(InvokeCheckpointingAsync)); + + async Task InvokeCheckpointingAsync(Task executorTask) + { + Executor executor = await executorTask.ConfigureAwait(false); + await executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellation).ConfigureAwait(false); + } + } + + internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellation = default) + { + this.CheckEnded(); + + return Task.WhenAll(this._executors.Values.Select(InvokeCheckpointRestoredAsync)); + + async Task InvokeCheckpointRestoredAsync(Task executorTask) + { + Executor executor = await executorTask.ConfigureAwait(false); + await executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellation).ConfigureAwait(false); + } + } internal ValueTask ExportStateAsync() { + this.CheckEnded(); + if (this.QueuedEvents.Count > 0) { throw new InvalidOperationException("Cannot export state when there are queued events. Please process or clear the events before exporting state."); } - Dictionary> queuedMessages = this._nextStep.ExportMessages(); + Dictionary> queuedMessages = this._nextStep.ExportMessages(); RunnerStateData result = new(instantiatedExecutors: [.. this._executors.Keys], queuedMessages, outstandingRequests: [.. this._externalRequests.Values]); @@ -153,6 +255,8 @@ internal sealed class InProcessRunnerContext : IRunnerContext internal async ValueTask RepublishUnservicedRequestsAsync(CancellationToken cancellation = default) { + this.CheckEnded(); + if (this.HasUnservicedRequests) { foreach (string requestId in this._externalRequests.Keys) @@ -165,6 +269,8 @@ internal sealed class InProcessRunnerContext : IRunnerContext internal async ValueTask ImportStateAsync(Checkpoint checkpoint) { + this.CheckEnded(); + if (this.QueuedEvents.Count > 0) { throw new InvalidOperationException("Cannot import state when there are queued events. Please process or clear the events before importing state."); @@ -192,4 +298,22 @@ internal sealed class InProcessRunnerContext : IRunnerContext await Task.WhenAll(executorTasks).ConfigureAwait(false); } + + [System.Diagnostics.CodeAnalysis.SuppressMessage("Maintainability", "CA1513:Use ObjectDisposedException throw helper", + Justification = "Does not exist in NetFx 4.7.2")] + internal void CheckEnded() + { + if (Volatile.Read(ref this._runEnded) == 1) + { + throw new InvalidOperationException($"Workflow run '{this._runId}' has been ended. Please start a new Run or StreamingRun."); + } + } + + public async ValueTask EndRunAsync() + { + if (Interlocked.Exchange(ref this._runEnded, 1) == 0) + { + await this._workflow.ReleaseOwnershipAsync(this).ConfigureAwait(false); + } + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/PortableValue.cs b/dotnet/src/Microsoft.Agents.Workflows/PortableValue.cs index 623a253fbd..6898440e91 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/PortableValue.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/PortableValue.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft. All rights reserved. using System; +using System.Diagnostics.CodeAnalysis; using System.Text.Json.Serialization; using Microsoft.Agents.Workflows.Checkpointing; @@ -92,82 +93,78 @@ public sealed class PortableValue /// If the underlying value implements delayed deserialization, this method will attempt to /// deserialize it to the specified type. If the value is already of the requested type, it is returned directly. /// Otherwise, the default value for TValue is returned. - /// - /// For nullable value types, make sure to make be nullable, e.g. int?, - /// otherwise the default non-null value of the type is returned when the value is missing. Use - /// to get the correct behavior when unable to pass in the explicit-nullable type. /// /// The type to which the value should be cast or deserialized. /// The value cast or deserialized to type TValue if possible; otherwise, the default value for type TValue. - public TValue? As() - { - if (this.Value is IDelayedDeserialization delayedDeserialization) - { - this._deserializedValueCache ??= delayedDeserialization.Deserialize(); - } - - if (this.Value is TValue typedValue) - { - return typedValue; - } - - return default; - } - - /// - /// Attempts to retrieve the underlying value as the specified nullable value type, deserializing if - /// necessary. - /// - /// If the underlying value implements delayed deserialization, this method will attempt to - /// deserialize it to the specified type. If the value is already of the requested type, it is returned directly. - /// Otherwise, null is returned. - /// The value type to which the value should be cast or deserialized. - /// The value cast or deserialized to type TValue if possible; otherwise, null. - public TValue? AsValue() where TValue : struct - { - if (this.Value is IDelayedDeserialization delayedDeserialization) - { - this._deserializedValueCache ??= delayedDeserialization.Deserialize(); - } - - if (this.Value is TValue typedValue) - { - return typedValue; - } - - return default; - } + public TValue? As() => this.Is(out TValue? value) ? value : default; /// /// Determines whether the current value can be represented as the specified type. /// /// The type to test for compatibility with the current value. /// true if the current value can be represented as type TValue; otherwise, false. - public bool Is() => this.IsType(typeof(TValue)); + public bool Is() => this.Is(out _); + + /// + /// Determines whether the current value can be represented as the specified type. + /// + /// The type to test for compatibility with the current value. + /// When this method returns, contains the value cast or deserialized to type TValue + /// if the conversion succeeded, or null if the conversion failed. + /// true if the current value can be represented as type TValue; otherwise, false. + public bool Is([NotNullWhen(true)] out TValue? value) + { + if (this.Value is IDelayedDeserialization delayedDeserialization) + { + this._deserializedValueCache ??= delayedDeserialization.Deserialize(); + } + + if (this.Value is TValue typedValue) + { + value = typedValue; + return true; + } + + value = default; + return false; + } /// /// Attempts to retrieve the underlying value as the specified type, deserializing if necessary. /// /// The type to which the value should be cast or deserialized. /// The value cast or deserialized to type targetType if possible; otherwise, null. - public object? AsType(Type targetType) - { - Throw.IfNull(targetType); - - if (this.Value is IDelayedDeserialization delayedDeserialization) - { - this._deserializedValueCache ??= delayedDeserialization.Deserialize(targetType); - } - - return this.Value is not null && targetType.IsAssignableFrom(this.Value.GetType()) - ? this.Value - : this._deserializedValueCache = null; - } + public object? AsType(Type targetType) => this.IsType(targetType, out object? value) ? value : null; /// /// Determines whether the current instance can be assigned to the specified target type. /// /// The type to compare with the current instance. Cannot be null. /// true if the current instance can be assigned to targetType; otherwise, false. - public bool IsType(Type targetType) => this.AsType(targetType) is not null; + public bool IsType(Type targetType) => this.IsType(targetType, out _); + + /// + /// Determines whether the current instance can be assigned to the specified target type. + /// + /// The type to compare with the current instance. Cannot be null. + /// When this method returns, contains the value cast or deserialized to type TValue + /// if the conversion succeeded, or null if the conversion failed. + /// true if the current instance can be assigned to targetType; otherwise, false. + public bool IsType(Type targetType, out object? value) + { + Throw.IfNull(targetType); + if (this.Value is IDelayedDeserialization delayedDeserialization) + { + this._deserializedValueCache ??= delayedDeserialization.Deserialize(targetType); + } + + if (this.Value is not null && targetType.IsAssignableFrom(this.Value.GetType())) + { + value = this.Value; + return true; + } + + value = null; + return false; + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Run.cs b/dotnet/src/Microsoft.Agents.Workflows/Run.cs index b468145952..33013fad0c 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Run.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Run.cs @@ -38,7 +38,7 @@ public enum RunStatus /// Represents a workflow run that tracks execution status and emitted workflow events, supporting resumption /// with responses to . /// -public class Run +public sealed class Run { internal static async ValueTask CaptureStreamAsync(StreamingRun run, CancellationToken cancellation = default) { @@ -147,4 +147,7 @@ public class Run return await this.RunToNextHaltAsync(cancellation).ConfigureAwait(false); } + + /// + public ValueTask EndRunAsync() => this._streamingRun.EndRunAsync(); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/StreamingRun.cs b/dotnet/src/Microsoft.Agents.Workflows/StreamingRun.cs index 9a64ec8096..7921c43b41 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/StreamingRun.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/StreamingRun.cs @@ -15,7 +15,7 @@ namespace Microsoft.Agents.Workflows; /// A run instance supporting a streaming form of receiving workflow events, and providing /// a mechanism to send responses back to the workflow. /// -public class StreamingRun +public sealed class StreamingRun { private TaskCompletionSource? _waitForResponseSource; private readonly ISuperStepRunner _stepRunner; @@ -157,6 +157,14 @@ public class StreamingRun eventSink.Add(e); } } + + /// + /// Signals the end of the current run and initiates any necessary cleanup operations asynchronously. + /// Enables the underlying Workflow instance to be reused in subsequent runs. + /// + /// A ValueTask that represents the asynchronous operation. The task is complete when the run has + /// ended and cleanup is finished. + public ValueTask EndRunAsync() => this._stepRunner.RequestEndRunAsync(); } /// diff --git a/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs index 4fab7abcd6..3de7079d09 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs @@ -2,7 +2,9 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.Workflows.Checkpointing; using Microsoft.Shared.Diagnostics; @@ -99,6 +101,71 @@ public class Workflow OutputExecutors = this.OutputExecutors }; } + + private bool _needsReset; + private bool IsResettable => this.Registrations.Values.All(registration => !registration.IsUnresettableSharedInstance); + + private async ValueTask TryResetExecutorRegistrationsAsync() + { + if (this.IsResettable) + { + foreach (ExecutorRegistration registration in this.Registrations.Values) + { + if (!await registration.TryResetAsync().ConfigureAwait(false)) + { + return false; + } + } + + this._needsReset = false; + return true; + } + + return false; + } + + private object? _ownerToken; + internal void TakeOwnership(object ownerToken) + { + object? maybeToken = Interlocked.CompareExchange(ref this._ownerToken, ownerToken, null); + if (maybeToken == null && this._needsReset) + { + // There is no owner, but the workflow failed to reset on ownership release (because there are + // shared executors). + throw new InvalidOperationException( + "Cannot reuse Workflow with shared Executor instances that do not implement IResettableExecutor." + ); + } + + if (maybeToken != null && !ReferenceEquals(maybeToken, ownerToken)) + { + // Someone else owns the workflow + Debug.Assert(maybeToken != null); + throw new InvalidOperationException("Cannot use a Workflow in multiple simultaneous (Streaming)Runs."); + } + + this._needsReset = true; + } + + [System.Diagnostics.CodeAnalysis.SuppressMessage("Maintainability", "CA1513:Use ObjectDisposedException throw helper", + Justification = "Does not exist in NetFx 4.7.2")] + internal async ValueTask ReleaseOwnershipAsync(object ownerToken) + { + if (this._ownerToken == null) + { + throw new InvalidOperationException("Attempting to release ownership of a Workflow that is not owned."); + } + + if (!ReferenceEquals(this._ownerToken, this._ownerToken)) + { + throw new InvalidOperationException("Attempt to release ownership of a Workflow by non-owner."); + } + + await this.TryResetExecutorRegistrationsAsync().ConfigureAwait(false); + + Interlocked.CompareExchange(ref this._ownerToken, null, ownerToken); + this._ownerToken = null; + } } /// diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/AgentWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/AgentWorkflowBuilderTests.cs index 9872decf0d..67d60f75a3 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/AgentWorkflowBuilderTests.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/AgentWorkflowBuilderTests.cs @@ -387,23 +387,30 @@ public class AgentWorkflowBuilderTests StringBuilder sb = new(); StreamingRun run = await InProcessExecution.StreamAsync(workflow, input); - await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); - - WorkflowOutputEvent? output = null; - await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) + try { - if (evt is AgentRunUpdateEvent executorComplete) - { - sb.Append(executorComplete.Data); - } - else if (evt is WorkflowOutputEvent e) - { - output = e; - break; - } - } + await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); - return (sb.ToString(), output?.As>()); + WorkflowOutputEvent? output = null; + await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) + { + if (evt is AgentRunUpdateEvent executorComplete) + { + sb.Append(executorComplete.Data); + } + else if (evt is WorkflowOutputEvent e) + { + output = e; + break; + } + } + + return (sb.ToString(), output?.As>()); + } + finally + { + await run.EndRunAsync(); + } } private sealed class DoubleEchoAgentWithBarrier(string name, StrongBox> barrier, StrongBox remaining) : DoubleEchoAgent(name) diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs index 1f6fbcb9d8..e6035af1be 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeMapSmokeTests.cs @@ -1,7 +1,9 @@ // Copyright (c) Microsoft. All rights reserved. using System.Collections.Generic; +using System.Linq; using System.Threading.Tasks; +using FluentAssertions; using Microsoft.Agents.Workflows.Execution; namespace Microsoft.Agents.Workflows.UnitTests; @@ -27,10 +29,20 @@ public class EdgeMapSmokeTests EdgeMap edgeMap = new(runContext, workflowEdges, [], "executor1", null); - await edgeMap.InvokeEdgeAsync(fanInEdge, "executor1", new("part1")); - MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + DeliveryMapping? mapping = await edgeMap.PrepareDeliveryForEdgeAsync(fanInEdge, new("part1", "executor1")); + mapping.Should().BeNull(); - await edgeMap.InvokeEdgeAsync(fanInEdge, "executor2", new("part2")); - MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, ("executor3", ["part1", "part2"])); + mapping = await edgeMap.PrepareDeliveryForEdgeAsync(fanInEdge, new("part2", "executor2")); + mapping.Should().NotBeNull(); + List deliveries = mapping.Deliveries.ToList(); + + deliveries.Should().HaveCount(2).And.AllSatisfy(delivery => delivery.TargetId.Should().Be("executor3")); + + HashSet expectedMessages = ["part1", "part2"]; + foreach (MessageDelivery delivery in deliveries) + { + string message = delivery.Envelope.As()!; + expectedMessages.Remove(message); + } } } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs index 053758472b..5286953adb 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/EdgeRunnerTests.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; +using FluentAssertions; using Microsoft.Agents.Workflows.Execution; namespace Microsoft.Agents.Workflows.UnitTests; @@ -34,20 +35,21 @@ public class EdgeRunnerTests DirectEdgeData edgeData = new("executor1", "executor2", new EdgeId(0), condition); DirectEdgeRunner runner = new(runContext, edgeData); - MessageEnvelope envelope = new(MessageVariant1, targetId: targetId); + MessageEnvelope envelope = new(MessageVariant1, "executor1", targetId: targetId); - await runner.ChaseAsync(envelope, tracer: null); + DeliveryMapping? mapping = await runner.ChaseEdgeAsync(envelope, stepTracer: null); bool expectMessage = (!conditionMatch.HasValue || conditionMatch.Value) && (!targetMatch.HasValue || targetMatch.Value); if (expectMessage) { - MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, ("executor2", [MessageVariant1])); + mapping.Should().NotBeNull(); + mapping.CheckDeliveries(["executor2"], [MessageVariant1]); } else { - MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + mapping.Should().BeNull(); } } @@ -93,26 +95,34 @@ public class EdgeRunnerTests FanOutEdgeData edgeData = new("executor1", ["executor2", "executor3"], new EdgeId(0), assigner); FanOutEdgeRunner runner = new(runContext, edgeData); - MessageEnvelope envelope = new("test", targetId: targetId); + MessageEnvelope envelope = new("test", "executor1", targetId: targetId); - await runner.ChaseAsync(envelope, tracer: null); + DeliveryMapping? mapping = await runner.ChaseEdgeAsync(envelope, stepTracer: 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 = []; + HashSet expectedReceivers = new(); if (expectForwardFrom2) { - expectedForwards.Add(("executor2", ["test"])); + expectedReceivers.Add("executor2"); } if (expectForwardFrom3) { - expectedForwards.Add(("executor3", ["test"])); + expectedReceivers.Add("executor3"); } - MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, expectedForwards.ToArray()); + if (!expectForwardFrom2 && !expectForwardFrom3) + { + mapping.Should().BeNull(); + } + else + { + mapping.Should().NotBeNull(); + mapping.CheckDeliveries(expectedReceivers, ["test"]); + } } [Fact] @@ -153,7 +163,6 @@ public class EdgeRunnerTests // Step 3: Send message from executor1, should not forward yet. // Step 4: Send message from executor2, should forward now. - FanInEdgeState state = runner.CreateState(); await RunIterationAsync(); // Repeat the same sequence, to ensure state is properly reset inside of FanInEdgeState. @@ -162,20 +171,26 @@ public class EdgeRunnerTests async ValueTask RunIterationAsync() { - await runner.ChaseAsync("executor1", new("part1"), state, tracer: null); + //await runner.ChaseAsync("executor1", new("part1"), state, tracer: null); + //MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + DeliveryMapping? mapping = await runner.ChaseEdgeAsync(new("part1", "executor1"), stepTracer: null); + mapping.Should().BeNull(); - MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + //await runner.ChaseAsync("executor2", new("part-for-1", targetId: "executor1"), state, tracer: null); + //MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages); + mapping = await runner.ChaseEdgeAsync(new("part-for-1", "executor2", targetId: "executor1"), stepTracer: null); + mapping.Should().BeNull(); - 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); + mapping = await runner.ChaseEdgeAsync(new("part2", "executor1", targetId: "executor3"), stepTracer: null); + mapping.Should().BeNull(); - 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"])); + //await runner.ChaseAsync("executor2", new("final part"), state, tracer: null); + //MessageDeliveryValidation.CheckForwarded(runContext.QueuedMessages, ("executor3", ["part1", "part2", "final part"])); + mapping = await runner.ChaseEdgeAsync(new("final part", "executor2"), stepTracer: null); + mapping.Should().NotBeNull(); + mapping.CheckDeliveries(["executor3"], ["part1", "part2", "final part"]); } } } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/InProcessStateTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/InProcessStateTests.cs index 3983d2b4a3..083be3d48d 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/InProcessStateTests.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/InProcessStateTests.cs @@ -103,6 +103,9 @@ public class InProcessStateTests Run run = await InProcessExecution.RunAsync(workflow, new()); run.Status.Should().Be(RunStatus.Idle); + + writer.Completed.Should().BeTrue(); + validator.Completed.Should().BeTrue(); } [Fact] @@ -129,8 +132,11 @@ public class InProcessStateTests Checkpointed checkpointed = await InProcessExecution.RunAsync(workflow, new(), CheckpointManager.Default); - checkpointed.Checkpoints.Should().HaveCount(6); + checkpointed.Checkpoints.Should().HaveCount(5); checkpointed.Run.Status.Should().Be(RunStatus.Idle); + + writer.Completed.Should().BeTrue(); + validator.Completed.Should().BeTrue(); } [Fact] diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/JsonSerializationTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/JsonSerializationTests.cs index a93c775cb0..f4b4884f76 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/JsonSerializationTests.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/JsonSerializationTests.cs @@ -378,7 +378,7 @@ public class JsonSerializationTests { const string Message = "TestMessage"; - MessageEnvelope envelope = new(Message, new TypeId(typeof(object)), targetId: "Target1"); + MessageEnvelope envelope = new(Message, "Source1", new TypeId(typeof(object)), targetId: "Target1"); PortableMessageEnvelope value = new(envelope); PortableMessageEnvelope result = RunJsonRoundtrip(value); @@ -402,7 +402,7 @@ public class JsonSerializationTests { ChatMessage message = new(ChatRole.User, "Hello, world!"); - MessageEnvelope envelope = new(message, new TypeId(typeof(object)), targetId: "Target1"); + MessageEnvelope envelope = new(message, "Source1", new TypeId(typeof(object)), targetId: "Target1"); PortableMessageEnvelope value = new(envelope); PortableMessageEnvelope result = RunJsonRoundtrip(value); @@ -433,7 +433,7 @@ public class JsonSerializationTests { TestJsonSerializable message = new() { Id = 42, Name = "Test" }; - MessageEnvelope envelope = new(message, new TypeId(typeof(object)), targetId: "Target1"); + MessageEnvelope envelope = new(message, "Source1", new TypeId(typeof(object)), targetId: "Target1"); PortableMessageEnvelope value = new(envelope); PortableMessageEnvelope result = RunJsonRoundtrip(value, TestCustomSerializedJsonOptions); @@ -462,15 +462,12 @@ public class JsonSerializationTests outstandingRequests: [TestExternalRequest] ); - static Dictionary> CreateQueuedMessages() + static Dictionary> CreateQueuedMessages() { - Dictionary> result = []; + Dictionary> result = []; - MessageEnvelope externalEnvelope = new(TestExternalResponse); - result.Add(ExecutorIdentity.None, [new(externalEnvelope)]); - - MessageEnvelope internalEnvelope = new("InternalMessage"); - result.Add("TestExecutor1", [new(internalEnvelope)]); + MessageEnvelope internalEnvelope = new("InternalMessage", "TestExecutor1"); + result.Add("TestExecutor2", [new(internalEnvelope)]); return result; } @@ -485,7 +482,7 @@ public class JsonSerializationTests (Action)(actual => actual.Should().Be(prototype))).ToArray()); result.QueuedMessages.Should().HaveCount(prototype.QueuedMessages.Count); - foreach (ExecutorIdentity key in prototype.QueuedMessages.Keys) + foreach (string key in prototype.QueuedMessages.Keys) { result.QueuedMessages.Should().ContainKey(key); @@ -524,7 +521,7 @@ public class JsonSerializationTests private static PortableValue CreateEdgeState(TMessage message) where TMessage : notnull { FanInEdgeState state = TestFanInEdgeState; - _ = state.ProcessMessage("SourceExecutor1", new MessageEnvelope(message, typeof(TMessage))); + _ = state.ProcessMessage("SourceExecutor1", new MessageEnvelope(message, "SourceExecutor1", typeof(TMessage))); return new(state); } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs index 7ad386e328..e009615702 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/MessageDeliveryValidation.cs @@ -10,6 +10,40 @@ namespace Microsoft.Agents.Workflows.UnitTests; internal static class MessageDeliveryValidation { + public static void CheckDeliveries(this DeliveryMapping mapping, HashSet receiverIds, HashSet messages) + { + HashSet unseenReceivers = [.. receiverIds]; + HashSet unseenMessages = [.. messages]; + + foreach (IGrouping grouping in mapping.Deliveries.GroupBy(delivery => delivery.TargetId)) + { + string receiverId = grouping.Key; + + receiverIds.Should().Contain(receiverId); + unseenReceivers.Remove(grouping.Key); + + foreach (MessageDelivery delivery in grouping) + { + object messageValue; + if (delivery.Envelope.Message is PortableValue portableValue) + { + portableValue.IsDelayedDeserialization.Should().BeFalse(); + messageValue = portableValue.Value; + } + else + { + messageValue = delivery.Envelope.Message; + } + + messages.Should().Contain(messageValue); + unseenMessages.Remove(messageValue); + } + } + + unseenReceivers.Should().BeEmpty(); + unseenMessages.Should().BeEmpty(); + } + public static void CheckForwarded(Dictionary> queuedMessages, params (string expectedSender, List expectedMessages)[] expectedForwards) { queuedMessages.Should().HaveCount(expectedForwards.Length); diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs index 0f4348bf32..775998cc22 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs @@ -92,7 +92,7 @@ internal sealed class GuessNumberExecutor : ReflectingExecutor, IMessageHandler +internal sealed class JudgeExecutor : ReflectingExecutor, IMessageHandler, IResettableExecutor { private readonly int _targetNumber; @@ -122,4 +122,10 @@ internal sealed class JudgeExecutor : ReflectingExecutor, IMessag { this.Tries = await context.ReadStateAsync("TryCount").ConfigureAwait(false) ?? 0; } + + public ValueTask ResetAsync() + { + this.Tries = null; + return default; + } } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs index c69618f4de..483745dce7 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs @@ -39,6 +39,8 @@ internal static class Step5EntryPoint if (rehydrateToRestore) { + await handle.EndRunAsync().ConfigureAwait(false); + checkpointed = await InProcessExecution.ResumeStreamAsync(workflow, targetCheckpoint, checkpointManager, runId: handle.RunId, cancellation: CancellationToken.None) .ConfigureAwait(false); handle = checkpointed.Run; @@ -59,7 +61,7 @@ internal static class Step5EntryPoint result = await RunStreamToHaltOrMaxStepAsync().ConfigureAwait(false); result.Should().NotBeNull(); - checkpoints.Should().HaveCount(7); + checkpoints.Should().HaveCount(6); cancellationSource.Dispose(); diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs index b3e8a07c82..d7fcd0d4a0 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestRunContext.cs @@ -61,11 +61,11 @@ public class TestRunContext : IRunnerContext this.QueuedMessages[sourceId] = deliveryQueue = []; } - deliveryQueue.Add(new(message, targetId: targetId)); + deliveryQueue.Add(new(message, sourceId, targetId: targetId)); return default; } - StepContext IRunnerContext.Advance() => + ValueTask IRunnerContext.AdvanceAsync() => throw new NotImplementedException(); public Dictionary Executors { get; } = []; diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestingExecutor.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestingExecutor.cs index 7eb534a796..50896d6968 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestingExecutor.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestingExecutor.cs @@ -15,6 +15,10 @@ internal abstract class TestingExecutor : Executor, IDisposable private readonly HashSet _linkedTokens = []; private CancellationTokenSource _internalCts = new(); + public int Iterations { get; private set; } + public bool AtEnd => this._nextActionIndex >= this._actions.Length; + public bool Completed => !this._loop && this.AtEnd; + protected TestingExecutor(string id, bool loop = false, params Func>[] actions) : base(id) { this._loop = loop; @@ -41,10 +45,11 @@ internal abstract class TestingExecutor : Executor, IDisposable private int _nextActionIndex; private ValueTask RouteToActionsAsync(TIn message, IWorkflowContext context) { - if (this._nextActionIndex >= this._actions.Length) + if (this.AtEnd) { if (this._loop) { + this.Iterations++; this._nextActionIndex = 0; } else