feat: Solidify Concurrency Model for Inproc Workflows (#925)

* Disallow execution of multiple workflows at once
* Define notion of executor Reset() to allow reuse of workflows with shared executor instances
* Switch to delivering all messages to a single executor sequentially, with executors running in parallel

dotnet/src/Microsoft.Agents.Workflows/AgentWorkflowBuilder.cs

@@ -145,7 +145,7 @@ public static partial class AgentWorkflowBuilder
     /// <summary>
     /// Executor that runs the agent and forwards all messages, input and output, to the next executor.
     /// </summary>
-    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<ChatMessage> _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;
+        }
     }
 
     /// <summary>
     /// Provides an executor that batches received chat messages that it then publishes as the final result
     /// when receiving a <see cref="TurnToken"/>.
     /// </summary>
-    private sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages")
+    private sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages"), IResettableExecutor
     {
         protected override ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default)
             => context.YieldOutputAsync(messages);
+
+        ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync();
     }
 
     /// <summary>Executor that forwards all messages.</summary>
-    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<ChatMessage>((message, context) => context.SendMessageAsync(message))
                     .AddHandler<List<ChatMessage>>((messages, context) => context.SendMessageAsync(messages))
                     .AddHandler<TurnToken>((turnToken, context) => context.SendMessageAsync(turnToken));
+
+        public ValueTask ResetAsync() => default;
     }
 
     /// <summary>
     /// Provides an executor that batches received chat messages that it then releases when
     /// receiving a <see cref="TurnToken"/>.
     /// </summary>
-    private sealed class BatchChatMessagesToListExecutor(string id) : ChatProtocolExecutor(id)
+    private sealed class BatchChatMessagesToListExecutor(string id) : ChatProtocolExecutor(id), IResettableExecutor
     {
         protected override ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default)
             => context.SendMessageAsync(messages);
+
+        ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync();
     }
 
     /// <summary>
     /// 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.
     /// </summary>
-    private sealed class ConcurrentEndExecutor : Executor
+    private sealed class ConcurrentEndExecutor : Executor, IResettableExecutor
     {
         private readonly int _expectedInputs;
         private readonly Func<IList<List<ChatMessage>>, List<ChatMessage>> _aggregator;
@@ -238,6 +250,12 @@ public static partial class AgentWorkflowBuilder
             this._remaining = expectedInputs;
         }
 
+        private void Reset()
+        {
+            this._allResults = new List<List<ChatMessage>>(this._expectedInputs);
+            this._remaining = this._expectedInputs;
+        }
+
         protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
             routeBuilder.AddHandler<List<ChatMessage>>(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;
+        }
     }
 
     /// <summary>
@@ -428,7 +452,7 @@ public static partial class AgentWorkflowBuilder
         }
 
         /// <summary>Executor used at the start of a handoffs workflow to accumulate messages and emit them as HandoffState upon receiving a turn token.</summary>
-        private sealed class StartHandoffsExecutor() : Executor("HandoffStart")
+        private sealed class StartHandoffsExecutor() : Executor("HandoffStart"), IResettableExecutor
         {
             private readonly List<ChatMessage> _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;
+            }
         }
 
         /// <summary>Executor used at the end of a handoff workflow to raise a final completed event.</summary>
-        private sealed class EndHandoffsExecutor() : Executor("HandoffEnd")
+        private sealed class EndHandoffsExecutor() : Executor("HandoffEnd"), IResettableExecutor
         {
             protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
                 routeBuilder.AddHandler<HandoffState>((handoff, context) =>
                     context.YieldOutputAsync(handoff.Messages));
+
+            public ValueTask ResetAsync() => default;
         }
 
         /// <summary>Executor used to represent an agent in a handoffs workflow, responding to <see cref="HandoffState"/> events.</summary>
         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<AIAgent, ExecutorIsh> agentMap, Func<IReadOnlyList<AIAgent>, GroupChatManager> managerFactory) : Executor("GroupChatHost")
+        private sealed class GroupChatHost(AIAgent[] agents, Dictionary<AIAgent, ExecutorIsh> agentMap, Func<IReadOnlyList<AIAgent>, GroupChatManager> managerFactory) : Executor("GroupChatHost"), IResettableExecutor
         {
             private readonly AIAgent[] _agents = agents;
             private readonly Dictionary<AIAgent, ExecutorIsh> _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;
+            }
         }
     }
 

dotnet/src/Microsoft.Agents.Workflows/ChatProtocolExecutor.cs

@@ -18,6 +18,14 @@ internal abstract class ChatProtocolExecutor(string id, ChatProtocolExecutorOpti
     private List<ChatMessage> _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)

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);
     }
 }

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 <see cref="IDelayedDeserialization"/> and <see cref="PortableValue"/> to abstract
-/// away the speicfics of a given serialization format in favor of <see cref="PortableValue.As{TValue}"/> and
-/// <see cref="PortableValue.Is{TValue}"/>.
+/// away the speicfics of a given serialization format in favor of <see cref="PortableValue.As{TValue}()"/> and
+/// <see cref="PortableValue.Is{TValue}()"/> and related methods.
 /// </summary>
 /// <param name="marshaller"></param>
 internal sealed class PortableValueConverter(JsonMarshaller marshaller) : JsonConverter<PortableValue>

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<MessageEnvelope> _envelopes;
+    private readonly IEnumerable<Executor> _targets;
+
+    public DeliveryMapping(IEnumerable<MessageEnvelope> envelopes, IEnumerable<Executor> 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<Executor> targets) : this([envelope], targets) { }
+    public DeliveryMapping(IEnumerable<MessageEnvelope> envelopes, Executor target) : this(envelopes, [target]) { }
+
+    public IEnumerable<MessageDelivery> 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));
+        }
+    }
+}

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<Executor> FindRouterAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer)
                                     .ConfigureAwait(false);
 
-    public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
+    protected internal override async ValueTask<DeliveryMapping?> 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;
     }
 }

dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs

@@ -10,12 +10,23 @@ namespace Microsoft.Agents.Workflows.Execution;
 
 internal sealed class EdgeMap
 {
-    private readonly Dictionary<EdgeId, object> _edgeRunners = [];
-    private readonly Dictionary<EdgeId, FanInEdgeState> _fanInState = [];
+    private readonly Dictionary<EdgeId, EdgeRunner> _edgeRunners = [];
+    private readonly Dictionary<EdgeId, IStatefulEdgeRunner> _statefulRunners = new();
     private readonly Dictionary<string, InputEdgeRunner> _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<string, HashSet<Edge>> workflowEdges,
                    IEnumerable<InputPort> 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<IEnumerable<object?>> InvokeEdgeAsync(Edge edge, string sourceId, MessageEnvelope message)
+    public ValueTask<DeliveryMapping?> 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<object?> 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<IEnumerable<object?>> InvokeInputAsync(MessageEnvelope envelope)
+    public ValueTask<DeliveryMapping?> PrepareDeliveryForInputAsync(MessageEnvelope message)
     {
-        return [await this._inputRunner.ChaseAsync(envelope, this._stepTracer).ConfigureAwait(false)];
+        return this._inputRunner.ChaseEdgeAsync(message, this._stepTracer);
     }
 
-    public async ValueTask<IEnumerable<object?>> InvokeResponseAsync(ExternalResponse response)
+    public ValueTask<DeliveryMapping?> 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<Dictionary<EdgeId, PortableValue>> ExportStateAsync()
+    internal async ValueTask<Dictionary<EdgeId, PortableValue>> ExportStateAsync()
     {
         Dictionary<EdgeId, PortableValue> 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<EdgeId, PortableValue> importedState = checkpoint.EdgeStateData;
 
-        this._fanInState.Clear();
         foreach (EdgeId id in importedState.Keys)
         {
             PortableValue exportedState = importedState[id];
-
-            FanInEdgeState? fanInState = exportedState.As<FanInEdgeState>();
-            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;
     }
 }

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<PortableValue> ExportStateAsync();
+    ValueTask ImportStateAsync(PortableValue state);
+}
+
+internal abstract class EdgeRunner
+{
+    // TODO: Can this be sync?
+    protected internal abstract ValueTask<DeliveryMapping?> ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer);
+}
+
 internal abstract class EdgeRunner<TEdgeData>(
-    IRunnerContext runContext, TEdgeData edgeData)
+    IRunnerContext runContext, TEdgeData edgeData) : EdgeRunner()
 {
     protected IRunnerContext RunContext { get; } = Throw.IfNull(runContext);
     protected TEdgeData EdgeData { get; } = Throw.IfNull(edgeData);

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<FanInEdgeData>(runContext, edgeData)
+    EdgeRunner<FanInEdgeData>(runContext, edgeData),
+    IStatefulEdgeRunner
 {
-    private IWorkflowContext BoundContext { get; } = runContext.Bind(edgeData.SinkId);
+    private FanInEdgeState _state = new(edgeData);
 
-    public FanInEdgeState CreateState() => new(this.EdgeData);
-
-    public ValueTask<IEnumerable<object?>> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer)
+    protected internal override async ValueTask<DeliveryMapping?> 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<MessageEnvelope>? releasedMessages = state.ProcessMessage(sourceId, envelope);
+        // source.Id is guaranteed to be non-null here because source is not None.
+        IEnumerable<MessageEnvelope>? 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<IEnumerable<object?>> ForwardReleasedMessagesAsync(IEnumerable<MessageEnvelope> 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<Task<object?>> messageTasks = [];
+        return new DeliveryMapping(releasedMessages.Where(envelope => target.CanHandle(envelope.MessageType)), target);
+    }
 
-        foreach (MessageEnvelope releasedEnvelope in releasedMessages)
+    public ValueTask<PortableValue> 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}");
     }
 }

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<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
+    protected internal override async ValueTask<DeliveryMapping?> ChaseEdgeAsync(MessageEnvelope envelope, IStepTracer? stepTracer)
     {
         object message = envelope.Message;
-        List<string> targets =
+        IEnumerable<string> 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<string> 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<object?> 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<Executor> 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;

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<StepContext> AdvanceAsync();
     IWorkflowContext Bind(string executorId);
     ValueTask<Executor> EnsureExecutorAsync(string executorId, IStepTracer? tracer);
 }

dotnet/src/Microsoft.Agents.Workflows/Execution/ISuperStepRunner.cs

@@ -19,4 +19,6 @@ internal interface ISuperStepRunner
     event EventHandler<WorkflowEvent>? WorkflowEvent;
 
     ValueTask<bool> RunSuperStepAsync(CancellationToken cancellation);
+
+    ValueTask RequestEndRunAsync();
 }

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<Executor> FindExecutorAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false);
-
-    public async ValueTask<object?> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
+    protected internal override async ValueTask<DeliveryMapping?> 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<Executor> FindExecutorAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false);
 }

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; }
+}

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()))
         {

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<string> instantiatedExecutors, Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> queuedMessages, List<ExternalRequest> outstandingRequests)
+internal sealed class RunnerStateData(HashSet<string> instantiatedExecutors, Dictionary<string, List<PortableMessageEnvelope>> queuedMessages, List<ExternalRequest> outstandingRequests)
 {
     public HashSet<string> InstantiatedExecutors { get; } = instantiatedExecutors;
-    public Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> QueuedMessages { get; } = queuedMessages;
+    public Dictionary<string, List<PortableMessageEnvelope>> QueuedMessages { get; } = queuedMessages;
     public List<ExternalRequest> OutstandingRequests { get; } = outstandingRequests;
 }

dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs

@@ -8,15 +8,15 @@ namespace Microsoft.Agents.Workflows.Execution;
 
 internal sealed class StepContext
 {
-    public Dictionary<ExecutorIdentity, List<MessageEnvelope>> QueuedMessages { get; } = [];
+    public Dictionary<string, List<MessageEnvelope>> QueuedMessages { get; } = [];
 
     public bool HasMessages => this.QueuedMessages.Values.Any(messageList => messageList.Count > 0);
 
-    public List<MessageEnvelope> MessagesFor(string? executorId)
+    public List<MessageEnvelope> 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<ExecutorIdentity, List<PortableMessageEnvelope>> ExportMessages()
+    internal Dictionary<string, List<PortableMessageEnvelope>> ExportMessages()
     {
         return this.QueuedMessages.Keys.ToDictionary(
             keySelector: identity => identity,
@@ -33,9 +33,9 @@ internal sealed class StepContext
         );
     }
 
-    internal void ImportMessages(Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> messages)
+    internal void ImportMessages(Dictionary<string, List<PortableMessageEnvelope>> messages)
     {
-        foreach (ExecutorIdentity identity in messages.Keys)
+        foreach (string identity in messages.Keys)
         {
             this.QueuedMessages[identity] = messages[identity].ConvertAll(UnwrapExportedState);
         }

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<bool> 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;
 

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+public interface IResettableExecutor
+{
+    /// <summary>
+    /// Reset the executor
+    /// </summary>
+    /// <returns>A <see cref="ValueTask"/> representing the completion of the reset operation.</returns>
+    ValueTask ResetAsync()
+#if NET
+    {
+        return default;
+    }
+#else
+    ;
+#endif
+}

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<bool> EnqueueMessageAsync<T>(T message)
+    private async ValueTask<bool> 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<bool> EnqueueMessageAsync(object message)
-    {
-        Throw.IfNull(message);
+    public ValueTask<bool> EnqueueMessageAsync<T>(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<bool> 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<IEnumerable<object?>> 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<IEnumerable<object?>> 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<StreamingRun> 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<StreamingRun> 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<StreamingRun> StreamAsync<TInput>(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<Run> 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<Run> 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<Run> RunAsync<TInput>(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<bool> 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<MessageEnvelope> 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<Task<IEnumerable<object?>>> edgeTasks = [];
-        foreach (ExecutorIdentity sender in currentStep.QueuedMessages.Keys)
-        {
-            IEnumerable<MessageEnvelope> 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<Edge>? outgoingEdges))
-            {
-                foreach (Edge outgoingEdge in outgoingEdges)
-                {
-                    edgeTasks.AddRange(senderMessages.Select(envelope => this.EdgeMap.InvokeEdgeAsync(outgoingEdge, sender.Id, envelope).AsTask()));
-                }
-            }
-        }
+        List<Task> 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<object?> 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<CheckpointInfo> _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();
 }

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<string, ExecutorRegistration> _executorRegistrations;
-    private readonly Dictionary<string, Executor> _executors = [];
-    private readonly Dictionary<string, ExternalRequest> _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<string, Task<Executor>> _executors = new();
+    private readonly ConcurrentQueue<Func<ValueTask>> _queuedExternalDeliveries = new();
+
+    private readonly Dictionary<string, ExternalRequest> _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<Executor> EnsureExecutorAsync(string executorId, IStepTracer? tracer)
     {
-        if (!this._executors.TryGetValue(executorId, out var executor))
+        this.CheckEnded();
+        Task<Executor> executorTask = this._executors.GetOrAdd(executorId, CreateExecutorAsync);
+
+        async Task<Executor> 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>(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<StepContext> 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<Edge>? 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<WorkflowEvent> 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<Executor> 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<Executor> executorTask)
+        {
+            Executor executor = await executorTask.ConfigureAwait(false);
+            await executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellation).ConfigureAwait(false);
+        }
+    }
 
     internal ValueTask<RunnerStateData> 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<ExecutorIdentity, List<PortableMessageEnvelope>> queuedMessages = this._nextStep.ExportMessages();
+        Dictionary<string, List<PortableMessageEnvelope>> 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);
+        }
+    }
 }

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
     /// <remarks>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 <typeparamref name="TValue"/> be nullable, e.g. <c>int?</c>,
-    /// otherwise the default non-null value of the type is returned when the value is missing. Use <see cref="AsValue{TValue}"/>
-    /// to get the correct behavior when unable to pass in the explicit-nullable type.
     /// </remarks>
     /// <typeparam name="TValue">The type to which the value should be cast or deserialized.</typeparam>
     /// <returns>The value cast or deserialized to type TValue if possible; otherwise, the default value for type TValue.</returns>
-    public TValue? As<TValue>()
-    {
-        if (this.Value is IDelayedDeserialization delayedDeserialization)
-        {
-            this._deserializedValueCache ??= delayedDeserialization.Deserialize<TValue>();
-        }
-
-        if (this.Value is TValue typedValue)
-        {
-            return typedValue;
-        }
-
-        return default;
-    }
-
-    /// <summary>
-    /// Attempts to retrieve the underlying value as the specified nullable value type, deserializing if
-    /// necessary.
-    /// </summary>
-    /// <remarks>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.</remarks>
-    /// <typeparam name="TValue">The value type to which the value should be cast or deserialized.</typeparam>
-    /// <returns>The value cast or deserialized to type TValue if possible; otherwise, null.</returns>
-    public TValue? AsValue<TValue>() where TValue : struct
-    {
-        if (this.Value is IDelayedDeserialization delayedDeserialization)
-        {
-            this._deserializedValueCache ??= delayedDeserialization.Deserialize<TValue>();
-        }
-
-        if (this.Value is TValue typedValue)
-        {
-            return typedValue;
-        }
-
-        return default;
-    }
+    public TValue? As<TValue>() => this.Is(out TValue? value) ? value : default;
 
     /// <summary>
     /// Determines whether the current value can be represented as the specified type.
     /// </summary>
     /// <typeparam name="TValue">The type to test for compatibility with the current value.</typeparam>
     /// <returns>true if the current value can be represented as type TValue; otherwise, false.</returns>
-    public bool Is<TValue>() => this.IsType(typeof(TValue));
+    public bool Is<TValue>() => this.Is<TValue>(out _);
+
+    /// <summary>
+    /// Determines whether the current value can be represented as the specified type.
+    /// </summary>
+    /// <typeparam name="TValue">The type to test for compatibility with the current value.</typeparam>
+    /// <param name="value">When this method returns, contains the value cast or deserialized to type TValue
+    /// if the conversion succeeded, or null if the conversion failed.</param>
+    /// <returns>true if the current value can be represented as type TValue; otherwise, false.</returns>
+    public bool Is<TValue>([NotNullWhen(true)] out TValue? value)
+    {
+        if (this.Value is IDelayedDeserialization delayedDeserialization)
+        {
+            this._deserializedValueCache ??= delayedDeserialization.Deserialize<TValue>();
+        }
+
+        if (this.Value is TValue typedValue)
+        {
+            value = typedValue;
+            return true;
+        }
+
+        value = default;
+        return false;
+    }
 
     /// <summary>
     /// Attempts to retrieve the underlying value as the specified type, deserializing if necessary.
     /// </summary>
     /// <param name="targetType">The type to which the value should be cast or deserialized.</param>
     /// <returns>The value cast or deserialized to type targetType if possible; otherwise, null.</returns>
-    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;
 
     /// <summary>
     /// Determines whether the current instance can be assigned to the specified target type.
     /// </summary>
     /// <param name="targetType">The type to compare with the current instance. Cannot be null.</param>
     /// <returns>true if the current instance can be assigned to targetType; otherwise, false.</returns>
-    public bool IsType(Type targetType) => this.AsType(targetType) is not null;
+    public bool IsType(Type targetType) => this.IsType(targetType, out _);
+
+    /// <summary>
+    /// Determines whether the current instance can be assigned to the specified target type.
+    /// </summary>
+    /// <param name="targetType">The type to compare with the current instance. Cannot be null.</param>
+    /// <param name="value">When this method returns, contains the value cast or deserialized to type TValue
+    /// if the conversion succeeded, or null if the conversion failed.</param>
+    /// <returns>true if the current instance can be assigned to targetType; otherwise, false.</returns>
+    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;
+    }
 }

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 <see cref="RequestInfoEvent"/>.
 /// </summary>
-public class Run
+public sealed class Run
 {
     internal static async ValueTask<Run> CaptureStreamAsync(StreamingRun run, CancellationToken cancellation = default)
     {
@@ -147,4 +147,7 @@ public class Run
 
         return await this.RunToNextHaltAsync(cancellation).ConfigureAwait(false);
     }
+
+    /// <inheritdoc cref="StreamingRun.EndRunAsync"/>
+    public ValueTask EndRunAsync() => this._streamingRun.EndRunAsync();
 }

dotnet/src/Microsoft.Agents.Workflows/StreamingRun.cs

@@ -15,7 +15,7 @@ namespace Microsoft.Agents.Workflows;
 /// A <see cref="Workflow"/> run instance supporting a streaming form of receiving workflow events, and providing
 /// a mechanism to send responses back to the workflow.
 /// </summary>
-public class StreamingRun
+public sealed class StreamingRun
 {
     private TaskCompletionSource<object>? _waitForResponseSource;
     private readonly ISuperStepRunner _stepRunner;
@@ -157,6 +157,14 @@ public class StreamingRun
             eventSink.Add(e);
         }
     }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A ValueTask that represents the asynchronous operation. The task is complete when the run has
+    /// ended and cleanup is finished.</returns>
+    public ValueTask EndRunAsync() => this._stepRunner.RequestEndRunAsync();
 }
 
 /// <summary>

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<bool> 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;
+    }
 }
 
 /// <summary>

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<List<ChatMessage>>());
+            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<List<ChatMessage>>());
+        }
+        finally
+        {
+            await run.EndRunAsync();
+        }
     }
 
     private sealed class DoubleEchoAgentWithBarrier(string name, StrongBox<TaskCompletionSource<bool>> barrier, StrongBox<int> remaining) : DoubleEchoAgent(name)

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<MessageDelivery> deliveries = mapping.Deliveries.ToList();
+
+        deliveries.Should().HaveCount(2).And.AllSatisfy(delivery => delivery.TargetId.Should().Be("executor3"));
+
+        HashSet<string> expectedMessages = ["part1", "part2"];
+        foreach (MessageDelivery delivery in deliveries)
+        {
+            string message = delivery.Envelope.As<string>()!;
+            expectedMessages.Remove(message);
+        }
     }
 }

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<string> expectedMessages)> expectedForwards = [];
+        HashSet<string> 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"]);
         }
     }
 }

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/InProcessStateTests.cs

@@ -103,6 +103,9 @@ public class InProcessStateTests
         Run run = await InProcessExecution.RunAsync<TurnToken>(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<Run> checkpointed = await InProcessExecution.RunAsync<TurnToken>(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]

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<ExecutorIdentity, List<PortableMessageEnvelope>> CreateQueuedMessages()
+            static Dictionary<string, List<PortableMessageEnvelope>> CreateQueuedMessages()
             {
-                Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> result = [];
+                Dictionary<string, List<PortableMessageEnvelope>> 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<string>)(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>(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);
     }

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<string> receiverIds, HashSet<object> messages)
+    {
+        HashSet<string> unseenReceivers = [.. receiverIds];
+        HashSet<object> unseenMessages = [.. messages];
+
+        foreach (IGrouping<string, MessageDelivery> 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<string, List<MessageEnvelope>> queuedMessages, params (string expectedSender, List<string> expectedMessages)[] expectedForwards)
     {
         queuedMessages.Should().HaveCount(expectedForwards.Length);

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs

@@ -92,7 +92,7 @@ internal sealed class GuessNumberExecutor : ReflectingExecutor<GuessNumberExecut
     }
 }
 
-internal sealed class JudgeExecutor : ReflectingExecutor<JudgeExecutor>, IMessageHandler<int, NumberSignal>
+internal sealed class JudgeExecutor : ReflectingExecutor<JudgeExecutor>, IMessageHandler<int, NumberSignal>, IResettableExecutor
 {
     private readonly int _targetNumber;
 
@@ -122,4 +122,10 @@ internal sealed class JudgeExecutor : ReflectingExecutor<JudgeExecutor>, IMessag
     {
         this.Tries = await context.ReadStateAsync<int?>("TryCount").ConfigureAwait(false) ?? 0;
     }
+
+    public ValueTask ResetAsync()
+    {
+        this.Tries = null;
+        return default;
+    }
 }

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();
 

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<StepContext> IRunnerContext.AdvanceAsync() =>
         throw new NotImplementedException();
 
     public Dictionary<string, Executor> Executors { get; } = [];

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/TestingExecutor.cs

@@ -15,6 +15,10 @@ internal abstract class TestingExecutor<TIn, TOut> : Executor, IDisposable
     private readonly HashSet<CancellationToken> _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<TIn, IWorkflowContext, CancellationToken, ValueTask<TOut>>[] actions) : base(id)
     {
         this._loop = loop;
@@ -41,10 +45,11 @@ internal abstract class TestingExecutor<TIn, TOut> : Executor, IDisposable
     private int _nextActionIndex;
     private ValueTask<TOut> RouteToActionsAsync(TIn message, IWorkflowContext context)
     {
-        if (this._nextActionIndex >= this._actions.Length)
+        if (this.AtEnd)
         {
             if (this._loop)
             {
+                this.Iterations++;
                 this._nextActionIndex = 0;
             }
             else