.NET: feat: Implement Checkpointing API (#420)

* feat: Implement Checkpointing API

* refactor: Normalzie Namespaces and break out multi-class files

* feat: Support checkpointing in AIAgentHostExecutor

* test: Representation tests

* feat: Add Step-level Tracing and WorkflowEvents

* feat: Add Checkpointing Sample and Smoke Test

* Fixes an issue where StateManager was not properly clearing the incoming queued updates.
* Fixes order of checkpointing and in-step event publication
* Adds import of RunContext state on LoadCheckpoint
* Add re-firing of events for unserviced ExternalRequests on Checkpoint load

* docs: Add documentation to publics

* Also adds documentation to ICheckpointManager which may go public

* refactor: Fix Union Aggregators and add Tests

* fix: Fix issues raised in PR comments and remove dead code

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

@@ -0,0 +1,48 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Represents a checkpoint with a unique identifier and a timestamp indicating when it was created.
+/// </summary>
+public class CheckpointInfo : IEquatable<CheckpointInfo>
+{
+    /// <summary>
+    /// The unique identifier for the checkpoint.
+    /// </summary>
+    public string CheckpointId { get; } = Guid.NewGuid().ToString("N");
+
+    /// <summary>
+    /// The date and time when the object was created, in Coordinated Universal Time (UTC).
+    /// </summary>
+    public DateTimeOffset CreatedAt { get; } = DateTimeOffset.UtcNow;
+
+    /// <inheritdoc/>
+    public bool Equals(CheckpointInfo? other)
+    {
+        if (other == null)
+        {
+            return false;
+        }
+
+        return this.CheckpointId == other.CheckpointId &&
+               this.CreatedAt == other.CreatedAt;
+    }
+
+    /// <inheritdoc/>
+    public override bool Equals(object? obj)
+    {
+        return this.Equals(obj as CheckpointInfo);
+    }
+
+    /// <inheritdoc/>
+    public override int GetHashCode()
+    {
+        return HashCode.Combine(this.CheckpointId, this.CreatedAt);
+    }
+
+    /// <inheritdoc/>
+    public override string ToString() => $"CheckpointId: {this.CheckpointId}, CreatedAt: {this.CreatedAt:O}";
+}

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

@@ -0,0 +1,36 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Agents.Workflows.Checkpointing;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// An in-memory implementation of <see cref="ICheckpointManager"/> that stores checkpoints in a dictionary.
+/// </summary>
+public sealed class CheckpointManager : ICheckpointManager
+{
+    private readonly Dictionary<CheckpointInfo, Checkpoint> _checkpoints = new();
+
+    ValueTask<CheckpointInfo> ICheckpointManager.CommitCheckpointAsync(Checkpoint checkpoint)
+    {
+        Throw.IfNull(checkpoint);
+
+        this._checkpoints[checkpoint] = checkpoint;
+        return new(checkpoint);
+    }
+
+    ValueTask<Checkpoint> ICheckpointManager.LookupCheckpointAsync(CheckpointInfo checkpointInfo)
+    {
+        Throw.IfNull(checkpointInfo);
+
+        if (!this._checkpoints.TryGetValue(checkpointInfo, out Checkpoint? checkpoint))
+        {
+            throw new KeyNotFoundException($"Checkpoint not found: {checkpointInfo}");
+        }
+
+        return new ValueTask<Checkpoint>(checkpoint);
+    }
+}

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

@@ -0,0 +1,49 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Agents.Workflows.Checkpointing;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Represents a workflow run that supports checkpointing.
+/// </summary>
+/// <typeparam name="TRun">The type of the underlying workflow run handle</typeparam>
+/// <seealso cref="Run"/>
+/// <seealso cref="Run{TResult}"/>
+/// <seealso cref="StreamingRun"/>
+/// <seealso cref="StreamingRun{TResult}"/>
+public class Checkpointed<TRun>
+{
+    internal Checkpointed(TRun run, ICheckpointingRunner runner)
+    {
+        this.Run = Throw.IfNull(run);
+        this._runner = Throw.IfNull(runner);
+    }
+
+    private readonly ICheckpointingRunner _runner;
+
+    /// <summary>
+    /// Gets the workflow run associated with this <see cref="Checkpointed{TRun}"/> instance.
+    /// </summary>
+    /// <seealso cref="Run"/>
+    /// <seealso cref="Run{TResult}"/>
+    /// <seealso cref="StreamingRun"/>
+    /// <seealso cref="StreamingRun{TResult}"/>
+    public TRun Run { get; }
+
+    /// <inheritdoc cref="ICheckpointingRunner.Checkpoints"/>
+    public IReadOnlyList<CheckpointInfo> Checkpoints => this._runner.Checkpoints;
+
+    /// <summary>
+    /// Gets the most recent checkpoint information.
+    /// </summary>
+    public CheckpointInfo? LastCheckpoint => this.Checkpoints.Count > 0 ? this.Checkpoints[this.Checkpoints.Count - 1] : null;
+
+    /// <inheritdoc cref="ICheckpointingRunner.RestoreCheckpointAsync"/>
+    public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default)
+        => this._runner.RestoreCheckpointAsync(checkpointInfo, cancellation);
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/Checkpoint.cs

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using Microsoft.Agents.Workflows.Execution;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class Checkpoint : CheckpointInfo
+{
+    internal Checkpoint(
+        int stepNumber,
+        WorkflowInfo workflow,
+        RunnerStateData runnerData,
+        Dictionary<ScopeKey, ExportedState> stateData,
+        Dictionary<EdgeConnection, ExportedState> edgeStateData)
+    {
+        this.StepNumber = Throw.IfLessThan(stepNumber, -1); // -1 is a special flag indicating the initial checkpoint.
+        this.Workflow = Throw.IfNull(workflow);
+        this.RunnerData = Throw.IfNull(runnerData);
+        this.State = Throw.IfNull(stateData);
+        this.EdgeState = Throw.IfNull(edgeStateData);
+    }
+
+    public bool IsInitial => this.StepNumber == -1;
+
+    public int StepNumber { get; }
+    public WorkflowInfo Workflow { get; }
+    public RunnerStateData RunnerData { get; }
+
+    public readonly Dictionary<ScopeKey, ExportedState> State = new();
+    public readonly Dictionary<EdgeConnection, ExportedState> EdgeState = new();
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/DirectEdgeInfo.cs

@@ -0,0 +1,14 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class DirectEdgeInfo(DirectEdgeData data) : EdgeInfo(Edge.Type.Direct, data.Connection)
+{
+    public bool HasCondition => data.Condition != null;
+
+    protected override bool IsMatchInternal(EdgeData edgeData)
+    {
+        return edgeData is DirectEdgeData directEdge
+            && this.HasCondition == (directEdge.Condition != null);
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/EdgeInfo.cs

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Microsoft.Agents.Workflows.Execution;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal abstract class EdgeInfo(Edge.Type edgeType, EdgeConnection connection)
+{
+    public Edge.Type EdgeType => edgeType;
+    public EdgeConnection Connection { get; } = Throw.IfNull(connection);
+
+    public bool IsMatch(Edge edge)
+    {
+        return this.EdgeType == edge.EdgeType
+            && this.Connection.Equals(edge.Data.Connection)
+            && this.IsMatchInternal(edge.Data);
+    }
+
+    protected virtual bool IsMatchInternal(EdgeData edgeData) => true;
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ExecutorInfo.cs

@@ -0,0 +1,24 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal record class ExecutorInfo(TypeId ExecutorType, string ExecutorId)
+{
+    public bool IsMatch<T>() where T : Executor
+    {
+        return this.ExecutorType.IsMatch<T>()
+            && this.ExecutorId == typeof(T).Name;
+    }
+
+    public bool IsMatch(Executor executor)
+    {
+        return this.ExecutorType.IsMatch(executor.GetType())
+            && this.ExecutorId == executor.Id;
+    }
+
+    public bool IsMatch(ExecutorRegistration registration)
+    {
+        return this.ExecutorType.IsMatch(registration.ExecutorType)
+            && this.ExecutorId == registration.Id;
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ExportedState.cs

@@ -0,0 +1,12 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class ExportedState(object state)
+{
+    public Type RuntimeType => Throw.IfNull(state).GetType();
+    public object Value => Throw.IfNull(state);
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/FanInEdgeInfo.cs

@@ -0,0 +1,5 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class FanInEdgeInfo(FanInEdgeData data) : EdgeInfo(Edge.Type.FanIn, data.Connection);

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/FanOutEdgeInfo.cs

@@ -0,0 +1,14 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class FanOutEdgeInfo(FanOutEdgeData data) : EdgeInfo(Edge.Type.FanOut, data.Connection)
+{
+    public bool HasAssigner => data.EdgeAssigner != null;
+
+    protected override bool IsMatchInternal(EdgeData edgeData)
+    {
+        return edgeData is FanOutEdgeData fanOutEdge
+            && this.HasAssigner == (fanOutEdge.EdgeAssigner != null);
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ICheckpointManager.cs

@@ -0,0 +1,28 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+/// <summary>
+/// A manager for storing and retrieving workflow execution checkpoints.
+/// </summary>
+internal interface ICheckpointManager
+{
+    /// <summary>
+    /// Commits the specified checkpoint and returns information that can be used to retrieve it later.
+    /// </summary>
+    /// <param name="checkpoint">The <see cref="Checkpoint"/> to be committed.</param>
+    /// <returns>A <see cref="CheckpointInfo"/> representing the incoming checkpoint.</returns>
+    ValueTask<CheckpointInfo> CommitCheckpointAsync(Checkpoint checkpoint);
+
+    /// <summary>
+    /// Retrieves the checkpoint associated with the specified checkpoint information.
+    /// </summary>
+    /// <param name="checkpointInfo">The information used to identify the checkpoint.</param>
+    /// <returns>A <see cref="ValueTask{TResult}"/> representing the asynchronous operation. The result contains the <see
+    /// cref="Checkpoint"/> associated with the specified <paramref name="checkpointInfo"/>.</returns>
+    /// <exception cref="KeyNotFoundException">Thrown if the checkpoint is not found.</exception>
+    ValueTask<Checkpoint> LookupCheckpointAsync(CheckpointInfo checkpointInfo);
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ICheckpointingRunner.cs

@@ -0,0 +1,15 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal interface ICheckpointingRunner
+{
+    // TODO: Convert this to a multi-timeline (e.g.: Live timeline + forks for orphaned checkpoints due to timetravel)
+    IReadOnlyList<CheckpointInfo> Checkpoints { get; }
+
+    ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default);
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/InputPortInfo.cs

@@ -0,0 +1,5 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal record class InputPortInfo(TypeId InputType, TypeId OutputType, string PortId);

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/RepresentationExtensions.cs

@@ -0,0 +1,59 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal static class RepresentationExtensions
+{
+    public static ExecutorInfo ToExecutorInfo(this ExecutorRegistration registration)
+    {
+        Throw.IfNull(registration);
+        return new ExecutorInfo(new TypeId(registration.ExecutorType), registration.Id);
+    }
+
+    public static EdgeInfo ToEdgeInfo(this Edge edge)
+    {
+        Throw.IfNull(edge);
+        return edge.EdgeType switch
+        {
+            Edge.Type.Direct => new DirectEdgeInfo(edge.DirectEdgeData!),
+            Edge.Type.FanOut => new FanOutEdgeInfo(edge.FanOutEdgeData!),
+            Edge.Type.FanIn => new FanInEdgeInfo(edge.FanInEdgeData!),
+            _ => throw new NotSupportedException($"Unsupported edge type: {edge.EdgeType}")
+        };
+    }
+
+    public static InputPortInfo ToPortInfo(this InputPort port)
+    {
+        Throw.IfNull(port);
+        return new(new TypeId(port.Request), new TypeId(port.Response), port.Id);
+    }
+
+    private static WorkflowInfo ToWorkflowInfo<TInput>(this Workflow<TInput> workflow, TypeId? outputType, string? outputExecutorId)
+    {
+        Throw.IfNull(workflow);
+
+        Dictionary<string, ExecutorInfo> executors =
+            workflow.Registrations.Values.ToDictionary(
+                keySelector: registration => registration.Id,
+                elementSelector: ToExecutorInfo);
+
+        Dictionary<string, List<EdgeInfo>> edges = workflow.Edges.Keys.ToDictionary(
+            keySelector: sourceId => sourceId,
+            elementSelector: sourceId => workflow.Edges[sourceId].Select(ToEdgeInfo).ToList());
+
+        HashSet<InputPortInfo> inputPorts = new(workflow.Ports.Values.Select(ToPortInfo));
+
+        return new WorkflowInfo(executors, edges, inputPorts, new TypeId(workflow.InputType), workflow.StartExecutorId, outputType, outputExecutorId);
+    }
+
+    public static WorkflowInfo ToWorkflowInfo<TInput>(this Workflow<TInput> workflow)
+        => workflow.ToWorkflowInfo(outputType: null, outputExecutorId: null);
+
+    public static WorkflowInfo GetInfo<TInput, TResult>(this Workflow<TInput, TResult> workflow)
+        => workflow.ToWorkflowInfo(outputType: new TypeId(typeof(TResult)), outputExecutorId: workflow.OutputCollectorId);
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/TypeId.cs

@@ -0,0 +1,20 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class TypeId(Type type)
+{
+    public string AssemblyName => Throw.IfNull(type.Assembly.FullName);
+    public string TypeName => Throw.IfNull(type.FullName);
+
+    public bool IsMatch(Type type)
+    {
+        return this.AssemblyName == type.Assembly.FullName
+            && this.TypeName == type.FullName;
+    }
+
+    public bool IsMatch<T>() => this.IsMatch(typeof(T));
+}

dotnet/src/Microsoft.Agents.Workflows/Checkpointing/WorkflowInfo.cs

@@ -0,0 +1,111 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Checkpointing;
+
+internal class WorkflowInfo
+{
+    internal WorkflowInfo(
+        Dictionary<string, ExecutorInfo> executors,
+        Dictionary<string, List<EdgeInfo>> edges,
+        HashSet<InputPortInfo> inputPorts,
+        TypeId inputType,
+        string startExecutorId,
+        TypeId? outputType = null,
+        string? outputCollectorId = null)
+    {
+        this.Executors = Throw.IfNull(executors);
+        this.Edges = Throw.IfNull(edges);
+        this.InputPorts = Throw.IfNull(inputPorts);
+
+        this.InputType = Throw.IfNull(inputType);
+        this.StartExecutorId = Throw.IfNullOrEmpty(startExecutorId);
+
+        if (outputType != null && outputCollectorId != null)
+        {
+            this.OutputType = outputType;
+            this.OutputCollectorId = outputCollectorId;
+        }
+        else if (outputCollectorId != null)
+        {
+            throw new InvalidOperationException(
+                $"Either both or none of OutputType and OutputCollectorId must be set. ({nameof(outputType)}: {outputType} vs. {nameof(outputCollectorId)}: {outputCollectorId})"
+            );
+        }
+    }
+
+    public Dictionary<string, ExecutorInfo> Executors { get; }
+    public Dictionary<string, List<EdgeInfo>> Edges { get; }
+    public HashSet<InputPortInfo> InputPorts { get; }
+
+    public TypeId InputType { get; }
+    public string StartExecutorId { get; }
+
+    public TypeId? OutputType { get; }
+    public string? OutputCollectorId { get; }
+
+    private bool IsMatch(Workflow workflow)
+    {
+        if (workflow is null)
+        {
+            return false;
+        }
+
+        if (!this.InputType.IsMatch(workflow.InputType))
+        {
+            return false;
+        }
+
+        if (this.StartExecutorId != workflow.StartExecutorId)
+        {
+            return false;
+        }
+
+        // Validate the executors
+        if (workflow.Registrations.Count != this.Executors.Count ||
+            this.Executors.Keys.Any(
+            executorId => workflow.Registrations.TryGetValue(executorId, out ExecutorRegistration? registration)
+                       && !this.Executors[executorId].IsMatch(registration)))
+        {
+            return false;
+        }
+
+        // Validate the edges
+        if (workflow.Edges.Count != this.Edges.Count ||
+            this.Edges.Keys.Any(
+                sourceId =>
+                    // If the sourceId is not present in the workflow edges, or
+                    !workflow.Edges.TryGetValue(sourceId, out var edgeList) ||
+                    // If the edge list count does not match, or
+                    edgeList.Count != this.Edges[sourceId].Count ||
+                    // If any edge in the workflow edge list does not match the corresponding edge in this.Edges[sourceId]
+                    !edgeList.All(edge => this.Edges[sourceId].Any(e => e.IsMatch(edge)))
+            ))
+        {
+            return false;
+        }
+
+        // Validate the input ports
+        if (workflow.Ports.Count != this.InputPorts.Count ||
+            this.InputPorts.Any(portInfo =>
+                !workflow.Ports.TryGetValue(portInfo.PortId, out InputPort? port) ||
+                !portInfo.InputType.IsMatch(port.Request) ||
+                !portInfo.OutputType.IsMatch(port.Response)))
+        {
+            return false;
+        }
+
+        return true;
+    }
+
+    public bool IsMatch<TInput>(Workflow<TInput> workflow) => this.IsMatch(workflow as Workflow);
+
+    public bool IsMatch<TInput, TResult>(Workflow<TInput, TResult> workflow)
+        => this.IsMatch(workflow as Workflow)
+           && this.OutputType != null && this.OutputType.IsMatch(typeof(TResult))
+           && this.OutputCollectorId != null && this.OutputCollectorId == workflow.OutputCollectorId;
+}

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

@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
+using Microsoft.Agents.Workflows.Execution;
 using PredicateT = System.Func<object?, bool>;
 
 namespace Microsoft.Agents.Workflows;
@@ -11,7 +12,7 @@ namespace Microsoft.Agents.Workflows;
 /// <param name="sourceId">The id of the source executor node.</param>
 /// <param name="sinkId">The id of the target executor node.</param>
 /// <param name="condition">A predicate determining whether the edge is active for a given message.</param>
-public sealed class DirectEdgeData(string sourceId, string sinkId, PredicateT? condition = null)
+public sealed class DirectEdgeData(string sourceId, string sinkId, PredicateT? condition = null) : EdgeData
 {
     /// <summary>
     /// The Id of the source <see cref="Executor"/> node.
@@ -28,4 +29,7 @@ public sealed class DirectEdgeData(string sourceId, string sinkId, PredicateT? c
     /// the edge is always active when a message is generated by the source.
     /// </summary>
     public PredicateT? Condition => condition;
+
+    /// <inheritdoc />
+    internal override EdgeConnection Connection { get; } = new([sourceId], [sinkId]);
 }

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

@@ -45,7 +45,7 @@ public sealed class Edge
     /// <seealso cref="DirectEdgeData"/>
     /// <seealso cref="FanOutEdgeData"/>
     /// <seealso cref="FanInEdgeData"/>
-    public object Data { get; init; }
+    public EdgeData Data { get; init; }
 
     internal Edge(DirectEdgeData data)
     {

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

@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Microsoft.Agents.Workflows.Execution;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// A base class for edge data, providing access to the <see cref="EdgeConnection"/> representation of the edge.
+/// </summary>
+public abstract class EdgeData
+{
+    /// <summary>
+    /// Gets the connection representation of the edge.
+    /// </summary>
+    internal abstract EdgeConnection Connection { get; }
+}

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

@@ -10,13 +10,13 @@ internal class DirectEdgeRunner(IRunnerContext runContext, DirectEdgeData edgeDa
 {
     public IWorkflowContext WorkflowContext { get; } = runContext.Bind(edgeData.SinkId);
 
-    private async ValueTask<Executor> FindRouterAsync()
+    private async ValueTask<Executor> FindRouterAsync(IStepTracer? tracer)
     {
-        return await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId)
+        return await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer)
                                     .ConfigureAwait(false);
     }
 
-    public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope)
+    public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
     {
         if (envelope.TargetId != null && this.EdgeData.SinkId != envelope.TargetId)
         {
@@ -29,9 +29,10 @@ internal class DirectEdgeRunner(IRunnerContext runContext, DirectEdgeData edgeDa
             return [];
         }
 
-        Executor target = await this.FindRouterAsync().ConfigureAwait(false);
+        Executor target = await this.FindRouterAsync(tracer).ConfigureAwait(false);
         if (target.CanHandle(envelope.MessageType))
         {
+            tracer?.TraceActivated(target.Id);
             return [await target.ExecuteAsync(message, envelope.MessageType, this.WorkflowContext).ConfigureAwait(false)];
         }
 

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

@@ -0,0 +1,94 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows.Execution;
+
+/// <summary>
+/// A representation for the connection structure of an edge of any multiplicity, defined by an ordered list
+/// of sources and sinks connected by this edge. Can also function as a unique identifier for the edge.
+/// </summary>
+/// <remarks>
+/// Ordering is relevant because in at least one case, the order of sinks is significant for the execution of
+/// the edge: <see cref="FanOutEdgeData"/>.
+/// </remarks>
+/// <param name="sourceIds">An ordered list of unique identifiers of the sources connected by this edge.</param>
+/// <param name="sinkIds">An ordered list of unique identifiers of the sinks connected by this edge.</param>
+public class EdgeConnection(List<string> sourceIds, List<string> sinkIds) : IEquatable<EdgeConnection>
+{
+    /// <summary>
+    /// Creates a new <see cref="EdgeConnection"/> instance with the specified source and sink IDs, ensuring that all
+    /// IDs are unique.
+    /// </summary>
+    /// <param name="sourceIds">A list of source IDs. Each ID must be unique within the list.</param>
+    /// <param name="sinkIds">A list of sink IDs. Each ID must be unique within the list.</param>
+    /// <returns>An <see cref="EdgeConnection"/> instance containing the specified source and sink IDs.</returns>
+    /// <exception cref="ArgumentNullException">Throw if <paramref name="sourceIds"/> or <paramref name="sinkIds"/>
+    /// is <see langword="null"/></exception>
+    /// <exception cref="ArgumentException">Thrown if <paramref name="sourceIds"/> or <paramref name="sinkIds"/>
+    /// contains duplicate values.</exception>
+    public static EdgeConnection CreateChecked(List<string> sourceIds, List<string> sinkIds)
+    {
+        HashSet<string> sourceSet = new(Throw.IfNull(sourceIds));
+        HashSet<string> sinkSet = new(Throw.IfNull(sinkIds));
+
+        if (sourceSet.Count != sourceIds.Count)
+        {
+            throw new ArgumentException("Source IDs must be unique.", nameof(sourceIds));
+        }
+
+        if (sinkSet.Count != sinkIds.Count)
+        {
+            throw new ArgumentException("Sink IDs must be unique.", nameof(sinkIds));
+        }
+
+        return new EdgeConnection(sourceIds, sinkIds);
+    }
+
+    /// <inheritdoc />
+    public bool Equals(EdgeConnection? other)
+    {
+        if (other is null)
+        {
+            return false;
+        }
+
+        if (object.ReferenceEquals(this, other))
+        {
+            return true;
+        }
+
+        return this.SourceIds.SequenceEqual(other.SourceIds) &&
+               this.SinkIds.SequenceEqual(other.SinkIds);
+    }
+
+    /// <inheritdoc />
+    public override bool Equals(object? obj)
+    {
+        return this.Equals(obj as EdgeConnection);
+    }
+
+    /// <inheritdoc />
+    public override int GetHashCode()
+    {
+        return HashCode.Combine(
+            this.SourceIds.Count,
+            this.SinkIds.Count,
+            this.SourceIds.Aggregate(0, (hash, id) => HashCode.Combine(hash, id.GetHashCode())),
+            this.SinkIds.Aggregate(0, (hash, id) => HashCode.Combine(hash, id.GetHashCode()))
+        );
+    }
+
+    /// <summary>
+    /// The unique identifiers of the sources connected by this edge.
+    /// </summary>
+    public List<string> SourceIds { get; } = sourceIds;
+
+    /// <summary>
+    /// The unique identifiers of the sinks connected by this edge.
+    /// </summary>
+    public List<string> SinkIds { get; } = sinkIds;
+}

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

@@ -4,20 +4,23 @@ using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using Microsoft.Agents.Workflows.Checkpointing;
 
 namespace Microsoft.Agents.Workflows.Execution;
 
 internal class EdgeMap
 {
-    private readonly Dictionary<Edge, object> _edgeRunners = new();
-    private readonly Dictionary<Edge, FanInEdgeState> _fanInState = new();
+    private readonly Dictionary<EdgeConnection, object> _edgeRunners = new();
+    private readonly Dictionary<EdgeConnection, FanInEdgeState> _fanInState = new();
     private readonly Dictionary<string, InputEdgeRunner> _portEdgeRunners;
     private readonly InputEdgeRunner _inputRunner;
+    private readonly IStepTracer? _stepTracer;
 
     public EdgeMap(IRunnerContext runContext,
                    Dictionary<string, HashSet<Edge>> workflowEdges,
                    IEnumerable<InputPort> workflowPorts,
-                   string startExecutorId)
+                   string startExecutorId,
+                   IStepTracer? stepTracer = null)
     {
         foreach (Edge edge in workflowEdges.Values.SelectMany(e => e))
         {
@@ -29,7 +32,7 @@ internal class EdgeMap
                 _ => throw new NotSupportedException($"Unsupported edge type: {edge.EdgeType}")
             };
 
-            this._edgeRunners[edge] = edgeRunner;
+            this._edgeRunners[edge.Data.Connection] = edgeRunner;
         }
 
         this._portEdgeRunners = workflowPorts.ToDictionary(
@@ -38,11 +41,13 @@ internal class EdgeMap
             );
 
         this._inputRunner = new InputEdgeRunner(runContext, startExecutorId);
+        this._stepTracer = stepTracer;
     }
 
     public async ValueTask<IEnumerable<object?>> InvokeEdgeAsync(Edge edge, string sourceId, MessageEnvelope message)
     {
-        if (!this._edgeRunners.TryGetValue(edge, out object? edgeRunner))
+        EdgeConnection connection = edge.Data.Connection;
+        if (!this._edgeRunners.TryGetValue(connection, out object? edgeRunner))
         {
             throw new InvalidOperationException($"Edge {edge} not found in the edge map.");
         }
@@ -58,23 +63,23 @@ internal class EdgeMap
             // between the Runners, we can normalize it behind an IFace.
             case Edge.Type.Direct:
             {
-                DirectEdgeRunner runner = (DirectEdgeRunner)this._edgeRunners[edge];
-                edgeResults = await runner.ChaseAsync(message).ConfigureAwait(false);
+                DirectEdgeRunner runner = (DirectEdgeRunner)this._edgeRunners[connection];
+                edgeResults = await runner.ChaseAsync(message, this._stepTracer).ConfigureAwait(false);
                 break;
             }
 
             case Edge.Type.FanOut:
             {
-                FanOutEdgeRunner runner = (FanOutEdgeRunner)this._edgeRunners[edge];
-                edgeResults = await runner.ChaseAsync(message).ConfigureAwait(false);
+                FanOutEdgeRunner runner = (FanOutEdgeRunner)this._edgeRunners[connection];
+                edgeResults = await runner.ChaseAsync(message, this._stepTracer).ConfigureAwait(false);
                 break;
             }
 
             case Edge.Type.FanIn:
             {
-                FanInEdgeState state = this._fanInState[edge];
-                FanInEdgeRunner runner = (FanInEdgeRunner)this._edgeRunners[edge];
-                edgeResults = [await runner.ChaseAsync(sourceId, message, state).ConfigureAwait(false)];
+                FanInEdgeState state = this._fanInState[connection];
+                FanInEdgeRunner runner = (FanInEdgeRunner)this._edgeRunners[connection];
+                edgeResults = [await runner.ChaseAsync(sourceId, message, state, this._stepTracer).ConfigureAwait(false)];
                 break;
             }
 
@@ -89,7 +94,7 @@ internal class EdgeMap
     // TODO: Should we promote Input to a true "FlowEdge" type?
     public async ValueTask<IEnumerable<object?>> InvokeInputAsync(MessageEnvelope envelope)
     {
-        return [await this._inputRunner.ChaseAsync(envelope).ConfigureAwait(false)];
+        return [await this._inputRunner.ChaseAsync(envelope, this._stepTracer).ConfigureAwait(false)];
     }
 
     public async ValueTask<IEnumerable<object?>> InvokeResponseAsync(ExternalResponse response)
@@ -99,6 +104,41 @@ internal class EdgeMap
             throw new InvalidOperationException($"Port {response.Port.Id} not found in the edge map.");
         }
 
-        return [await portRunner.ChaseAsync(new MessageEnvelope(response)).ConfigureAwait(false)];
+        return [await portRunner.ChaseAsync(new MessageEnvelope(response), this._stepTracer).ConfigureAwait(false)];
+    }
+
+    internal ValueTask<Dictionary<EdgeConnection, ExportedState>> ExportStateAsync()
+    {
+        Dictionary<EdgeConnection, ExportedState> exportedStates = new();
+
+        // Right now there is only fan-in state
+        foreach (EdgeConnection connection in this._fanInState.Keys)
+        {
+            FanInEdgeState state = this._fanInState[connection];
+            exportedStates[connection] = new ExportedState(state);
+        }
+
+        return new ValueTask<Dictionary<EdgeConnection, ExportedState>>(exportedStates);
+    }
+
+    internal ValueTask ImportStateAsync(Checkpoint checkpoint)
+    {
+        Dictionary<EdgeConnection, ExportedState> importedState = checkpoint.EdgeState;
+
+        this._fanInState.Clear();
+        foreach (EdgeConnection connection in importedState.Keys)
+        {
+            ExportedState exportedState = importedState[connection];
+            if (exportedState.Value is FanInEdgeState fanInState)
+            {
+                this._fanInState[connection] = fanInState;
+            }
+            else
+            {
+                throw new InvalidOperationException($"Unsupported exported state type: {exportedState.GetType()} for connection {connection}");
+            }
+        }
+
+        return default;
     }
 }

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

@@ -12,7 +12,7 @@ internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData
 
     public FanInEdgeState CreateState() => new(this.EdgeData);
 
-    public async ValueTask<object?> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state)
+    public async ValueTask<object?> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer)
     {
         if (envelope.TargetId != null && this.EdgeData.SinkId != envelope.TargetId)
         {
@@ -28,11 +28,12 @@ internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData
             return null;
         }
 
-        Executor target = await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId)
+        Executor target = await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer)
                                                    .ConfigureAwait(false);
 
         if (target.CanHandle(message.GetType()))
         {
+            tracer?.TraceActivated(target.Id);
             return await target.ExecuteAsync(message, envelope.MessageType, this.BoundContext)
                                .ConfigureAwait(false);
         }

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

@@ -14,13 +14,13 @@ internal class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData edgeDa
             sinkId => sinkId,
             sinkId => runContext.Bind(sinkId));
 
-    public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope)
+    public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
     {
         object message = envelope.Message;
         List<string> targets =
-            this.EdgeData.PartitionAssigner == null
+            this.EdgeData.EdgeAssigner == null
                 ? this.EdgeData.SinkIds
-                : this.EdgeData.PartitionAssigner(message, this.BoundContexts.Count)
+                : this.EdgeData.EdgeAssigner(message, this.BoundContexts.Count)
                                .Select(i => this.EdgeData.SinkIds[i]).ToList();
 
         IEnumerable<string> filteredTargets = envelope.TargetId != null
@@ -32,11 +32,12 @@ internal class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData edgeDa
 
         async Task<object?> ProcessTargetAsync(string targetId)
         {
-            Executor executor = await this.RunContext.EnsureExecutorAsync(targetId)
+            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);
             }

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

@@ -9,9 +9,7 @@ internal interface IRunnerContext : IExternalRequestSink
     ValueTask AddEventAsync(WorkflowEvent workflowEvent);
     ValueTask SendMessageAsync(string sourceId, object message, string? targetId = null);
 
-    // TODO: State Management
-
     StepContext Advance();
     IWorkflowContext Bind(string executorId);
-    ValueTask<Executor> EnsureExecutorAsync(string executorId);
+    ValueTask<Executor> EnsureExecutorAsync(string executorId, IStepTracer? tracer);
 }

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

@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows.Execution;
+
+internal interface IStepTracer
+{
+    void TraceActivated(string executorId);
+    void TraceCheckpointCreated(CheckpointInfo checkpoint);
+    void TraceIntantiated(string executorId);
+    void TraceStatePublished();
+}

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

@@ -19,16 +19,17 @@ internal class InputEdgeRunner(IRunnerContext runContext, string sinkId)
         return new InputEdgeRunner(runContext, port.Id);
     }
 
-    private async ValueTask<Executor> FindExecutorAsync()
+    private async ValueTask<Executor> FindExecutorAsync(IStepTracer? tracer)
     {
-        return await this.RunContext.EnsureExecutorAsync(this.EdgeData).ConfigureAwait(false);
+        return await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false);
     }
 
-    public async ValueTask<object?> ChaseAsync(MessageEnvelope envelope)
+    public async ValueTask<object?> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
     {
-        Executor target = await this.FindExecutorAsync().ConfigureAwait(false);
+        Executor target = await this.FindExecutorAsync(tracer).ConfigureAwait(false);
         if (target.CanHandle(envelope.MessageType))
         {
+            tracer?.TraceActivated(target.Id);
             return await target.ExecuteAsync(envelope.Message, envelope.MessageType, this.WorkflowContext)
                                .ConfigureAwait(false);
         }

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

@@ -0,0 +1,12 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using Microsoft.Agents.Workflows.Checkpointing;
+
+namespace Microsoft.Agents.Workflows.Execution;
+
+internal class RunnerStateData(Dictionary<ExecutorIdentity, List<ExportedState>> queuedMessages, List<ExternalRequest> outstandingRequests)
+{
+    public Dictionary<ExecutorIdentity, List<ExportedState>> QueuedMessages { get; } = queuedMessages;
+    public List<ExternalRequest> OutstandingRequests { get; } = outstandingRequests;
+}

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

@@ -2,7 +2,9 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
+using Microsoft.Agents.Workflows.Checkpointing;
 using Microsoft.Shared.Diagnostics;
 
 namespace Microsoft.Agents.Workflows.Execution;
@@ -69,7 +71,7 @@ internal class StateManager
         return default;
     }
 
-    public async ValueTask PublishUpdatesAsync()
+    public async ValueTask PublishUpdatesAsync(IStepTracer? tracer)
     {
         Dictionary<ScopeId, Dictionary<string, List<StateUpdate>>> updatesByScope = new();
 
@@ -89,10 +91,57 @@ internal class StateManager
             stateUpdates.Add(this._queuedUpdates[key]);
         }
 
+        if (updatesByScope.Count > 0 && tracer != null)
+        {
+            tracer.TraceStatePublished();
+        }
+
         foreach (ScopeId scope in updatesByScope.Keys)
         {
             StateScope stateScope = this.GetOrCreateScope(scope);
             await stateScope.WriteStateAsync(updatesByScope[scope]).ConfigureAwait(false);
         }
+
+        this._queuedUpdates.Clear();
+    }
+
+    private static IEnumerable<KeyValuePair<ScopeKey, ExportedState>> ExportScope(StateScope scope)
+    {
+        foreach (KeyValuePair<string, ExportedState> state in scope.ExportStates())
+        {
+            yield return new(new ScopeKey(scope.ScopeId, state.Key), state.Value);
+        }
+    }
+
+    internal async ValueTask<Dictionary<ScopeKey, ExportedState>> ExportStateAsync()
+    {
+        if (this._queuedUpdates.Count != 0)
+        {
+            throw new InvalidOperationException("Cannot export state while there are queued updates. Call PublishUpdatesAsync() first.");
+        }
+
+        return this._scopes.Values.SelectMany(ExportScope).ToDictionary(kvp => kvp.Key, kvp => kvp.Value);
+    }
+
+    internal ValueTask ImportStateAsync(Checkpoint checkpoint)
+    {
+        // TODO: Should this be a warning instead?
+        if (this._queuedUpdates.Count != 0)
+        {
+            throw new InvalidOperationException("Cannot import state while there are queued updates. Call PublishUpdatesAsync() first.");
+        }
+
+        this._queuedUpdates.Clear();
+        this._scopes.Clear();
+
+        Dictionary<ScopeKey, ExportedState> importedState = checkpoint.State;
+
+        foreach (ScopeKey scopeKey in importedState.Keys)
+        {
+            StateScope scope = this.GetOrCreateScope(scopeKey.ScopeId);
+            scope.ImportState(scopeKey.Key, importedState[scopeKey]);
+        }
+
+        return default;
     }
 }

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

@@ -2,7 +2,9 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
+using Microsoft.Agents.Workflows.Checkpointing;
 using Microsoft.Shared.Diagnostics;
 
 namespace Microsoft.Agents.Workflows.Execution;
@@ -61,4 +63,22 @@ internal class StateScope
 
         return default;
     }
+
+    public IEnumerable<KeyValuePair<string, ExportedState>> ExportStates()
+    {
+        return this._stateData.Keys.Select(WrapStates);
+
+        KeyValuePair<string, ExportedState> WrapStates(string key)
+        {
+            return new(key, new(this._stateData[key]));
+        }
+    }
+
+    public void ImportState(string key, ExportedState state)
+    {
+        Throw.IfNullOrEmpty(key);
+        Throw.IfNull(state);
+
+        this._stateData[key] = state.Value;
+    }
 }

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

@@ -1,7 +1,9 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
+using Microsoft.Agents.Workflows.Checkpointing;
 
 namespace Microsoft.Agents.Workflows.Execution;
 
@@ -20,4 +22,28 @@ internal class StepContext
 
         return messages;
     }
+
+    // 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<ExportedState>> ExportMessages()
+    {
+        return this.QueuedMessages.Keys.ToDictionary(
+            keySelector: identity => identity,
+            elementSelector: identity => this.QueuedMessages[identity]
+                                             .Select(v => new ExportedState(v))
+                                             .ToList()
+        );
+    }
+
+    internal void ImportMessages(Dictionary<ExecutorIdentity, List<ExportedState>> messages)
+    {
+        foreach (ExecutorIdentity identity in messages.Keys)
+        {
+            this.QueuedMessages[identity] = messages[identity].Select(UnwrapExportedState).ToList();
+        }
+
+        MessageEnvelope UnwrapExportedState(ExportedState es)
+            => es.Value as MessageEnvelope
+            ?? throw new InvalidDataException($"Expected a MessageEnvelope in the ExportedState. Got {es.RuntimeType}");
+    }
 }

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

@@ -5,6 +5,16 @@ using Microsoft.Shared.Diagnostics;
 
 namespace Microsoft.Agents.Workflows.Execution;
 
+/// <summary>
+/// Represents a unique key used to identify an update within a specific scope.
+/// </summary>
+/// <remarks>An <see cref="UpdateKey"/> is composed of a <see cref="ScopeId"/> and a key, similar
+/// to <see cref="ScopeKey"/>. The difference is in how equality is determined: Unlike ScopeKey,
+/// two UpdateKeys that differ only by their ScopeId's ExecutorId are considered different, because
+/// updates coming from different executors need to be tracked separately, until they are marged (if
+/// appropriate) and published during a step transition.</remarks>
+/// <param name="scopeId"></param>
+/// <param name="key"></param>
 internal class UpdateKey(ScopeId scopeId, string key)
 {
     public ScopeId ScopeId { get; } = Throw.IfNull(scopeId);

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

@@ -4,6 +4,7 @@ using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Reflection;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Agents.Workflows.Execution;
 
@@ -110,6 +111,22 @@ public abstract class Executor : IIdentified
         return result.Result;
     }
 
+    /// <summary>
+    /// Invoked before a checkpoint is saved, allowing custom pre-save logic in derived classes.
+    /// </summary>
+    /// <param name="context">The workflow context.</param>
+    /// <returns>A ValueTask representing the asynchronous operation.</returns>
+    /// <param name="cancellation"></param>
+    protected internal virtual ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => default;
+
+    /// <summary>
+    /// Invoked after a checkpoint is loaded, allowing custom post-load logic in derived classes.
+    /// </summary>
+    /// <param name="context">The workflow context.</param>
+    /// <returns>A ValueTask representing the asynchronous operation.</returns>
+    /// <param name="cancellation"></param>
+    protected internal virtual ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => default;
+
     /// <summary>
     /// A set of <see cref="Type"/>s, representing the messages this executor can handle.
     /// </summary>

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

@@ -103,10 +103,27 @@ public sealed class ExecutorIsh :
     };
 
     /// <summary>
-    /// Gets an <see cref="ExecutorProvider{T}"/> that can be used to obtain an <see cref="Executor"/> instance
+    /// Gets the registration details for the current executor.
+    /// </summary>
+    /// <remarks>The returned registration depends on the type of the executor. If the executor is unbound, an
+    /// <see cref="InvalidOperationException"/> is thrown. For other executor types, the registration  includes the
+    /// appropriate ID, type, and provider based on the executor's configuration.</remarks>
+    internal ExecutorRegistration Registration => new(this.Id, this.RuntimeType, this.ExecutorProvider);
+
+    private System.Type RuntimeType => this.ExecutorType switch
+    {
+        Type.Unbound => throw new InvalidOperationException($"ExecutorIsh with ID '{this.Id}' is unbound."),
+        Type.Executor => this._executorValue!.GetType(),
+        Type.InputPort => typeof(RequestInfoExecutor),
+        Type.Agent => typeof(AIAgentHostExecutor),
+        _ => throw new InvalidOperationException($"Unknown ExecutorIsh type: {this.ExecutorType}")
+    };
+
+    /// <summary>
+    /// Gets an <see cref="Func{Executor}"/> that can be used to obtain an <see cref="Executor"/> instance
     /// corresponding to this <see cref="ExecutorIsh"/>.
     /// </summary>
-    public ExecutorProvider<Executor> ExecutorProvider => this.ExecutorType switch
+    private Func<Executor> ExecutorProvider => this.ExecutorType switch
     {
         Type.Unbound => throw new InvalidOperationException($"Executor with ID '{this.Id}' is unbound."),
         Type.Executor => () => this._executorValue!,

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

@@ -0,0 +1,30 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using Microsoft.Shared.Diagnostics;
+
+using ExecutorFactoryF = System.Func<Microsoft.Agents.Workflows.Executor>;
+
+namespace Microsoft.Agents.Workflows;
+
+internal class ExecutorRegistration(string id, Type executorType, ExecutorFactoryF provider)
+{
+    public string Id { get; } = Throw.IfNullOrEmpty(id);
+    public Type ExecutorType { get; } = Throw.IfNull(executorType);
+    public ExecutorFactoryF Provider { get; } = Throw.IfNull(provider);
+
+    public override string ToString() => $"{this.ExecutorType.Name}({this.Id})";
+
+    private Executor CheckId(Executor executor)
+    {
+        if (executor.Id != this.Id)
+        {
+            throw new InvalidOperationException(
+                $"Executor ID mismatch: expected '{this.Id}', but got '{executor.Id}'.");
+        }
+
+        return executor;
+    }
+
+    public Executor CreateInstance() => this.CheckId(this.Provider());
+}

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

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft. All rights reserved.
 
-using System;
 using System.Collections.Generic;
+using Microsoft.Agents.Workflows.Execution;
 
 namespace Microsoft.Agents.Workflows;
 
@@ -10,7 +10,7 @@ namespace Microsoft.Agents.Workflows;
 /// </summary>
 /// <param name="sourceIds">An enumeration of ids of the source executor nodes.</param>
 /// <param name="sinkId">The id of the target executor node.</param>
-public sealed class FanInEdgeData(List<string> sourceIds, string sinkId)
+public sealed class FanInEdgeData(List<string> sourceIds, string sinkId) : EdgeData
 {
     /// <summary>
     /// The ordered list of Ids of the source <see cref="Executor"/> nodes.
@@ -22,5 +22,6 @@ public sealed class FanInEdgeData(List<string> sourceIds, string sinkId)
     /// </summary>
     public string SinkId => sinkId;
 
-    internal Guid UniqueKey { get; } = Guid.NewGuid();
+    /// <inheritdoc />
+    internal override EdgeConnection Connection { get; } = new(sourceIds, [sinkId]);
 }

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

@@ -1,7 +1,9 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
-using PartitionerT = System.Func<object?, int, System.Collections.Generic.IEnumerable<int>>;
+using Microsoft.Agents.Workflows.Execution;
+
+using AssignerF = System.Func<object?, int, System.Collections.Generic.IEnumerable<int>>;
 
 namespace Microsoft.Agents.Workflows;
 
@@ -11,11 +13,11 @@ namespace Microsoft.Agents.Workflows;
 /// </summary>
 /// <param name="sourceId">The id of the source executor node.</param>
 /// <param name="sinkIds">A list of ids of the target executor nodes.</param>
-/// <param name="partitioner">A function that maps an incoming message to a subset of the target executor nodes.</param>
+/// <param name="assigner">A function that maps an incoming message to a subset of the target executor nodes.</param>
 public sealed class FanOutEdgeData(
     string sourceId,
     List<string> sinkIds,
-    PartitionerT? partitioner = null)
+    AssignerF? assigner = null) : EdgeData
 {
     /// <summary>
     /// The Id of the source <see cref="Executor"/> node.
@@ -31,5 +33,8 @@ public sealed class FanOutEdgeData(
     /// A function mapping an incoming message to a subset of the target executor nodes (or optionally all of them).
     /// If <see langword="null"/>, all destination nodes are selected.
     /// </summary>
-    public PartitionerT? PartitionAssigner => partitioner;
+    public AssignerF? EdgeAssigner => assigner;
+
+    /// <inheritdoc />
+    internal override EdgeConnection Connection { get; } = new([sourceId], sinkIds);
 }

dotnet/src/Microsoft.Agents.Workflows/InProc/InProcStepTracer.cs

@@ -0,0 +1,91 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using Microsoft.Agents.Workflows.Execution;
+
+namespace Microsoft.Agents.Workflows.InProc;
+
+internal sealed class InProcStepTracer : IStepTracer
+{
+    private int _nextStepNumber = 0;
+
+    public int StepNumber => this._nextStepNumber - 1;
+    public bool StateUpdated { get; private set; } = false;
+    public CheckpointInfo? Checkpoint { get; private set; } = null;
+
+    public HashSet<string> Instantiated { get; } = [];
+    public HashSet<string> Activated { get; } = [];
+
+    public void TraceIntantiated(string executorId) => this.Instantiated.Add(executorId);
+    public void TraceActivated(string executorId) => this.Activated.Add(executorId);
+    public void TraceStatePublished() => this.StateUpdated = true;
+    public void TraceCheckpointCreated(CheckpointInfo checkpoint) => this.Checkpoint = checkpoint;
+
+    /// <summary>
+    /// Reset the tracer to the specified step number.
+    /// </summary>
+    /// <param name="lastStepNumber">The Step Number of the last SuperStep. Note that Step Numbers are 0-indexed.</param>
+    public void Reload(int lastStepNumber = 0) => this._nextStepNumber = lastStepNumber + 1;
+
+    public SuperStepStartedEvent Advance(StepContext step)
+    {
+        this._nextStepNumber++;
+        this.Activated.Clear();
+        this.Instantiated.Clear();
+
+        this.StateUpdated = false;
+        this.Checkpoint = null;
+
+        HashSet<string> sendingExecutors = [];
+        bool hasExternalMessages = false;
+
+        foreach (ExecutorIdentity identity in step.QueuedMessages.Keys)
+        {
+            if (identity == ExecutorIdentity.None)
+            {
+                hasExternalMessages = true;
+            }
+            else
+            {
+                sendingExecutors.Add(identity.Id!);
+            }
+        }
+
+        return new SuperStepStartedEvent(this.StepNumber, new SuperStepStartInfo(sendingExecutors)
+        {
+            HasExternalMessages = hasExternalMessages
+        });
+    }
+
+    public SuperStepCompletedEvent Complete(bool nextStepHasActions, bool hasPendingRequests)
+    {
+        return new SuperStepCompletedEvent(this.StepNumber, new SuperStepCompletionInfo(this.Activated, this.Instantiated)
+        {
+            HasPendingMessages = nextStepHasActions,
+            HasPendingRequests = hasPendingRequests,
+            StateUpdated = this.StateUpdated,
+            Checkpoint = this.Checkpoint,
+        });
+    }
+
+    public override string ToString()
+    {
+        StringBuilder sb = new();
+        if (this.Instantiated.Count != 0)
+        {
+            sb.Append("Instantiated: ");
+            sb.Append(string.Join(", ", this.Instantiated.OrderBy(id => id, StringComparer.Ordinal)));
+            sb.AppendLine();
+        }
+        if (this.Activated.Count != 0)
+        {
+            sb.Append("Activated: ");
+            sb.Append(string.Join(", ", this.Activated.OrderBy(id => id, StringComparer.Ordinal)));
+            sb.AppendLine();
+        }
+        return sb.ToString();
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs

@@ -3,9 +3,11 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.IO;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Agents.Workflows.Checkpointing;
 using Microsoft.Agents.Workflows.Execution;
 using Microsoft.Shared.Diagnostics;
 
@@ -18,16 +20,17 @@ namespace Microsoft.Agents.Workflows.InProc;
 /// within the current process, without distributed coordination. It is primarily intended for testing, debugging, or
 /// scenarios where workflow execution does not require executor distribution. </para></remarks>
 /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
-internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
+internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner where TInput : notnull
 {
-    public InProcessRunner(Workflow<TInput> workflow)
+    public InProcessRunner(Workflow<TInput> workflow, ICheckpointManager? checkpointManager)
     {
         this.Workflow = Throw.IfNull(workflow);
         this.RunContext = new InProcessRunnerContext<TInput>(workflow);
+        this.CheckpointManager = checkpointManager;
 
         // Initialize the runners for each of the edges, along with the state for edges that
         // need it.
-        this.EdgeMap = new EdgeMap(this.RunContext, this.Workflow.Edges, this.Workflow.Ports.Values, this.Workflow.StartExecutorId);
+        this.EdgeMap = new EdgeMap(this.RunContext, this.Workflow.Edges, this.Workflow.Ports.Values, this.Workflow.StartExecutorId, this.StepTracer);
     }
 
     public async ValueTask<bool> IsValidInputAsync<TMessage>(TMessage message)
@@ -42,7 +45,7 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
             return true;
         }
 
-        Executor startingExecutor = await this.RunContext.EnsureExecutorAsync(this.Workflow.StartExecutorId).ConfigureAwait(false);
+        Executor startingExecutor = await this.RunContext.EnsureExecutorAsync(this.Workflow.StartExecutorId, tracer: null).ConfigureAwait(false);
         return startingExecutor.CanHandle(type);
     }
 
@@ -64,9 +67,11 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
         return this.RunContext.AddExternalMessageAsync(response);
     }
 
+    private InProcStepTracer StepTracer { get; } = new();
     private Dictionary<string, string> PendingCalls { get; } = new();
     private Workflow<TInput> Workflow { get; init; }
     private InProcessRunnerContext<TInput> RunContext { get; init; }
+    private ICheckpointManager? CheckpointManager { get; }
     private EdgeMap EdgeMap { get; init; }
 
     event EventHandler<WorkflowEvent>? ISuperStepRunner.WorkflowEvent
@@ -107,6 +112,19 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
         return this.EdgeMap.InvokeResponseAsync(response);
     }
 
+    public async ValueTask<StreamingRun> ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
+    {
+        Throw.IfNull(checkpoint);
+        if (this.CheckpointManager is null)
+        {
+            throw new InvalidOperationException("This runner was not configured with a CheckpointManager, so it cannot restore checkpoints.");
+        }
+
+        await this.RestoreCheckpointAsync(checkpoint, cancellation).ConfigureAwait(false);
+
+        return new StreamingRun(this);
+    }
+
     public async ValueTask<StreamingRun> StreamAsync(TInput input, CancellationToken cancellation = default)
     {
         await this.RunContext.AddExternalMessageAsync(input).ConfigureAwait(false);
@@ -114,6 +132,14 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
         return new StreamingRun(this);
     }
 
+    internal async ValueTask<Run> ResumeAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
+    {
+        StreamingRun streamingRun = await this.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false);
+        cancellation.ThrowIfCancellationRequested();
+
+        return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
+    }
+
     public async ValueTask<Run> RunAsync(TInput input, CancellationToken cancellation = default)
     {
         StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false);
@@ -125,6 +151,9 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
     bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests;
     bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions;
 
+    public IReadOnlyList<CheckpointInfo> Checkpoints => this._checkpoints;
+    private CheckpointInfo? LastCheckpoint => this.Checkpoints[this.Checkpoints.Count - 1];
+
     async ValueTask<bool> ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellation)
     {
         cancellation.ThrowIfCancellationRequested();
@@ -142,6 +171,8 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
 
     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 = new();
         foreach (ExecutorIdentity sender in currentStep.QueuedMessages.Keys)
@@ -165,37 +196,121 @@ internal class InProcessRunner<TInput> : ISuperStepRunner where TInput : notnull
         // 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);
 
-        // Commit the state updates (so they are visible to the next step)
-        await this.RunContext.StateManager.PublishUpdatesAsync().ConfigureAwait(false);
-
         // After the message handler invocations, we may have some events to deliver
         foreach (WorkflowEvent @event in this.RunContext.QueuedEvents)
         {
             this.RaiseWorkflowEvent(@event);
         }
-
         this.RunContext.QueuedEvents.Clear();
+
+        await this.CheckpointAsync().ConfigureAwait(false);
+
+        this.RaiseWorkflowEvent(this.StepTracer.Complete(this.RunContext.NextStepHasActions, this.RunContext.HasUnservicedRequests));
+    }
+
+    private WorkflowInfo? _workflowInfoCache = null;
+    private readonly List<CheckpointInfo> _checkpoints = [];
+    internal async ValueTask CheckpointAsync(CancellationToken cancellation = default)
+    {
+        if (this.CheckpointManager == null)
+        {
+            return;
+        }
+
+        // Notify all the executors that they should prepare for checkpointing.
+        Task prepareTask = this.RunContext.PrepareForCheckpointAsync(cancellation);
+
+        // Create a representation of the current workflow if it does not already exist.
+        if (this._workflowInfoCache == null)
+        {
+            this._workflowInfoCache = this.Workflow.ToWorkflowInfo();
+        }
+
+        RunnerStateData runnerData = await this.RunContext.ExportStateAsync().ConfigureAwait(false);
+        Dictionary<EdgeConnection, ExportedState> edgeData = await this.EdgeMap.ExportStateAsync().ConfigureAwait(false);
+
+        await prepareTask.ConfigureAwait(false);
+        await this.RunContext.StateManager.PublishUpdatesAsync(this.StepTracer).ConfigureAwait(false);
+
+        Dictionary<ScopeKey, ExportedState> stateData = await this.RunContext.StateManager.ExportStateAsync().ConfigureAwait(false);
+
+        Checkpoint checkpoint = new(this.StepTracer.StepNumber, this._workflowInfoCache, runnerData, stateData, edgeData);
+        CheckpointInfo checkpointInfo = await this.CheckpointManager.CommitCheckpointAsync(checkpoint).ConfigureAwait(false);
+        this.StepTracer.TraceCheckpointCreated(checkpointInfo);
+        this._checkpoints.Add(checkpointInfo);
+    }
+
+    public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default)
+    {
+        Throw.IfNull(checkpointInfo);
+        if (this.CheckpointManager is null)
+        {
+            throw new InvalidOperationException("This run was not configured with a CheckpointManager, so it cannot restore checkpoints.");
+        }
+
+        Checkpoint checkpoint = await this.CheckpointManager.LookupCheckpointAsync(checkpointInfo)
+                                                            .ConfigureAwait(false);
+
+        // Validate the checkpoint is compatible with this workflow
+        if (!this.CheckWorkflowMatch(checkpoint))
+        {
+            // TODO: ArgumentException?
+            throw new InvalidDataException("The specified checkpoint is not compatible with the workflow associated with this runner.");
+        }
+
+        await this.RunContext.StateManager.ImportStateAsync(checkpoint).ConfigureAwait(false);
+        Task executorNotifyTask = this.RunContext.NotifyCheckpointLoadedAsync(cancellation);
+
+        await this.RunContext.ImportStateAsync(checkpoint).ConfigureAwait(false);
+        ValueTask republishRequestsTask = this.RunContext.RepublishUnservicedRequestsAsync(cancellation);
+
+        await this.EdgeMap.ImportStateAsync(checkpoint).ConfigureAwait(false);
+        await Task.WhenAll(executorNotifyTask, republishRequestsTask.AsTask()).ConfigureAwait(false);
+
+        this.StepTracer.Reload(this.StepTracer.StepNumber);
+    }
+
+    protected virtual bool CheckWorkflowMatch(Checkpoint checkpoint)
+    {
+        return checkpoint.Workflow.IsMatch<TInput>(this.Workflow);
     }
 }
 
-internal class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult> where TInput : notnull
+internal class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult>, ICheckpointingRunner where TInput : notnull
 {
     private readonly Workflow<TInput, TResult> _workflow;
-    private readonly ISuperStepRunner _innerRunner;
+    private readonly InProcessRunner<TInput> _innerRunner;
 
-    public InProcessRunner(Workflow<TInput, TResult> workflow)
+    public InProcessRunner(Workflow<TInput, TResult> workflow, CheckpointManager? checkpointManager)
     {
         this._workflow = Throw.IfNull(workflow);
-        this._innerRunner = new InProcessRunner<TInput>(workflow);
+
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager);
+        this._innerRunner = runner;
+    }
+
+    internal async ValueTask<StreamingRun<TResult>> ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
+    {
+        await this._innerRunner.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false);
+
+        return new StreamingRun<TResult>(this);
     }
 
     public async ValueTask<StreamingRun<TResult>> StreamAsync(TInput input, CancellationToken cancellation = default)
     {
-        await this._innerRunner.EnqueueMessageAsync(input).ConfigureAwait(false);
+        await ((ISuperStepRunner)this._innerRunner).EnqueueMessageAsync(input).ConfigureAwait(false);
 
         return new StreamingRun<TResult>(this);
     }
 
+    public async ValueTask<Run<TResult>> ResumeAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
+    {
+        StreamingRun<TResult> streamingRun = await this.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false);
+        cancellation.ThrowIfCancellationRequested();
+
+        return await Run<TResult>.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
+    }
+
     public async ValueTask<Run<TResult>> RunAsync(TInput input, CancellationToken cancellation = default)
     {
         StreamingRun<TResult> streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false);
@@ -204,8 +319,15 @@ internal class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult> whe
         return await Run<TResult>.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
     }
 
+    public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default)
+        => this._innerRunner.RestoreCheckpointAsync(checkpointInfo, cancellation);
+
+    internal ValueTask CheckpointAsync() => this._innerRunner.CheckpointAsync();
+
     /// <inheritdoc cref="Workflow{TInput, TResult}.RunningOutput"/>
     public TResult? RunningOutput => this._workflow.RunningOutput;
 
     ISuperStepRunner IRunnerWithOutput<TResult>.StepRunner => this._innerRunner;
+
+    public IReadOnlyList<CheckpointInfo> Checkpoints => this._innerRunner.Checkpoints;
 }

dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs

@@ -2,8 +2,11 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+
+using Microsoft.Agents.Workflows.Checkpointing;
 using Microsoft.Agents.Workflows.Execution;
 using Microsoft.Agents.Workflows.Specialized;
 using Microsoft.Extensions.Logging;
@@ -14,25 +17,26 @@ namespace Microsoft.Agents.Workflows.InProc;
 internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
 {
     private StepContext _nextStep = new();
-    private readonly Dictionary<string, ExecutorProvider<Executor>> _executorProviders;
+    private readonly Dictionary<string, ExecutorRegistration> _executorRegistrations;
     private readonly Dictionary<string, Executor> _executors = new();
     private readonly Dictionary<string, ExternalRequest> _externalRequests = new();
 
     public InProcessRunnerContext(Workflow workflow, ILogger? logger = null)
     {
-        this._executorProviders = Throw.IfNull(workflow).ExecutorProviders;
+        this._executorRegistrations = Throw.IfNull(workflow).Registrations;
     }
 
-    public async ValueTask<Executor> EnsureExecutorAsync(string executorId)
+    public async ValueTask<Executor> EnsureExecutorAsync(string executorId, IStepTracer? tracer)
     {
         if (!this._executors.TryGetValue(executorId, out var executor))
         {
-            if (!this._executorProviders.TryGetValue(executorId, out var provider))
+            if (!this._executorRegistrations.TryGetValue(executorId, out var registration))
             {
                 throw new InvalidOperationException($"Executor with ID '{executorId}' is not registered.");
             }
 
-            this._executors[executorId] = executor = provider();
+            this._executors[executorId] = executor = registration.Provider();
+            tracer?.TraceActivated(executorId);
 
             if (executor is RequestInfoExecutor requestInputExecutor)
             {
@@ -107,4 +111,65 @@ internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
         public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null)
             => RunnerContext.StateManager.ReadStateAsync<T>(ExecutorId, scopeName, key);
     }
+
+    internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default)
+    {
+        return Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellation).AsTask()));
+    }
+
+    internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellationToken = default)
+    {
+        return Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellationToken).AsTask()));
+    }
+
+    internal ValueTask<RunnerStateData> ExportStateAsync()
+    {
+        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<ExportedState>> queuedMessages = this._nextStep.ExportMessages();
+
+        RunnerStateData result = new(queuedMessages, this._externalRequests.Values.ToList());
+
+        return new(result);
+    }
+
+    internal async ValueTask RepublishUnservicedRequestsAsync(CancellationToken cancellation = default)
+    {
+        if (this.HasUnservicedRequests)
+        {
+            foreach (string requestId in this._externalRequests.Keys)
+            {
+                await this.AddEventAsync(new RequestInfoEvent(this._externalRequests[requestId]))
+                          .ConfigureAwait(false);
+            }
+        }
+    }
+
+    internal ValueTask ImportStateAsync(Checkpoint checkpoint)
+    {
+        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.");
+        }
+
+        RunnerStateData importedState = checkpoint.RunnerData;
+
+        this._nextStep = new StepContext();
+        this._nextStep.ImportMessages(importedState.QueuedMessages);
+
+        this._externalRequests.Clear();
+
+        foreach (ExternalRequest request in importedState.OutstandingRequests)
+        {
+            // TODO: Reduce the amount of data we need to store in the checkpoint by not storing the entire request object.
+            // For example, the Port object is not needed - we should be able to reconstruct it from the ID and the workflow
+            // definition.
+            this._externalRequests[request.RequestId] = request;
+        }
+
+        return default;
+    }
 }

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

@@ -24,12 +24,67 @@ public static class InProcessExecution
     /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
     /// <returns>A <see cref="ValueTask{StreamingRun}"/> that represents the asynchronous operation. The result contains a <see
     /// cref="StreamingRun"/> for managing and interacting with the streaming run.</returns>
-    public static ValueTask<StreamingRun> StreamAsync<TInput>(Workflow<TInput> workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull
+    public static ValueTask<StreamingRun> StreamAsync<TInput>(
+        Workflow<TInput> workflow,
+        TInput input,
+        CancellationToken cancellation = default) where TInput : notnull
     {
-        InProcessRunner<TInput> runner = new(workflow);
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager: null);
         return runner.StreamAsync(input, cancellation);
     }
 
+    /// <summary>
+    /// Initiates an asynchronous streaming execution using the specified input, with checkpointing.
+    /// </summary>
+    /// <remarks>The returned <see cref="StreamingRun"/> provides methods to observe and control
+    /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or
+    /// cancelled.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="input">The input message to be processed as part of the streaming run.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="ValueTask{StreamingRun}"/> that represents the asynchronous operation. The result contains a <see
+    /// cref="StreamingRun"/> for managing and interacting with the streaming run.</returns>
+    public static async ValueTask<Checkpointed<StreamingRun>> StreamAsync<TInput>(
+        Workflow<TInput> workflow,
+        TInput input,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager);
+        StreamingRun result = await runner.StreamAsync(input, cancellation).ConfigureAwait(false);
+
+        await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
+    /// <summary>
+    /// Resumes an asynchronous streaming execution for the specified input from a checkpoint.
+    /// </summary>
+    /// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
+    /// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
+    /// streaming execution will be terminated.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
+    /// run.</returns>
+    public static async ValueTask<Checkpointed<StreamingRun>> ResumeStreamAsync<TInput>(
+        Workflow<TInput> workflow,
+        CheckpointInfo fromCheckpoint,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager);
+        StreamingRun result = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
     /// <summary>
     /// Initiates an asynchronous streaming execution for the specified input.
     /// </summary>
@@ -43,12 +98,69 @@ public static class InProcessExecution
     /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
     /// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
     /// run.</returns>
-    public static ValueTask<StreamingRun<TResult>> StreamAsync<TInput, TResult>(Workflow<TInput, TResult> workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull
+    public static ValueTask<StreamingRun<TResult>> StreamAsync<TInput, TResult>(
+        Workflow<TInput, TResult> workflow,
+        TInput input,
+        CancellationToken cancellation = default) where TInput : notnull
     {
-        InProcessRunner<TInput, TResult> runner = new(workflow);
+        InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager: null);
         return runner.StreamAsync(input, cancellation);
     }
 
+    /// <summary>
+    /// Initiates an asynchronous streaming execution for the specified input, with checkpointing.
+    /// </summary>
+    /// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
+    /// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
+    /// streaming execution will be terminated.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="input">The input value to be processed by the streaming run.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
+    /// run.</returns>
+    public static async ValueTask<Checkpointed<StreamingRun<TResult>>> StreamAsync<TInput, TResult>(
+        Workflow<TInput, TResult> workflow,
+        TInput input,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager);
+        StreamingRun<TResult> result = await runner.StreamAsync(input, cancellation).ConfigureAwait(false);
+
+        await runner.CheckpointAsync().ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
+    /// <summary>
+    /// Resumes an asynchronous streaming execution of the workflow from a checkpoint.
+    /// </summary>
+    /// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
+    /// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
+    /// streaming execution will be terminated.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
+    /// run.</returns>
+    public static async ValueTask<Checkpointed<StreamingRun<TResult>>> ResumeStreamAsync<TInput, TResult>(
+        Workflow<TInput, TResult> workflow,
+        CheckpointInfo fromCheckpoint,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager);
+        StreamingRun<TResult> result = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
     /// <summary>
     /// Initiates a non-streaming execution of the workflow with the specified input.
     /// </summary>
@@ -60,12 +172,65 @@ public static class InProcessExecution
     /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
     /// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
     /// cref="Run"/> for managing and interacting with the streaming run.</returns>
-    public static ValueTask<Run> RunAsync<TInput>(Workflow<TInput> workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull
+    public static ValueTask<Run> RunAsync<TInput>(
+        Workflow<TInput> workflow,
+        TInput input,
+        CancellationToken cancellation = default) where TInput : notnull
     {
-        InProcessRunner<TInput> runner = new(workflow);
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager: null);
         return runner.RunAsync(input, cancellation);
     }
 
+    /// <summary>
+    /// Initiates a non-streaming execution of the workflow with the specified input, with checkpointing.
+    /// </summary>
+    /// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
+    /// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="input">The input message to be processed as part of the run.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
+    /// cref="Run"/> for managing and interacting with the streaming run.</returns>
+    public static async ValueTask<Checkpointed<Run>> RunAsync<TInput>(
+        Workflow<TInput> workflow,
+        TInput input,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager);
+        Run result = await runner.RunAsync(input, cancellation).ConfigureAwait(false);
+
+        await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
+    /// <summary>
+    /// Resumes a non-streaming execution of the workflow from a checkpoint.
+    /// </summary>
+    /// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
+    /// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
+    /// cref="Run"/> for managing and interacting with the streaming run.</returns>
+    public static async ValueTask<Checkpointed<Run>> ResumeAsync<TInput>(
+        Workflow<TInput> workflow,
+        CheckpointInfo fromCheckpoint,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput> runner = new(workflow, checkpointManager);
+        Run result = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
     /// <summary>
     /// Initiates a non-streaming execution of the workflow with the specified input.
     /// </summary>
@@ -78,9 +243,64 @@ public static class InProcessExecution
     /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
     /// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
     /// cref="Run"/> for managing and interacting with the streaming run.</returns>
-    public static ValueTask<Run<TResult>> RunAsync<TInput, TResult>(Workflow<TInput, TResult> workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull
+    public static ValueTask<Run<TResult>> RunAsync<TInput, TResult>(
+        Workflow<TInput, TResult> workflow,
+        TInput input,
+        CancellationToken cancellation = default) where TInput : notnull
     {
-        InProcessRunner<TInput, TResult> runner = new(workflow);
+        InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager: null);
         return runner.RunAsync(input, cancellation);
     }
+
+    /// <summary>
+    /// Initiates a non-streaming execution of the workflow with the specified input, with checkpointing.
+    /// </summary>
+    /// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
+    /// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="input">The input message to be processed as part of the run.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
+    /// cref="Run"/> for managing and interacting with the streaming run.</returns>
+    public static async ValueTask<Checkpointed<Run<TResult>>> RunAsync<TInput, TResult>(
+        Workflow<TInput, TResult> workflow,
+        TInput input,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager);
+        Run<TResult> result = await runner.RunAsync(input, cancellation).ConfigureAwait(false);
+
+        await runner.CheckpointAsync().ConfigureAwait(false);
+
+        return new(result, runner);
+    }
+
+    /// <summary>
+    /// Resumes a non-streaming execution of the workflow from a checkpoint.
+    /// </summary>
+    /// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
+    /// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
+    /// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
+    /// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
+    /// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
+    /// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
+    /// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
+    /// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
+    /// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
+    /// cref="Run"/> for managing and interacting with the streaming run.</returns>
+    public static async ValueTask<Checkpointed<Run<TResult>>> ResumeAsync<TInput, TResult>(
+        Workflow<TInput, TResult> workflow,
+        CheckpointInfo fromCheckpoint,
+        CheckpointManager checkpointManager,
+        CancellationToken cancellation = default) where TInput : notnull
+    {
+        InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager);
+        Run<TResult> result = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
+
+        return new(result, runner);
+    }
 }

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

@@ -2,25 +2,35 @@
 
 using Microsoft.Shared.Diagnostics;
 
-namespace Microsoft.Agents.Workflows.Execution;
+namespace Microsoft.Agents.Workflows;
 
 /// <summary>
 /// A unique identifier for a scope within an executor. If a scope name is not provided, it references the
 /// default scope private to the executor. Otherwise, regardless of the executorId, it references a shared
 /// scope with the specified name.
 /// </summary>
-/// <param name="executorId"></param>
-/// <param name="scopeName"></param>
-internal class ScopeId(string executorId, string? scopeName = null)
+/// <param name="executorId">The unique identifier for the executor associated with this ScopeId.</param>
+/// <param name="scopeName">The name of the scope, if any. If <see langword="null"/>, this ScopeId
+/// corresponds to the Executor's private scope.</param>
+public class ScopeId(string executorId, string? scopeName = null)
 {
+    /// <summary>
+    /// Gets the unique identifier of the executor.
+    /// </summary>
     public string ExecutorId { get; } = Throw.IfNullOrEmpty(executorId);
+
+    /// <summary>
+    /// Gets the name of the current scope, if any.
+    /// </summary>
     public string? ScopeName { get; } = scopeName;
 
+    /// <inheritdoc/>
     public override string ToString()
     {
         return $"{this.ExecutorId}/{this.ScopeName ?? "default"}";
     }
 
+    /// <inheritdoc/>
     public override bool Equals(object? obj)
     {
         if (obj is ScopeId other)
@@ -42,6 +52,7 @@ internal class ScopeId(string executorId, string? scopeName = null)
         return false;
     }
 
+    /// <inheritdoc/>
     public override int GetHashCode()
     {
         if (this.ScopeName is null)

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

@@ -0,0 +1,57 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Represents a unique key within a specific scope, combining a scope identifier and a key string.
+/// </summary>
+/// <param name="scopeId">The <see cref="ScopeId"/> associated with this key.</param>
+/// <param name="key">The unique key within the specified scope.</param>
+public class ScopeKey(ScopeId scopeId, string key)
+{
+    /// <summary>
+    /// The identifier for the scope associated with this key.
+    /// </summary>
+    public ScopeId ScopeId { get; } = Throw.IfNull(scopeId);
+
+    /// <summary>
+    /// The unique key within the specified scope.
+    /// </summary>
+    public string Key { get; } = Throw.IfNullOrEmpty(key);
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ScopeKey"/> class.
+    /// </summary>
+    /// <param name="executorId">The unique identifier for the executor.</param>
+    /// <param name="scopeName">The name of the scope, if any.</param>
+    /// <param name="key">The unique key within the specified scope.</param>
+    public ScopeKey(string executorId, string? scopeName, string key)
+        : this(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName), key)
+    { }
+
+    /// <inheritdoc/>
+    public override string ToString()
+    {
+        return $"{this.ScopeId}/{this.Key}";
+    }
+
+    /// <inheritdoc/>
+    public override bool Equals(object? obj)
+    {
+        if (obj is ScopeKey other)
+        {
+            // Unlike ScopeId, ScopeKey is equal only if both the Executor and ScopeName are the same
+            return this.ScopeId.Equals(other.ScopeId) && this.Key == other.Key;
+        }
+        return false;
+    }
+
+    /// <inheritdoc/>
+    public override int GetHashCode()
+    {
+        return HashCode.Combine(this.ScopeId, this.Key);
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs

@@ -1,6 +1,8 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
+using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.AI;
 using Microsoft.Extensions.AI.Agents;
@@ -20,7 +22,7 @@ internal class AIAgentHostExecutor : Executor
         this._emitEvents = emitEvents;
     }
 
-    private AgentThread EnsureThread()
+    private AgentThread EnsureThread(IWorkflowContext context)
     {
         if (this._thread != null)
         {
@@ -49,10 +51,48 @@ internal class AIAgentHostExecutor : Executor
         return default;
     }
 
+    private const string ThreadStateKey = nameof(AIAgentHostExecutor._thread);
+    private const string PendingMessagesStateKey = nameof(AIAgentHostExecutor._pendingMessages);
+    protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
+    {
+        Task threadTask = Task.CompletedTask;
+        if (this._thread != null)
+        {
+            JsonElement threadValue = await this._thread.SerializeAsync(cancellationToken: cancellation).ConfigureAwait(false);
+            threadTask = context.QueueStateUpdateAsync(ThreadStateKey, threadValue).AsTask();
+        }
+
+        Task messagesTask = Task.CompletedTask;
+        if (this._pendingMessages.Count > 0)
+        {
+            JsonElement messagesValue = this._pendingMessages.SerializeToJson();
+            messagesTask = context.QueueStateUpdateAsync(PendingMessagesStateKey, messagesValue).AsTask();
+        }
+
+        await Task.WhenAll(threadTask, messagesTask).ConfigureAwait(false);
+    }
+
+    protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
+    {
+        JsonElement? threadValue = await context.ReadStateAsync<JsonElement?>(ThreadStateKey).ConfigureAwait(false);
+        if (threadValue.HasValue)
+        {
+            this._thread = await this._agent.DeserializeThreadAsync(threadValue.Value, cancellationToken: cancellation)
+                                            .ConfigureAwait(false);
+        }
+
+        JsonElement? messagesValue = await context.ReadStateAsync<JsonElement?>(PendingMessagesStateKey).ConfigureAwait(false);
+        if (messagesValue.HasValue)
+        {
+            List<ChatMessage> messages = messagesValue.Value.DeserializeMessageList();
+            this._pendingMessages.AddRange(messages);
+        }
+    }
+
     public async ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context)
     {
         bool emitEvents = token.EmitEvents.HasValue ? token.EmitEvents.Value : this._emitEvents;
-        IAsyncEnumerable<AgentRunResponseUpdate> agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread());
+        IAsyncEnumerable<AgentRunResponseUpdate> agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread(context));
 
         List<AgentRunResponseUpdate> updates = new();
         await foreach (AgentRunResponseUpdate update in agentStream.ConfigureAwait(false))

dotnet/src/Microsoft.Agents.Workflows/Specialized/IOutputSink.cs

@@ -2,7 +2,7 @@
 
 namespace Microsoft.Agents.Workflows.Specialized;
 
-internal interface IOutputSink<TResult>
+internal interface IOutputSink<TResult> : IIdentified
 {
     TResult? Result { get; }
 }

dotnet/src/Microsoft.Agents.Workflows/Specialized/WorkflowJsonUtilities.cs

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+using Microsoft.Extensions.AI;
+
+namespace Microsoft.Agents.Workflows.Specialized;
+
+internal static partial class WorkflowJsonUtilities
+{
+    public static WorkflowJsonContext Default { get; } = new();
+
+    [JsonSerializable(typeof(ChatMessage))]
+    [JsonSerializable(typeof(List<ChatMessage>))]
+    internal sealed partial class WorkflowJsonContext : JsonSerializerContext;
+
+    public static JsonElement SerializeToJson(this List<ChatMessage> messages)
+    {
+        return JsonSerializer.SerializeToElement(messages, Default.ListChatMessage);
+    }
+
+    public static JsonElement SerializeToJson(this IEnumerable<ChatMessage> messages)
+        => messages.ToList().SerializeToJson();
+
+    public static List<ChatMessage> DeserializeMessageList(this JsonElement element)
+    {
+        return element.Deserialize<List<ChatMessage>>(Default.ListChatMessage) ?? [];
+    }
+}

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

@@ -2,6 +2,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace Microsoft.Agents.Workflows;
 
@@ -24,7 +25,7 @@ public static class StreamingAggregators
 {
     /// <summary>
     /// Creates a streaming aggregator that returns the result of applying the specified conversion function to the
-    /// first input value, or a default value if no input is provided.
+    /// first input value.
     /// </summary>
     /// <remarks>Subsequent inputs after the first are ignored by the aggregator. This method is useful for
     /// scenarios where only the first occurrence in a stream is relevant. The conversion function is invoked at most
@@ -33,13 +34,11 @@ public static class StreamingAggregators
     /// <typeparam name="TResult">The type of the result produced by the conversion function.</typeparam>
     /// <param name="conversion">A function that converts an input value of type <typeparamref name="TInput"/> to a result of type <typeparamref
     /// name="TResult"/>. This function is applied to the first input received.</param>
-    /// <param name="defaultValue">The value to return if no input is provided. </param>
-    /// <returns>A <see cref="StreamingAggregator{TInput, TResult}"/> that yields the converted result of the first input, or the
-    /// specified default value if no input is received.</returns>
-    public static StreamingAggregator<TInput, TResult> First<TInput, TResult>(Func<TInput, TResult> conversion, TResult? defaultValue = default)
+    /// <returns>A <see cref="StreamingAggregator{TInput, TResult}"/> that yields the converted result of the first input.</returns>
+    public static StreamingAggregator<TInput, TResult> First<TInput, TResult>(Func<TInput, TResult> conversion)
     {
         bool hasRun = false;
-        TResult? local = defaultValue;
+        TResult? local = default;
 
         return Aggregate;
 
@@ -48,6 +47,7 @@ public static class StreamingAggregators
             if (!hasRun)
             {
                 local = conversion(input);
+                hasRun = true;
             }
 
             return local;
@@ -55,15 +55,11 @@ public static class StreamingAggregators
     }
 
     /// <summary>
-    /// Creates a streaming aggregator that returns the first input element, or a specified default value if no elements
-    /// are provided.
+    /// Creates a streaming aggregator that returns the first input element.
     /// </summary>
     /// <typeparam name="TInput">The type of the input elements to aggregate.</typeparam>
-    /// <param name="defaultValue">The value to return if the input sequence contains no elements.</param>
-    /// <returns>A <see cref="StreamingAggregator{TInput, TInput}"/> that yields the first input element, or <paramref
-    /// name="defaultValue"/> if the sequence is empty.</returns>
-    public static StreamingAggregator<TInput, TInput> First<TInput>(TInput? defaultValue = default)
-        => First<TInput, TInput>(input => input, defaultValue);
+    /// <returns>A <see cref="StreamingAggregator{TInput, TInput}"/> that yields the first input element.</returns>
+    public static StreamingAggregator<TInput, TInput> First<TInput>() => First<TInput, TInput>(input => input);
 
     /// <summary>
     /// Creates a streaming aggregator that returns the result of applying the specified conversion to the most recent
@@ -72,12 +68,10 @@ public static class StreamingAggregators
     /// <typeparam name="TInput">The type of the input elements to be aggregated.</typeparam>
     /// <typeparam name="TResult">The type of the result produced by the conversion function.</typeparam>
     /// <param name="conversion">A function that converts each input value to a result. Cannot be null.</param>
-    /// <param name="defaultValue">The initial result value to use before any input is processed.</param>
-    /// <returns>A streaming aggregator that yields the converted value of the last input received, or the specified default
-    /// value if no input has been processed.</returns>
-    public static StreamingAggregator<TInput, TResult> Last<TInput, TResult>(Func<TInput, TResult> conversion, TResult? defaultValue = default)
+    /// <returns>A streaming aggregator that yields the converted value of the last input received.</returns>
+    public static StreamingAggregator<TInput, TResult> Last<TInput, TResult>(Func<TInput, TResult> conversion)
     {
-        TResult? local = defaultValue;
+        TResult? local = default;
 
         return Aggregate;
 
@@ -89,15 +83,11 @@ public static class StreamingAggregators
     }
 
     /// <summary>
-    /// Creates a streaming aggregator that returns the last element in a sequence, or a specified default value if the
-    /// sequence is empty.
+    /// Creates a streaming aggregator that returns the last element in a sequence.
     /// </summary>
     /// <typeparam name="TInput">The type of elements in the input sequence.</typeparam>
-    /// <param name="defaultValue">The value to return if the input sequence contains no elements.</param>
-    /// <returns>A <see cref="StreamingAggregator{TInput, TInput}"/> that yields the last element of the sequence, or <paramref
-    /// name="defaultValue"/> if the sequence is empty.</returns>
-    public static StreamingAggregator<TInput, TInput> Last<TInput>(TInput? defaultValue = default)
-        => Last<TInput, TInput>(input => input, defaultValue);
+    /// <returns>A <see cref="StreamingAggregator{TInput, TInput}"/> that yields the last element of the sequence.</returns>
+    public static StreamingAggregator<TInput, TInput> Last<TInput>() => Last<TInput, TInput>(input => input);
 
     /// <summary>
     /// Creates a streaming aggregator that produces the union of results by applying a conversion function to each
@@ -110,14 +100,11 @@ public static class StreamingAggregators
     /// far.</returns>
     public static StreamingAggregator<TInput, IEnumerable<TResult>> Union<TInput, TResult>(Func<TInput, TResult> conversion)
     {
-        List<TResult> results = new();
-
         return Aggregate;
 
         IEnumerable<TResult> Aggregate(TInput input, IEnumerable<TResult>? runningResult)
         {
-            results.Add(conversion(input));
-            return results;
+            return runningResult != null ? runningResult.Append(conversion(input)) : [conversion(input)];
         }
     }
 
@@ -130,5 +117,12 @@ public static class StreamingAggregators
     /// <returns>A StreamingAggregator that, when applied to multiple input sequences, returns an IEnumerable containing the
     /// union of all elements from those sequences.</returns>
     public static StreamingAggregator<TInput, IEnumerable<TInput>> Union<TInput>()
-        => Union<TInput, TInput>(input => input);
+    {
+        return Aggregate;
+
+        IEnumerable<TInput> Aggregate(TInput input, IEnumerable<TInput>? runningResult)
+        {
+            return runningResult != null ? runningResult.Append(input) : new[] { input };
+        }
+    }
 }

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

@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Event triggered when a SuperStep completed.
+/// </summary>
+/// <param name="stepNumber">The zero-based index of the SuperStep associated with this event.</param>
+/// <param name="completionInfo">Debug information about the state of the system on SuperStep completion.</param>
+public sealed class SuperStepCompletedEvent(int stepNumber, SuperStepCompletionInfo? completionInfo = null) : SuperStepEvent(stepNumber, data: completionInfo)
+{
+    /// <summary>
+    /// Gets the debug information about the state of the system on SuperStep completion.
+    /// </summary>
+    public SuperStepCompletionInfo? CompletionInfo => completionInfo;
+}

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

@@ -0,0 +1,44 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Debug information about the SuperStep that finished running.
+/// </summary>
+public sealed class SuperStepCompletionInfo(HashSet<string> activatedExecutors, HashSet<string>? instantiatedExecutors = null)
+{
+    /// <summary>
+    /// The unique identifiers of <see cref="Executor"/> instances that processed messages during this SuperStep
+    /// </summary>
+    public HashSet<string> ActivatedExecutors { get; } = Throw.IfNull(activatedExecutors);
+
+    /// <summary>
+    /// The unique identifiers of <see cref="Executor"/> instances newly created during this SuperStep
+    /// </summary>
+    public HashSet<string> InstantiatedExecutors { get; } = instantiatedExecutors ?? [];
+
+    /// <summary>
+    /// A flag indicating whether the managed state was written to during this SuperStep. If the run was started
+    /// with checkpointing, any updated during the checkpointing process are also included.
+    /// </summary>
+    public bool StateUpdated { get; init; }
+
+    /// <summary>
+    /// A flag indicating whether there are messages pending delivery after this SuperStep.
+    /// </summary>
+    public bool HasPendingMessages { get; init; }
+
+    /// <summary>
+    /// A flag indicating whether there are requests pending delivery after this SuperStep.
+    /// </summary>
+    public bool HasPendingRequests { get; init; }
+
+    /// <summary>
+    /// Gets the <see cref="CheckpointInfo"/> corresponding to the checkpoint created at the end of this SuperStep.
+    /// <see langword="null"/> if checkpointing was not enabled when the run was started.
+    /// </summary>
+    public CheckpointInfo? Checkpoint { get; init; } = null;
+}

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

@@ -0,0 +1,25 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Base class for SuperStep-scoped events, for example, <see cref="SuperStepCompletedEvent"/>
+/// </summary>
+public class SuperStepEvent(int stepNumber, object? data = null) : WorkflowEvent(data)
+{
+    /// <summary>
+    /// The zero-based index of the SuperStep associated with this event.
+    /// </summary>
+    public int StepNumber => stepNumber;
+
+    /// <inheritdoc/>
+    public override string ToString()
+    {
+        if (this.Data != null)
+        {
+            return $"{this.GetType().Name}(Step = {this.StepNumber}, Data: {this.Data.GetType()} = {this.Data})";
+        }
+
+        return $"{this.GetType().Name}(Step = {this.StepNumber})";
+    }
+}

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

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Debug information about the SuperStep starting to run.
+/// </summary>
+public sealed class SuperStepStartInfo(HashSet<string>? sendingExecutors = null)
+{
+    /// <summary>
+    /// The unique identifiers of <see cref="Executor"/> instances that sent messages during the previous SuperStep.
+    /// </summary>
+    public HashSet<string> SendingExecutors { get; } = sendingExecutors ?? [];
+
+    /// <summary>
+    /// Gets a value indicating whether there are any external messages queued during the previous SuperStep.
+    /// </summary>
+    public bool HasExternalMessages { get; init; }
+}

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

@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Event triggered when a SuperStep started.
+/// </summary>
+/// <param name="stepNumber">The zero-based index of the SuperStep associated with this event.</param>
+/// <param name="startInfo">Debug information about the state of the system on SuperStep start.</param>
+public sealed class SuperStepStartedEvent(int stepNumber, SuperStepStartInfo? startInfo = null) : SuperStepEvent(stepNumber, data: startInfo)
+{
+    /// <summary>
+    /// Gets the debug information about the state of the system on SuperStep start.
+    /// </summary>
+    public SuperStepStartInfo? StartInfo => startInfo;
+}

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

@@ -15,7 +15,7 @@ public class Workflow
     /// <summary>
     /// A dictionary of executor providers, keyed by executor ID.
     /// </summary>
-    public Dictionary<string, ExecutorProvider<Executor>> ExecutorProviders { get; internal init; } = new();
+    internal Dictionary<string, ExecutorRegistration> Registrations { get; init; } = new();
 
     /// <summary>
     /// Gets the collection of edges grouped by their source node identifier.
@@ -26,7 +26,7 @@ public class Workflow
     /// Gets the collection of external request ports, keyed by their ID.
     /// </summary>
     /// <remarks>
-    /// Each port has a corresponding entry in the <see cref="ExecutorProviders"/> dictionary.
+    /// Each port has a corresponding entry in the <see cref="Registrations"/> dictionary.
     /// </remarks>
     public Dictionary<string, InputPort> Ports { get; internal init; } = new();
 
@@ -73,7 +73,7 @@ public class Workflow<T> : Workflow
 
         return new Workflow<T, TResult>(this.StartExecutorId, outputSource)
         {
-            ExecutorProviders = this.ExecutorProviders,
+            Registrations = this.Registrations,
             Edges = this.Edges,
             Ports = this.Ports
         };
@@ -96,6 +96,11 @@ public class Workflow<TInput, TResult> : Workflow<TInput>
         this._output = Throw.IfNull(outputSource);
     }
 
+    /// <summary>
+    /// Gets the unique identifier of the output collector.
+    /// </summary>
+    public string OutputCollectorId => this._output.Id;
+
     /// <summary>
     /// The running (partial) output of the workflow, if any.
     /// </summary>

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

@@ -7,14 +7,6 @@ using Microsoft.Shared.Diagnostics;
 
 namespace Microsoft.Agents.Workflows;
 
-/// <summary>
-/// A factory method that produces an executor instance.
-/// </summary>
-/// <typeparam name="TExecutor">The executor type.</typeparam>
-/// <returns>A new <typeparamref name="TExecutor"/> instance.</returns>
-public delegate TExecutor ExecutorProvider<out TExecutor>()
-    where TExecutor : Executor;
-
 /// <summary>
 /// Provides a builder for constructing and configuring a workflow by defining executors and the connections between
 /// them.
@@ -30,7 +22,7 @@ public class WorkflowBuilder
         public override string ToString() => $"{this.SourceId} -> {this.TargetId}";
     }
 
-    private readonly Dictionary<string, ExecutorProvider<Executor>> _executors = new();
+    private readonly Dictionary<string, ExecutorRegistration> _executors = new();
     private readonly Dictionary<string, HashSet<Edge>> _edges = new();
     private readonly HashSet<string> _unboundExecutors = new();
     private readonly HashSet<EdgeId> _conditionlessEdges = new();
@@ -49,20 +41,17 @@ public class WorkflowBuilder
 
     private ExecutorIsh Track(ExecutorIsh executorish)
     {
-        ExecutorProvider<Executor> provider = executorish.ExecutorProvider;
-
         // If the executor is unbound, create an entry for it, unless it already exists.
         // Otherwise, update the entry for it, and remove the unbound tag
         if (executorish.IsUnbound && !this._executors.ContainsKey(executorish.Id))
         {
             // If this is an unbound executor, we need to track it separately
             this._unboundExecutors.Add(executorish.Id);
-            this._executors[executorish.Id] = provider;
         }
         else if (!executorish.IsUnbound)
         {
             // If we already have an executor with this ID, we need to update it (todo: should we throw on double binding?)
-            this._executors[executorish.Id] = provider;
+            this._executors[executorish.Id] = executorish.Registration;
         }
 
         if (executorish.ExecutorType == ExecutorIsh.Type.InputPort)
@@ -74,11 +63,6 @@ public class WorkflowBuilder
         return executorish;
     }
 
-    private void UpdateExecutor(string id, ExecutorProvider<Executor> provider)
-    {
-        this._executors[id] = provider;
-    }
-
     /// <summary>
     /// Binds the specified executor to the workflow, allowing it to participate in workflow execution.
     /// </summary>
@@ -93,7 +77,7 @@ public class WorkflowBuilder
                 $"Executor with ID '{executor.Id}' is already bound or does not exist in the workflow.");
         }
 
-        this._executors[executor.Id] = () => executor;
+        this._executors[executor.Id] = new ExecutorIsh(executor).Registration;
         this._unboundExecutors.Remove(executor.Id);
         return this;
     }
@@ -214,14 +198,14 @@ public class WorkflowBuilder
         }
 
         // Grab the start node, and make sure it has the right type?
-        if (!this._executors.TryGetValue(this._startExecutorId, out ExecutorProvider<Executor>? startProvider))
+        if (!this._executors.TryGetValue(this._startExecutorId, out ExecutorRegistration? startRegistration))
         {
             // TODO: This should never be able to be hit
             throw new InvalidOperationException($"Start executor with ID '{this._startExecutorId}' is not bound.");
         }
 
-        // TODO: Delay-instantiate the start executor, and ensure it is of type T.
-        Executor startExecutor = startProvider();
+        // TODO: Delay-instantiate the start executor, and ensure it take input of type T
+        Executor startExecutor = startRegistration.Provider();
 
         if (!startExecutor.InputTypes.Any(t => t.IsAssignableFrom(typeof(T))))
         {
@@ -233,7 +217,7 @@ public class WorkflowBuilder
 
         return new Workflow<T>(this._startExecutorId) // Why does it not see the default ctor?
         {
-            ExecutorProviders = this._executors,
+            Registrations = this._executors,
             Edges = this._edges,
             Ports = this._inputPorts
         };

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

@@ -0,0 +1,180 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Microsoft.Agents.Workflows.Checkpointing;
+using Microsoft.Agents.Workflows.Sample;
+using Microsoft.Agents.Workflows.Specialized;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+
+namespace Microsoft.Agents.Workflows.UnitTests;
+
+public class RepresentationTests
+{
+    private sealed class TestExecutor : Executor
+    {
+        protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder;
+    }
+
+    private sealed class TestAgent : AIAgent
+    {
+        public override Task<AgentRunResponse> RunAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+
+        public override IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+    }
+
+    private static InputPort TestInputPort =>
+        InputPort.Create<FunctionCallContent, FunctionResultContent>("ExternalFunction");
+
+    private static List<T> ListAggregator<T>(List<T>? current, T incoming)
+    {
+        if (current is null)
+        {
+            return [incoming];
+        }
+
+        current.Add(incoming);
+        return current;
+    }
+
+    private static void RunExecutorishInfoMatchTest(ExecutorIsh target)
+    {
+        ExecutorRegistration registration = target.Registration;
+        ExecutorInfo info = registration.ToExecutorInfo();
+
+        info.IsMatch(registration.Provider()).Should().BeTrue();
+    }
+
+    [Fact]
+    public void Test_Executorish_Infos()
+    {
+        int testsRun = 0;
+        RunExecutorishTest(new TestExecutor());
+        RunExecutorishTest(TestInputPort);
+        RunExecutorishTest(new TestAgent());
+
+        if (Enum.GetValues(typeof(ExecutorIsh.Type)).Length > testsRun + 1)
+        {
+            Assert.Fail("Not all ExecutorIsh types were tested.");
+        }
+
+        void RunExecutorishTest(ExecutorIsh executorish)
+        {
+            RunExecutorishInfoMatchTest(executorish);
+            testsRun++;
+        }
+    }
+
+    [Fact]
+    public void Test_SpecializedExecutor_Infos()
+    {
+        RunExecutorishInfoMatchTest(new AIAgentHostExecutor(new TestAgent()));
+        RunExecutorishInfoMatchTest(new RequestInfoExecutor(TestInputPort));
+
+        OutputCollectorExecutor<ChatMessage, IEnumerable<ChatMessage>> outputCollector = new(StreamingAggregators.Union<ChatMessage>());
+        RunExecutorishInfoMatchTest(outputCollector);
+    }
+
+    private static string Source(string id) => $"Source/{id}";
+    private static string Source(int id) => $"Source/{id}";
+    private static string Sink(string id) => $"Sink/{id}";
+    private static string Sink(int id) => $"Sink/{id}";
+
+    private static Func<object?, bool> Condition() => Condition<object>();
+    private static Func<TIn?, bool> Condition<TIn>() => _ => true;
+
+    private static Func<object?, int, IEnumerable<int>> EdgeAssigner() => EdgeAssigner<object>();
+    private static Func<TIn?, int, IEnumerable<int>> EdgeAssigner<TIn>() => (_, _) => [];
+
+    [Fact]
+    public void Test_EdgeInfos()
+    {
+        // Direct Edges
+        Edge directEdgeNoCondition = new(new DirectEdgeData(Source(1), Sink(2)));
+        RunEdgeInfoMatchTest(directEdgeNoCondition);
+
+        Edge directEdgeNoCondition2 = new(new DirectEdgeData(Source(1), Sink(2)));
+        RunEdgeInfoMatchTest(directEdgeNoCondition, directEdgeNoCondition2);
+
+        Edge directEdgeNoCondition3 = new(new DirectEdgeData(Source(3), Sink(4)));
+        RunEdgeInfoMatchTest(directEdgeNoCondition, directEdgeNoCondition3, expect: false);
+
+        Edge directEdgeWithCondition = new(new DirectEdgeData(Source(3), Sink(4), Condition()));
+        RunEdgeInfoMatchTest(directEdgeWithCondition);
+        RunEdgeInfoMatchTest(directEdgeNoCondition2, directEdgeWithCondition, expect: false);
+        RunEdgeInfoMatchTest(directEdgeNoCondition3, directEdgeWithCondition, expect: false);
+
+        // FanOut Edges
+        Edge fanOutEdgeNoAssigner = new(new FanOutEdgeData(Source(1), [Sink(2), Sink(3), Sink(4)]));
+        RunEdgeInfoMatchTest(fanOutEdgeNoAssigner);
+
+        Edge fanOutEdgeNoAssigner2 = new(new FanOutEdgeData(Source(1), [Sink(2), Sink(3), Sink(4)]));
+        RunEdgeInfoMatchTest(fanOutEdgeNoAssigner, fanOutEdgeNoAssigner2);
+
+        Edge fanOutEdgeNoAssigner3 = new(new FanOutEdgeData(Source(1), [Sink(3), Sink(4), Sink(2)]));
+        RunEdgeInfoMatchTest(fanOutEdgeNoAssigner, fanOutEdgeNoAssigner3, expect: false); // Order matters (though without Assigner maybe it shouldn't?)
+
+        Edge fanOutEdgeNoAssigner4 = new(new FanOutEdgeData(Source(1), [Sink(2), Sink(3), Sink(5)]));
+        Edge fanOutEdgeNoAssigner5 = new(new FanOutEdgeData(Source(2), [Sink(2), Sink(3), Sink(4)]));
+        RunEdgeInfoMatchTest(fanOutEdgeNoAssigner, fanOutEdgeNoAssigner4, expect: false); // Identity matters
+        RunEdgeInfoMatchTest(fanOutEdgeNoAssigner, fanOutEdgeNoAssigner5, expect: false);
+
+        Edge fanOutEdgeWithAssigner = new(new FanOutEdgeData(Source(1), [Sink(2), Sink(3), Sink(4)], EdgeAssigner()));
+        RunEdgeInfoMatchTest(fanOutEdgeWithAssigner);
+
+        // FanIn Edges
+        Edge fanInEdge = new(new FanInEdgeData([Source(1), Source(2), Source(3)], Sink(1)));
+        RunEdgeInfoMatchTest(fanInEdge);
+
+        Edge fanInEdge2 = new(new FanInEdgeData([Source(1), Source(2), Source(3)], Sink(1)));
+        RunEdgeInfoMatchTest(fanInEdge, fanInEdge2);
+
+        Edge fanInEdge3 = new(new FanInEdgeData([Source(2), Source(3), Source(1)], Sink(1)));
+        RunEdgeInfoMatchTest(fanInEdge, fanInEdge3, expect: false); // Order matters (though for FanIn maybe it shouldn't?)
+
+        Edge fanInEdge4 = new(new FanInEdgeData([Source(1), Source(2), Source(4)], Sink(1)));
+        Edge fanInEdge5 = new(new FanInEdgeData([Source(1), Source(2), Source(3)], Sink(2)));
+        RunEdgeInfoMatchTest(fanInEdge, fanInEdge4, expect: false); // Identity matters
+        RunEdgeInfoMatchTest(fanInEdge, fanInEdge5, expect: false);
+
+        void RunEdgeInfoMatchTest(Edge edge, Edge? comparatorEdge = null, bool expect = true)
+        {
+            comparatorEdge ??= edge;
+
+            EdgeInfo info = edge.ToEdgeInfo();
+            info.IsMatch(comparatorEdge).Should().Be(expect);
+        }
+    }
+
+    [Fact]
+    public void Test_Sample_WorkflowInfos()
+    {
+        RunWorkflowInfoMatchTest(Step1EntryPoint.WorkflowInstance);
+        RunWorkflowInfoMatchTest(Step2EntryPoint.WorkflowInstance);
+        RunWorkflowInfoMatchTest(Step3EntryPoint.WorkflowInstance);
+        RunWorkflowInfoMatchTest(Step4EntryPoint.WorkflowInstance);
+        // Step 5 reuses the workflow from Step 4, so we don't need to test it separately.
+        RunWorkflowInfoMatchTest(Step6EntryPoint.CreateWorkflow(2));
+        // Step 7 reuses the workflow from Step 6, so we don't need to test it separately.
+
+        RunWorkflowInfoMatchTest(Step1EntryPoint.WorkflowInstance, Step2EntryPoint.WorkflowInstance, expect: false);
+
+        void RunWorkflowInfoMatchTest<TInput>(Workflow<TInput> workflow, Workflow<TInput>? comparator = null, bool expect = true)
+        {
+            comparator ??= workflow;
+
+            WorkflowInfo info = workflow.ToWorkflowInfo();
+            info.IsMatch(comparator).Should().Be(expect);
+        }
+    }
+}

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs

@@ -8,16 +8,23 @@ namespace Microsoft.Agents.Workflows.Sample;
 
 internal static class Step1EntryPoint
 {
+    public static Workflow<string> WorkflowInstance
+    {
+        get
+        {
+            UppercaseExecutor uppercase = new();
+            ReverseTextExecutor reverse = new();
+
+            WorkflowBuilder builder = new(uppercase);
+            builder.AddEdge(uppercase, reverse);
+
+            return builder.Build<string>();
+        }
+    }
+
     public static async ValueTask RunAsync(TextWriter writer)
     {
-        UppercaseExecutor uppercase = new();
-        ReverseTextExecutor reverse = new();
-
-        WorkflowBuilder builder = new(uppercase);
-        builder.AddEdge(uppercase, reverse);
-
-        Workflow<string> workflow = builder.Build<string>();
-        StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Hello, World!").ConfigureAwait(false);
+        StreamingRun run = await InProcessExecution.StreamAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false);
 
         await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
         {

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs

@@ -3,21 +3,15 @@
 using System.IO;
 using System.Threading.Tasks;
 
+using static Microsoft.Agents.Workflows.Sample.Step1EntryPoint;
+
 namespace Microsoft.Agents.Workflows.Sample;
 
 internal static class Step1aEntryPoint
 {
     public static async ValueTask RunAsync(TextWriter writer)
     {
-        UppercaseExecutor uppercase = new();
-        ReverseTextExecutor reverse = new();
-
-        WorkflowBuilder builder = new(uppercase);
-        builder.AddEdge(uppercase, reverse);
-
-        Workflow<string> workflow = builder.Build<string>();
-
-        Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!").ConfigureAwait(false);
+        Run run = await InProcessExecution.RunAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false);
 
         Assert.Equal(RunStatus.Completed, run.Status);
 

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs

@@ -10,20 +10,26 @@ namespace Microsoft.Agents.Workflows.Sample;
 
 internal static class Step2EntryPoint
 {
+    public static Workflow<string> WorkflowInstance
+    {
+        get
+        {
+            string[] spamKeywords = { "spam", "advertisement", "offer" };
+
+            DetectSpamExecutor detectSpam = new(spamKeywords);
+            RespondToMessageExecutor respondToMessage = new();
+            RemoveSpamExecutor removeSpam = new();
+
+            return new WorkflowBuilder(detectSpam)
+                .AddEdge(detectSpam, respondToMessage, isSpam => isSpam is false) // If not spam, respond
+                .AddEdge(detectSpam, removeSpam, isSpam => isSpam is true) // If spam, remove
+                .Build<string>();
+        }
+    }
+
     public static async ValueTask<string> RunAsync(TextWriter writer, string input = "This is a spam message.")
     {
-        string[] spamKeywords = { "spam", "advertisement", "offer" };
-
-        DetectSpamExecutor detectSpam = new(spamKeywords);
-        RespondToMessageExecutor respondToMessage = new();
-        RemoveSpamExecutor removeSpam = new();
-
-        Workflow<string> workflow = new WorkflowBuilder(detectSpam)
-            .AddEdge(detectSpam, respondToMessage, isSpam => isSpam is false) // If not spam, respond
-            .AddEdge(detectSpam, removeSpam, isSpam => isSpam is true) // If spam, remove
-            .Build<string>();
-
-        StreamingRun handle = await InProcessExecution.StreamAsync(workflow, input).ConfigureAwait(false);
+        StreamingRun handle = await InProcessExecution.StreamAsync(WorkflowInstance, input).ConfigureAwait(false);
         await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false))
         {
             switch (evt)

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

@@ -2,6 +2,7 @@
 
 using System;
 using System.IO;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Agents.Workflows.Reflection;
 
@@ -9,17 +10,23 @@ namespace Microsoft.Agents.Workflows.Sample;
 
 internal static class Step3EntryPoint
 {
+    public static Workflow<NumberSignal> WorkflowInstance
+    {
+        get
+        {
+            GuessNumberExecutor guessNumber = new(1, 100);
+            JudgeExecutor judge = new(42); // Let's say the target number is 42
+
+            return new WorkflowBuilder(guessNumber)
+                .AddEdge(guessNumber, judge)
+                .AddEdge(judge, guessNumber)
+                .Build<NumberSignal>();
+        }
+    }
+
     public static async ValueTask<string> RunAsync(TextWriter writer)
     {
-        GuessNumberExecutor guessNumber = new(1, 100);
-        JudgeExecutor judge = new(42); // Let's say the target number is 42
-
-        Workflow<NumberSignal> workflow = new WorkflowBuilder(guessNumber)
-            .AddEdge(guessNumber, judge)
-            .AddEdge(judge, guessNumber)
-            .Build<NumberSignal>();
-
-        StreamingRun run = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false);
+        StreamingRun run = await InProcessExecution.StreamAsync(WorkflowInstance, NumberSignal.Init).ConfigureAwait(false);
 
         await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
         {
@@ -88,6 +95,8 @@ internal sealed class JudgeExecutor : ReflectingExecutor<JudgeExecutor>, IMessag
 {
     private readonly int _targetNumber;
 
+    internal int? Tries { get; private set; }
+
     public JudgeExecutor(int targetNumber)
     {
         this._targetNumber = targetNumber;
@@ -95,6 +104,15 @@ internal sealed class JudgeExecutor : ReflectingExecutor<JudgeExecutor>, IMessag
 
     public async ValueTask<NumberSignal> HandleAsync(int message, IWorkflowContext context)
     {
+        if (!this.Tries.HasValue)
+        {
+            this.Tries = 1;
+        }
+        else
+        {
+            this.Tries++;
+        }
+
         NumberSignal result;
         if (message == this._targetNumber)
         {
@@ -111,4 +129,14 @@ internal sealed class JudgeExecutor : ReflectingExecutor<JudgeExecutor>, IMessag
 
         return result;
     }
+
+    protected internal override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
+    {
+        return context.QueueStateUpdateAsync("TryCount", this.Tries);
+    }
+
+    protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
+    {
+        this.Tries = await context.ReadStateAsync<int>("TryCount").ConfigureAwait(false);
+    }
 }

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs

@@ -6,18 +6,30 @@ using System.Threading.Tasks;
 
 namespace Microsoft.Agents.Workflows.Sample;
 
-internal static class Step5EntryPoint
+internal static class Step4EntryPoint
 {
-    public static async ValueTask<string> RunAsync(TextWriter writer, Func<string, int> userGuessCallback)
+    public static Workflow<NumberSignal, string> CreateWorkflowInstance(out JudgeExecutor judge)
     {
         InputPort guessNumber = InputPort.Create<NumberSignal, int>("GuessNumber");
-        JudgeExecutor judge = new(42); // Let's say the target number is 42
+        judge = new(42); // Let's say the target number is 42
 
-        Workflow<NumberSignal, string> workflow = new WorkflowBuilder(guessNumber)
+        return new WorkflowBuilder(guessNumber)
             .AddEdge(guessNumber, judge)
             .AddEdge(judge, guessNumber, (message) => message is NumberSignal signal && signal != NumberSignal.Matched)
             .BuildWithOutput<NumberSignal, string>(judge, ComputeStreamingOutput, (NumberSignal s, string? _) => s == NumberSignal.Matched);
+    }
 
+    public static Workflow<NumberSignal, string> WorkflowInstance
+    {
+        get
+        {
+            return CreateWorkflowInstance(out _);
+        }
+    }
+
+    public static async ValueTask<string> RunAsync(TextWriter writer, Func<string, int> userGuessCallback)
+    {
+        Workflow<NumberSignal, string> workflow = WorkflowInstance;
         StreamingRun<string> handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false);
 
         await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false))

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs

@@ -0,0 +1,129 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+
+namespace Microsoft.Agents.Workflows.Sample;
+
+internal static class Step5EntryPoint
+{
+    private static CheckpointManager CheckpointManager { get; } = new();
+
+    public static async ValueTask<string> RunAsync(TextWriter writer, Func<string, int> userGuessCallback)
+    {
+        Workflow<NumberSignal, string> workflow = Step4EntryPoint.CreateWorkflowInstance(out JudgeExecutor judge);
+        Checkpointed<StreamingRun<string>> checkpointed =
+            await InProcessExecution.StreamAsync(workflow, NumberSignal.Init, CheckpointManager)
+                                    .ConfigureAwait(false);
+
+        List<CheckpointInfo> checkpoints = new();
+        CancellationTokenSource cancellationSource = new();
+
+        StreamingRun<string> handle = checkpointed.Run;
+        string? result = await RunStreamToHaltOrMaxStepAsync(6).ConfigureAwait(false);
+
+        result.Should().BeNull();
+        checkpoints.Should().HaveCount(6, "we should have two checkpoints, one for each step");
+        judge.Tries.Should().Be(2);
+
+        await checkpointed.RestoreCheckpointAsync(checkpoints[2], CancellationToken.None).ConfigureAwait(false);
+        judge.Tries.Should().Be(1);
+
+        cancellationSource.Dispose();
+        cancellationSource = new();
+
+        checkpoints.Clear();
+        result = await RunStreamToHaltOrMaxStepAsync().ConfigureAwait(false);
+
+        result.Should().NotBeNull();
+        checkpoints.Should().HaveCount(6);
+
+        cancellationSource.Dispose();
+
+        return result;
+
+        async ValueTask<string?> RunStreamToHaltOrMaxStepAsync(int? maxStep = null)
+        {
+            await foreach (WorkflowEvent evt in handle.WatchStreamAsync(cancellationSource.Token).ConfigureAwait(false))
+            {
+                switch (evt)
+                {
+                    case SuperStepCompletedEvent stepCompletedEvt:
+                        CheckpointInfo? checkpoint = stepCompletedEvt.CompletionInfo!.Checkpoint;
+                        if (checkpoint != null)
+                        {
+                            checkpoints.Add(checkpoint);
+                        }
+
+                        if (maxStep.HasValue && stepCompletedEvt.StepNumber >= maxStep.Value - 1)
+                        {
+                            cancellationSource.Cancel();
+                        }
+                        break;
+                    case RequestInfoEvent requestInputEvt:
+                        ExternalResponse response = ExecuteExternalRequest(requestInputEvt.Request, userGuessCallback, workflow.RunningOutput);
+                        await handle.SendResponseAsync(response).ConfigureAwait(false);
+                        break;
+                    case WorkflowCompletedEvent workflowCompleteEvt:
+                        // The workflow has completed successfully, return the result
+                        string workflowResult = workflowCompleteEvt.Data!.ToString()!;
+                        writer.WriteLine($"Result: {workflowResult}");
+                        return workflowResult;
+                    case ExecutorCompleteEvent executorCompleteEvt:
+                        writer.WriteLine($"'{executorCompleteEvt.ExecutorId}: {executorCompleteEvt.Data}");
+                        break;
+                }
+            }
+
+            if (cancellationSource.IsCancellationRequested)
+            {
+                return null;
+            }
+
+            throw new InvalidOperationException("Workflow failed to yield the completion event.");
+        }
+    }
+
+    private static ExternalResponse ExecuteExternalRequest(
+        ExternalRequest request,
+        Func<string, int> userGuessCallback,
+        string? runningState)
+    {
+        object result = request.Port.Id switch
+        {
+            "GuessNumber" => userGuessCallback(runningState ?? "Guess the number."),
+            _ => throw new NotSupportedException($"Request {request.Port.Id} is not supported")
+        };
+
+        return request.CreateResponse(result);
+    }
+
+    /// <summary>
+    /// This converts the incoming <see cref="NumberSignal"/> from the judge to a status text that can be displayed
+    /// to the user.
+    /// </summary>
+    /// <remarks>
+    /// This works correctly timing-wise because both the <see cref="StreamingAggregator{TInput, TOutput}"/> and the
+    /// <see cref="InputPort"/> are one edge from the <see cref="JudgeExecutor"/> (see the workflow definition in the
+    /// <see cref="RunAsync"/> method). That means they will get the <see cref="NumberSignal"/> at the same time (one
+    /// SuperStep after the Judge has generated it.)
+    /// </remarks>
+    /// <param name="signal"></param>
+    /// <param name="runningResult"></param>
+    /// <returns></returns>
+    private static string ComputeStreamingOutput(NumberSignal signal, string? runningResult)
+    {
+        return signal switch
+        {
+            NumberSignal.Matched => "You guessed correctly! You Win!",
+            NumberSignal.Above => "Your guess was too high. Try again.",
+            NumberSignal.Below => "Your guess was too low. Try again.",
+
+            _ => runningResult ?? string.Empty
+        };
+    }
+}

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs

@@ -12,7 +12,7 @@ using System.Threading.Tasks;
 using Microsoft.Extensions.AI;
 using Microsoft.Extensions.AI.Agents;
 
-namespace Microsoft.Agents.Workflows.UnitTests.Sample;
+namespace Microsoft.Agents.Workflows.Sample;
 
 internal static class Step6EntryPoint
 {

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs

@@ -7,7 +7,7 @@ using System.Threading.Tasks;
 using Microsoft.Extensions.AI;
 using Microsoft.Extensions.AI.Agents;
 
-namespace Microsoft.Agents.Workflows.UnitTests.Sample;
+namespace Microsoft.Agents.Workflows.Sample;
 
 internal static class Step7EntryPoint
 {

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

@@ -5,7 +5,6 @@ using System.IO;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Agents.Workflows.Sample;
-using Microsoft.Agents.Workflows.UnitTests.Sample;
 
 namespace Microsoft.Agents.Workflows.UnitTests;
 
@@ -72,7 +71,7 @@ public class SampleSmokeTest
     }
 
     [Fact]
-    public async Task Test_RunSample_Step5Async()
+    public async Task Test_RunSample_Step4Async()
     {
         using StringWriter writer = new();
 
@@ -81,6 +80,25 @@ public class SampleSmokeTest
             ("Your guess was too high. Try again.", 23),
             ("Your guess was too low. Try again.", 42));
 
+        string guessResult = await Step4EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext);
+        Assert.Equal("You guessed correctly! You Win!", guessResult);
+    }
+
+    [Fact]
+    public async Task Test_RunSample_Step5Async()
+    {
+        using StringWriter writer = new();
+
+        VerifyingPlaybackResponder<string, int> responder = new(
+            // Iteration 1
+            ("Guess the number.", 50),
+            ("Your guess was too high. Try again.", 23),
+
+            // Iteration 2
+            ("Your guess was too high. Try again.", 23),
+            ("Your guess was too low. Try again.", 42)
+         );
+
         string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext);
         Assert.Equal("You guessed correctly! You Win!", guessResult);
     }

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

@@ -75,7 +75,7 @@ public class StateSmokeTest
         Assert.Null(await manager.ReadStateAsync<object>(sharedScope1, "key2"));
 
         // Publish the write
-        await manager.PublishUpdatesAsync();
+        await manager.PublishUpdatesAsync(tracer: null);
 
         // Now all the executors should be able to see the new state
         Assert.NotNull(await manager.ReadStateAsync<object>(sharedScope1, Key));
@@ -105,7 +105,7 @@ public class StateSmokeTest
         // Try to publish the updates
         try
         {
-            await manager.PublishUpdatesAsync();
+            await manager.PublishUpdatesAsync(tracer: null);
             Assert.Fail("Expected InvalidOperationException due to conflicting writes.");
         }
         catch (InvalidOperationException)

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

@@ -0,0 +1,118 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Linq;
+using FluentAssertions;
+
+namespace Microsoft.Agents.Workflows.UnitTests;
+
+public class StreamingAggregatorsTests
+{
+    private static TResult? ApplyStreamingAggregator<TInput, TResult>(
+        StreamingAggregator<TInput, TResult> aggregator,
+        IEnumerable<TInput> inputs,
+        TResult? runningResult = default)
+    {
+        foreach (TInput input in inputs)
+        {
+            runningResult = aggregator(input, runningResult);
+        }
+
+        return runningResult!;
+    }
+
+    [Fact]
+    public void Test_StreamingAggregators_First()
+    {
+        IEnumerable<int> inputs = [1, 2, 3];
+        StreamingAggregator<int, int> aggregator = StreamingAggregators.First<int>();
+
+        int? runningResult = ApplyStreamingAggregator(aggregator, inputs);
+        runningResult.Should().Be(1);
+
+        // Ensure that subsequent inputs do not change the result
+        ApplyStreamingAggregator(aggregator, inputs.Skip(1), runningResult.Value)
+            .Should()
+            .Be(1, "subsequent inputs should not change the result of First aggregator");
+    }
+
+    [Fact]
+    public void Test_StreamingAggregators_First_WithConversion()
+    {
+        IEnumerable<int> inputs = [2, 4, 6];
+        StreamingAggregator<int, int> aggregator = StreamingAggregators.First<int, int>(input => input / 2);
+
+        int? runningResult = ApplyStreamingAggregator(aggregator, inputs);
+        runningResult.Should().Be(1);
+
+        // Ensure that subsequent inputs do not change the result
+        ApplyStreamingAggregator(aggregator, inputs.Skip(1), runningResult.Value)
+            .Should()
+            .Be(1, "subsequent inputs should not change the result of First aggregator with conversion");
+    }
+
+    [Fact]
+    public void Test_StreamingAggregators_Last()
+    {
+        IEnumerable<int> inputs = [1, 2, 3];
+        StreamingAggregator<int, int> aggregator = StreamingAggregators.Last<int>();
+
+        int? runningResult = ApplyStreamingAggregator(aggregator, inputs);
+        runningResult.Should().Be(3);
+
+        // Ensure that subsequent inputs do change the result
+        ApplyStreamingAggregator(aggregator, inputs.Take(2), runningResult.Value)
+            .Should()
+            .Be(2, "subsequent inputs should change the result of Last aggregator");
+    }
+
+    [Fact]
+    public void Test_StreamingAggregators_Last_WithConversion()
+    {
+        IEnumerable<int> inputs = [2, 4, 6];
+        StreamingAggregator<int, int> aggregator = StreamingAggregators.Last<int, int>(input => input / 2);
+
+        int? runningResult = ApplyStreamingAggregator(aggregator, inputs);
+        runningResult.Should().Be(3);
+
+        // Ensure that subsequent inputs do change the result
+        ApplyStreamingAggregator(aggregator, inputs.Take(2), runningResult.Value)
+            .Should()
+            .Be(2, "subsequent inputs should change the result of Last aggregator");
+    }
+
+    [Fact]
+    public void Test_StreamingAggregators_Union()
+    {
+        IEnumerable<int> inputs = [1, 2, 3];
+        StreamingAggregator<int, IEnumerable<int>> aggregator = StreamingAggregators.Union<int>();
+
+        IEnumerable<int>? runningResult = ApplyStreamingAggregator(aggregator, inputs);
+        runningResult.Should().BeEquivalentTo([1, 2, 3], "Union should accumulate all inputs in order");
+
+        // Ensure that subsequent inputs concatenate to the existing results
+        inputs = [4, 5];
+
+        ApplyStreamingAggregator(aggregator, inputs, runningResult)
+            .Should()
+            .BeEquivalentTo([1, 2, 3, 4, 5], "Union should accumulate all inputs in order including subsequent inputs");
+    }
+
+    [Fact]
+    public void Test_StreamingAggregators_Union_WithConversion()
+    {
+        IEnumerable<int> inputs = [2, 4, 6];
+        StreamingAggregator<int, IEnumerable<int>> aggregator = StreamingAggregators.Union<int, int>(input => input / 2);
+
+        IEnumerable<int>? runningResult = ApplyStreamingAggregator(aggregator, inputs);
+        runningResult.Should().BeEquivalentTo([1, 2, 3],
+            "Union with conversion should accumulate all converted inputs in order");
+
+        // Ensure that subsequent inputs concatenate to the existing results
+        inputs = [8, 10];
+        ApplyStreamingAggregator(aggregator, inputs, runningResult)
+            .Should()
+            .BeEquivalentTo([1, 2, 3, 4, 5],
+                "Union with conversion should accumulate all converted inputs in order including subsequent inputs");
+    }
+}