diff --git a/dotnet/src/Microsoft.Agents.Workflows/CheckpointInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/CheckpointInfo.cs new file mode 100644 index 0000000000..5a6f31b2ad --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/CheckpointInfo.cs @@ -0,0 +1,48 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; + +namespace Microsoft.Agents.Workflows; + +/// +/// Represents a checkpoint with a unique identifier and a timestamp indicating when it was created. +/// +public class CheckpointInfo : IEquatable +{ + /// + /// The unique identifier for the checkpoint. + /// + public string CheckpointId { get; } = Guid.NewGuid().ToString("N"); + + /// + /// The date and time when the object was created, in Coordinated Universal Time (UTC). + /// + public DateTimeOffset CreatedAt { get; } = DateTimeOffset.UtcNow; + + /// + public bool Equals(CheckpointInfo? other) + { + if (other == null) + { + return false; + } + + return this.CheckpointId == other.CheckpointId && + this.CreatedAt == other.CreatedAt; + } + + /// + public override bool Equals(object? obj) + { + return this.Equals(obj as CheckpointInfo); + } + + /// + public override int GetHashCode() + { + return HashCode.Combine(this.CheckpointId, this.CreatedAt); + } + + /// + public override string ToString() => $"CheckpointId: {this.CheckpointId}, CreatedAt: {this.CreatedAt:O}"; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/CheckpointManager.cs b/dotnet/src/Microsoft.Agents.Workflows/CheckpointManager.cs new file mode 100644 index 0000000000..c1caeab66a --- /dev/null +++ b/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; + +/// +/// An in-memory implementation of that stores checkpoints in a dictionary. +/// +public sealed class CheckpointManager : ICheckpointManager +{ + private readonly Dictionary _checkpoints = new(); + + ValueTask ICheckpointManager.CommitCheckpointAsync(Checkpoint checkpoint) + { + Throw.IfNull(checkpoint); + + this._checkpoints[checkpoint] = checkpoint; + return new(checkpoint); + } + + ValueTask 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); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointed.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointed.cs new file mode 100644 index 0000000000..ec9499b56c --- /dev/null +++ b/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; + +/// +/// Represents a workflow run that supports checkpointing. +/// +/// The type of the underlying workflow run handle +/// +/// +/// +/// +public class Checkpointed +{ + internal Checkpointed(TRun run, ICheckpointingRunner runner) + { + this.Run = Throw.IfNull(run); + this._runner = Throw.IfNull(runner); + } + + private readonly ICheckpointingRunner _runner; + + /// + /// Gets the workflow run associated with this instance. + /// + /// + /// + /// + /// + public TRun Run { get; } + + /// + public IReadOnlyList Checkpoints => this._runner.Checkpoints; + + /// + /// Gets the most recent checkpoint information. + /// + public CheckpointInfo? LastCheckpoint => this.Checkpoints.Count > 0 ? this.Checkpoints[this.Checkpoints.Count - 1] : null; + + /// + public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default) + => this._runner.RestoreCheckpointAsync(checkpointInfo, cancellation); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/Checkpoint.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/Checkpoint.cs new file mode 100644 index 0000000000..7390c38255 --- /dev/null +++ b/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 stateData, + Dictionary 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 State = new(); + public readonly Dictionary EdgeState = new(); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/DirectEdgeInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/DirectEdgeInfo.cs new file mode 100644 index 0000000000..1bcd6c3e7f --- /dev/null +++ b/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); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/EdgeInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/EdgeInfo.cs new file mode 100644 index 0000000000..c9aa4dd50a --- /dev/null +++ b/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; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ExecutorInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ExecutorInfo.cs new file mode 100644 index 0000000000..546cb824af --- /dev/null +++ b/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() where T : Executor + { + return this.ExecutorType.IsMatch() + && 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; + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ExportedState.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ExportedState.cs new file mode 100644 index 0000000000..fcdbf8bdc2 --- /dev/null +++ b/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); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/FanInEdgeInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/FanInEdgeInfo.cs new file mode 100644 index 0000000000..8ea914bb3b --- /dev/null +++ b/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); diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/FanOutEdgeInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/FanOutEdgeInfo.cs new file mode 100644 index 0000000000..7e4aa65502 --- /dev/null +++ b/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); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ICheckpointManager.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ICheckpointManager.cs new file mode 100644 index 0000000000..be93c391ea --- /dev/null +++ b/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; + +/// +/// A manager for storing and retrieving workflow execution checkpoints. +/// +internal interface ICheckpointManager +{ + /// + /// Commits the specified checkpoint and returns information that can be used to retrieve it later. + /// + /// The to be committed. + /// A representing the incoming checkpoint. + ValueTask CommitCheckpointAsync(Checkpoint checkpoint); + + /// + /// Retrieves the checkpoint associated with the specified checkpoint information. + /// + /// The information used to identify the checkpoint. + /// A representing the asynchronous operation. The result contains the associated with the specified . + /// Thrown if the checkpoint is not found. + ValueTask LookupCheckpointAsync(CheckpointInfo checkpointInfo); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ICheckpointingRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/ICheckpointingRunner.cs new file mode 100644 index 0000000000..fe29b9507f --- /dev/null +++ b/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 Checkpoints { get; } + + ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/InputPortInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/InputPortInfo.cs new file mode 100644 index 0000000000..abe2dddc91 --- /dev/null +++ b/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); diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/RepresentationExtensions.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/RepresentationExtensions.cs new file mode 100644 index 0000000000..96e3822d30 --- /dev/null +++ b/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(this Workflow workflow, TypeId? outputType, string? outputExecutorId) + { + Throw.IfNull(workflow); + + Dictionary executors = + workflow.Registrations.Values.ToDictionary( + keySelector: registration => registration.Id, + elementSelector: ToExecutorInfo); + + Dictionary> edges = workflow.Edges.Keys.ToDictionary( + keySelector: sourceId => sourceId, + elementSelector: sourceId => workflow.Edges[sourceId].Select(ToEdgeInfo).ToList()); + + HashSet 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(this Workflow workflow) + => workflow.ToWorkflowInfo(outputType: null, outputExecutorId: null); + + public static WorkflowInfo GetInfo(this Workflow workflow) + => workflow.ToWorkflowInfo(outputType: new TypeId(typeof(TResult)), outputExecutorId: workflow.OutputCollectorId); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/TypeId.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/TypeId.cs new file mode 100644 index 0000000000..077484442f --- /dev/null +++ b/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() => this.IsMatch(typeof(T)); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/WorkflowInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/Checkpointing/WorkflowInfo.cs new file mode 100644 index 0000000000..e879b3f71a --- /dev/null +++ b/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 executors, + Dictionary> edges, + HashSet 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 Executors { get; } + public Dictionary> Edges { get; } + public HashSet 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(Workflow workflow) => this.IsMatch(workflow as Workflow); + + public bool IsMatch(Workflow workflow) + => this.IsMatch(workflow as Workflow) + && this.OutputType != null && this.OutputType.IsMatch(typeof(TResult)) + && this.OutputCollectorId != null && this.OutputCollectorId == workflow.OutputCollectorId; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/DirectEdgeData.cs b/dotnet/src/Microsoft.Agents.Workflows/DirectEdgeData.cs index 19cdd296a0..165a43390e 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/DirectEdgeData.cs +++ b/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; namespace Microsoft.Agents.Workflows; @@ -11,7 +12,7 @@ namespace Microsoft.Agents.Workflows; /// The id of the source executor node. /// The id of the target executor node. /// A predicate determining whether the edge is active for a given message. -public sealed class DirectEdgeData(string sourceId, string sinkId, PredicateT? condition = null) +public sealed class DirectEdgeData(string sourceId, string sinkId, PredicateT? condition = null) : EdgeData { /// /// The Id of the source 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. /// public PredicateT? Condition => condition; + + /// + internal override EdgeConnection Connection { get; } = new([sourceId], [sinkId]); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Edge.cs b/dotnet/src/Microsoft.Agents.Workflows/Edge.cs index 27d08c0252..46452dc741 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Edge.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Edge.cs @@ -45,7 +45,7 @@ public sealed class Edge /// /// /// - public object Data { get; init; } + public EdgeData Data { get; init; } internal Edge(DirectEdgeData data) { diff --git a/dotnet/src/Microsoft.Agents.Workflows/EdgeData.cs b/dotnet/src/Microsoft.Agents.Workflows/EdgeData.cs new file mode 100644 index 0000000000..0ff69077cf --- /dev/null +++ b/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; + +/// +/// A base class for edge data, providing access to the representation of the edge. +/// +public abstract class EdgeData +{ + /// + /// Gets the connection representation of the edge. + /// + internal abstract EdgeConnection Connection { get; } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs index 30936d116d..ed44fbb555 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/DirectEdgeRunner.cs +++ b/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 FindRouterAsync() + private async ValueTask 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> ChaseAsync(MessageEnvelope envelope) + public async ValueTask> 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)]; } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeConnection.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeConnection.cs new file mode 100644 index 0000000000..bbd4ccd4cf --- /dev/null +++ b/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; + +/// +/// 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. +/// +/// +/// Ordering is relevant because in at least one case, the order of sinks is significant for the execution of +/// the edge: . +/// +/// An ordered list of unique identifiers of the sources connected by this edge. +/// An ordered list of unique identifiers of the sinks connected by this edge. +public class EdgeConnection(List sourceIds, List sinkIds) : IEquatable +{ + /// + /// Creates a new instance with the specified source and sink IDs, ensuring that all + /// IDs are unique. + /// + /// A list of source IDs. Each ID must be unique within the list. + /// A list of sink IDs. Each ID must be unique within the list. + /// An instance containing the specified source and sink IDs. + /// Throw if or + /// is + /// Thrown if or + /// contains duplicate values. + public static EdgeConnection CreateChecked(List sourceIds, List sinkIds) + { + HashSet sourceSet = new(Throw.IfNull(sourceIds)); + HashSet 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); + } + + /// + 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); + } + + /// + public override bool Equals(object? obj) + { + return this.Equals(obj as EdgeConnection); + } + + /// + 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())) + ); + } + + /// + /// The unique identifiers of the sources connected by this edge. + /// + public List SourceIds { get; } = sourceIds; + + /// + /// The unique identifiers of the sinks connected by this edge. + /// + public List SinkIds { get; } = sinkIds; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs index 7e6f961182..a8f86848dc 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/EdgeMap.cs +++ b/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 _edgeRunners = new(); - private readonly Dictionary _fanInState = new(); + private readonly Dictionary _edgeRunners = new(); + private readonly Dictionary _fanInState = new(); private readonly Dictionary _portEdgeRunners; private readonly InputEdgeRunner _inputRunner; + private readonly IStepTracer? _stepTracer; public EdgeMap(IRunnerContext runContext, Dictionary> workflowEdges, IEnumerable 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> 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> InvokeInputAsync(MessageEnvelope envelope) { - return [await this._inputRunner.ChaseAsync(envelope).ConfigureAwait(false)]; + return [await this._inputRunner.ChaseAsync(envelope, this._stepTracer).ConfigureAwait(false)]; } public async ValueTask> 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> ExportStateAsync() + { + Dictionary 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>(exportedStates); + } + + internal ValueTask ImportStateAsync(Checkpoint checkpoint) + { + Dictionary 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; } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs index dfce8c7984..222620a847 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanInEdgeRunner.cs +++ b/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 ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state) + public async ValueTask 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); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs index 32a91b1f45..b7e4723c2e 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/FanOutEdgeRunner.cs +++ b/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> ChaseAsync(MessageEnvelope envelope) + public async ValueTask> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer) { object message = envelope.Message; List 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 filteredTargets = envelope.TargetId != null @@ -32,11 +32,12 @@ internal class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData edgeDa async Task 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); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs index 8ee732b02c..6311c8b914 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerContext.cs +++ b/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 EnsureExecutorAsync(string executorId); + ValueTask EnsureExecutorAsync(string executorId, IStepTracer? tracer); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/IStepTracer.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/IStepTracer.cs new file mode 100644 index 0000000000..af093aa019 --- /dev/null +++ b/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(); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs index 33d9b0e0f0..903c2864de 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/InputEdgeRunner.cs +++ b/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 FindExecutorAsync() + private async ValueTask 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 ChaseAsync(MessageEnvelope envelope) + public async ValueTask 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); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/RunnerStateData.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/RunnerStateData.cs new file mode 100644 index 0000000000..9677ce6e57 --- /dev/null +++ b/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> queuedMessages, List outstandingRequests) +{ + public Dictionary> QueuedMessages { get; } = queuedMessages; + public List OutstandingRequests { get; } = outstandingRequests; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/StateManager.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/StateManager.cs index c0b6efbe2b..ca311a23fc 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/StateManager.cs +++ b/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>> 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> ExportScope(StateScope scope) + { + foreach (KeyValuePair state in scope.ExportStates()) + { + yield return new(new ScopeKey(scope.ScopeId, state.Key), state.Value); + } + } + + internal async ValueTask> 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 importedState = checkpoint.State; + + foreach (ScopeKey scopeKey in importedState.Keys) + { + StateScope scope = this.GetOrCreateScope(scopeKey.ScopeId); + scope.ImportState(scopeKey.Key, importedState[scopeKey]); + } + + return default; } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/StateScope.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/StateScope.cs index b832a9bb4a..6bd8bc18db 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/StateScope.cs +++ b/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> ExportStates() + { + return this._stateData.Keys.Select(WrapStates); + + KeyValuePair 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; + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs index 8b07d6339d..2aa8438f52 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/StepContext.cs +++ b/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> ExportMessages() + { + return this.QueuedMessages.Keys.ToDictionary( + keySelector: identity => identity, + elementSelector: identity => this.QueuedMessages[identity] + .Select(v => new ExportedState(v)) + .ToList() + ); + } + + internal void ImportMessages(Dictionary> 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}"); + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/UpdateKey.cs b/dotnet/src/Microsoft.Agents.Workflows/Execution/UpdateKey.cs index 15496cedb9..2f1252ea2a 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/UpdateKey.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Execution/UpdateKey.cs @@ -5,6 +5,16 @@ using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.Workflows.Execution; +/// +/// Represents a unique key used to identify an update within a specific scope. +/// +/// An is composed of a and a key, similar +/// to . 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. +/// +/// internal class UpdateKey(ScopeId scopeId, string key) { public ScopeId ScopeId { get; } = Throw.IfNull(scopeId); diff --git a/dotnet/src/Microsoft.Agents.Workflows/Executor.cs b/dotnet/src/Microsoft.Agents.Workflows/Executor.cs index ece973d7d1..f741a3515d 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Executor.cs +++ b/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; } + /// + /// Invoked before a checkpoint is saved, allowing custom pre-save logic in derived classes. + /// + /// The workflow context. + /// A ValueTask representing the asynchronous operation. + /// + protected internal virtual ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => default; + + /// + /// Invoked after a checkpoint is loaded, allowing custom post-load logic in derived classes. + /// + /// The workflow context. + /// A ValueTask representing the asynchronous operation. + /// + protected internal virtual ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => default; + /// /// A set of s, representing the messages this executor can handle. /// diff --git a/dotnet/src/Microsoft.Agents.Workflows/ExecutorIsh.cs b/dotnet/src/Microsoft.Agents.Workflows/ExecutorIsh.cs index a117f1ae0b..c303286acb 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/ExecutorIsh.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/ExecutorIsh.cs @@ -103,10 +103,27 @@ public sealed class ExecutorIsh : }; /// - /// Gets an that can be used to obtain an instance + /// Gets the registration details for the current executor. + /// + /// The returned registration depends on the type of the executor. If the executor is unbound, an + /// is thrown. For other executor types, the registration includes the + /// appropriate ID, type, and provider based on the executor's configuration. + 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}") + }; + + /// + /// Gets an that can be used to obtain an instance /// corresponding to this . /// - public ExecutorProvider ExecutorProvider => this.ExecutorType switch + private Func ExecutorProvider => this.ExecutorType switch { Type.Unbound => throw new InvalidOperationException($"Executor with ID '{this.Id}' is unbound."), Type.Executor => () => this._executorValue!, diff --git a/dotnet/src/Microsoft.Agents.Workflows/ExecutorRegistration.cs b/dotnet/src/Microsoft.Agents.Workflows/ExecutorRegistration.cs new file mode 100644 index 0000000000..d45a126472 --- /dev/null +++ b/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; + +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()); +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/FanInEdgeData.cs b/dotnet/src/Microsoft.Agents.Workflows/FanInEdgeData.cs index c98dd1d7ea..33c6fdc241 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/FanInEdgeData.cs +++ b/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; /// /// An enumeration of ids of the source executor nodes. /// The id of the target executor node. -public sealed class FanInEdgeData(List sourceIds, string sinkId) +public sealed class FanInEdgeData(List sourceIds, string sinkId) : EdgeData { /// /// The ordered list of Ids of the source nodes. @@ -22,5 +22,6 @@ public sealed class FanInEdgeData(List sourceIds, string sinkId) /// public string SinkId => sinkId; - internal Guid UniqueKey { get; } = Guid.NewGuid(); + /// + internal override EdgeConnection Connection { get; } = new(sourceIds, [sinkId]); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/FanOutEdgeData.cs b/dotnet/src/Microsoft.Agents.Workflows/FanOutEdgeData.cs index cd11bfb027..1fe83edae2 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/FanOutEdgeData.cs +++ b/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>; +using Microsoft.Agents.Workflows.Execution; + +using AssignerF = System.Func>; namespace Microsoft.Agents.Workflows; @@ -11,11 +13,11 @@ namespace Microsoft.Agents.Workflows; /// /// The id of the source executor node. /// A list of ids of the target executor nodes. -/// A function that maps an incoming message to a subset of the target executor nodes. +/// A function that maps an incoming message to a subset of the target executor nodes. public sealed class FanOutEdgeData( string sourceId, List sinkIds, - PartitionerT? partitioner = null) + AssignerF? assigner = null) : EdgeData { /// /// The Id of the source 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 , all destination nodes are selected. /// - public PartitionerT? PartitionAssigner => partitioner; + public AssignerF? EdgeAssigner => assigner; + + /// + internal override EdgeConnection Connection { get; } = new([sourceId], sinkIds); } diff --git a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcStepTracer.cs b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcStepTracer.cs new file mode 100644 index 0000000000..e126624f8e --- /dev/null +++ b/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 Instantiated { get; } = []; + public HashSet 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; + + /// + /// Reset the tracer to the specified step number. + /// + /// The Step Number of the last SuperStep. Note that Step Numbers are 0-indexed. + 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 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(); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs index 29217a099c..d1291ff982 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs +++ b/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. /// The type of input accepted by the workflow. Must be non-nullable. -internal class InProcessRunner : ISuperStepRunner where TInput : notnull +internal class InProcessRunner : ISuperStepRunner, ICheckpointingRunner where TInput : notnull { - public InProcessRunner(Workflow workflow) + public InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager) { this.Workflow = Throw.IfNull(workflow); this.RunContext = new InProcessRunnerContext(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 IsValidInputAsync(TMessage message) @@ -42,7 +45,7 @@ internal class InProcessRunner : 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 : ISuperStepRunner where TInput : notnull return this.RunContext.AddExternalMessageAsync(response); } + private InProcStepTracer StepTracer { get; } = new(); private Dictionary PendingCalls { get; } = new(); private Workflow Workflow { get; init; } private InProcessRunnerContext RunContext { get; init; } + private ICheckpointManager? CheckpointManager { get; } private EdgeMap EdgeMap { get; init; } event EventHandler? ISuperStepRunner.WorkflowEvent @@ -107,6 +112,19 @@ internal class InProcessRunner : ISuperStepRunner where TInput : notnull return this.EdgeMap.InvokeResponseAsync(response); } + public async ValueTask 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 StreamAsync(TInput input, CancellationToken cancellation = default) { await this.RunContext.AddExternalMessageAsync(input).ConfigureAwait(false); @@ -114,6 +132,14 @@ internal class InProcessRunner : ISuperStepRunner where TInput : notnull return new StreamingRun(this); } + internal async ValueTask 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 RunAsync(TInput input, CancellationToken cancellation = default) { StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false); @@ -125,6 +151,9 @@ internal class InProcessRunner : ISuperStepRunner where TInput : notnull bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests; bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions; + public IReadOnlyList Checkpoints => this._checkpoints; + private CheckpointInfo? LastCheckpoint => this.Checkpoints[this.Checkpoints.Count - 1]; + async ValueTask ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellation) { cancellation.ThrowIfCancellationRequested(); @@ -142,6 +171,8 @@ internal class InProcessRunner : 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>> edgeTasks = new(); foreach (ExecutorIdentity sender in currentStep.QueuedMessages.Keys) @@ -165,37 +196,121 @@ internal class InProcessRunner : ISuperStepRunner where TInput : notnull // that we would need to avoid firing the tasks when we call InvokeEdgeAsync, or RouteExternalMessageAsync. IEnumerable results = (await Task.WhenAll(edgeTasks).ConfigureAwait(false)).SelectMany(r => r); - // 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 _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 edgeData = await this.EdgeMap.ExportStateAsync().ConfigureAwait(false); + + await prepareTask.ConfigureAwait(false); + await this.RunContext.StateManager.PublishUpdatesAsync(this.StepTracer).ConfigureAwait(false); + + Dictionary 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(this.Workflow); } } -internal class InProcessRunner : IRunnerWithOutput where TInput : notnull +internal class InProcessRunner : IRunnerWithOutput, ICheckpointingRunner where TInput : notnull { private readonly Workflow _workflow; - private readonly ISuperStepRunner _innerRunner; + private readonly InProcessRunner _innerRunner; - public InProcessRunner(Workflow workflow) + public InProcessRunner(Workflow workflow, CheckpointManager? checkpointManager) { this._workflow = Throw.IfNull(workflow); - this._innerRunner = new InProcessRunner(workflow); + + InProcessRunner runner = new(workflow, checkpointManager); + this._innerRunner = runner; + } + + internal async ValueTask> ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default) + { + await this._innerRunner.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false); + + return new StreamingRun(this); } public async ValueTask> StreamAsync(TInput input, CancellationToken cancellation = default) { - await this._innerRunner.EnqueueMessageAsync(input).ConfigureAwait(false); + await ((ISuperStepRunner)this._innerRunner).EnqueueMessageAsync(input).ConfigureAwait(false); return new StreamingRun(this); } + public async ValueTask> 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> RunAsync(TInput input, CancellationToken cancellation = default) { StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false); @@ -204,8 +319,15 @@ internal class InProcessRunner : IRunnerWithOutput whe return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false); } + public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default) + => this._innerRunner.RestoreCheckpointAsync(checkpointInfo, cancellation); + + internal ValueTask CheckpointAsync() => this._innerRunner.CheckpointAsync(); + /// public TResult? RunningOutput => this._workflow.RunningOutput; ISuperStepRunner IRunnerWithOutput.StepRunner => this._innerRunner; + + public IReadOnlyList Checkpoints => this._innerRunner.Checkpoints; } diff --git a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs index c2206fdea6..f122f3058d 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs +++ b/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 : IRunnerContext { private StepContext _nextStep = new(); - private readonly Dictionary> _executorProviders; + private readonly Dictionary _executorRegistrations; private readonly Dictionary _executors = new(); private readonly Dictionary _externalRequests = new(); public InProcessRunnerContext(Workflow workflow, ILogger? logger = null) { - this._executorProviders = Throw.IfNull(workflow).ExecutorProviders; + this._executorRegistrations = Throw.IfNull(workflow).Registrations; } - public async ValueTask EnsureExecutorAsync(string executorId) + public async ValueTask 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 : IRunnerContext public ValueTask ReadStateAsync(string key, string? scopeName = null) => RunnerContext.StateManager.ReadStateAsync(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 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> 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; + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/InProcessExecution.cs b/dotnet/src/Microsoft.Agents.Workflows/InProcessExecution.cs index cda0cc5ad2..e435927f9e 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/InProcessExecution.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/InProcessExecution.cs @@ -24,12 +24,67 @@ public static class InProcessExecution /// A that can be used to cancel the streaming operation. /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - public static ValueTask StreamAsync(Workflow workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull + public static ValueTask StreamAsync( + Workflow workflow, + TInput input, + CancellationToken cancellation = default) where TInput : notnull { - InProcessRunner runner = new(workflow); + InProcessRunner runner = new(workflow, checkpointManager: null); return runner.StreamAsync(input, cancellation); } + /// + /// Initiates an asynchronous streaming execution using the specified input, with checkpointing. + /// + /// The returned provides methods to observe and control + /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or + /// cancelled. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The workflow to be executed. Must not be null. + /// The input message to be processed as part of the streaming run. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + public static async ValueTask> StreamAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + StreamingRun result = await runner.StreamAsync(input, cancellation).ConfigureAwait(false); + + await runner.CheckpointAsync(cancellation).ConfigureAwait(false); + + return new(result, runner); + } + + /// + /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. + /// + /// The returned can be used to retrieve results + /// as they become available. If the operation is cancelled via the token, the + /// streaming execution will be terminated. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The workflow to be executed. Must not be null. + /// The corresponding to the checkpoint from which to resume. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that provides access to the results of the streaming + /// run. + public static async ValueTask> ResumeStreamAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + StreamingRun result = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false); + + return new(result, runner); + } + /// /// Initiates an asynchronous streaming execution for the specified input. /// @@ -43,12 +98,69 @@ public static class InProcessExecution /// A that can be used to cancel the streaming operation. /// A that provides access to the results of the streaming /// run. - public static ValueTask> StreamAsync(Workflow workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull + public static ValueTask> StreamAsync( + Workflow workflow, + TInput input, + CancellationToken cancellation = default) where TInput : notnull { - InProcessRunner runner = new(workflow); + InProcessRunner runner = new(workflow, checkpointManager: null); return runner.StreamAsync(input, cancellation); } + /// + /// Initiates an asynchronous streaming execution for the specified input, with checkpointing. + /// + /// The returned can be used to retrieve results + /// as they become available. If the operation is cancelled via the token, the + /// streaming execution will be terminated. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The type of output produced by the workflow. + /// The workflow to be executed. Must not be null. + /// The input value to be processed by the streaming run. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that provides access to the results of the streaming + /// run. + public static async ValueTask>> StreamAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + StreamingRun result = await runner.StreamAsync(input, cancellation).ConfigureAwait(false); + + await runner.CheckpointAsync().ConfigureAwait(false); + + return new(result, runner); + } + + /// + /// Resumes an asynchronous streaming execution of the workflow from a checkpoint. + /// + /// The returned can be used to retrieve results + /// as they become available. If the operation is cancelled via the token, the + /// streaming execution will be terminated. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The type of output produced by the workflow. + /// The workflow to be executed. Must not be null. + /// The corresponding to the checkpoint from which to resume. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that provides access to the results of the streaming + /// run. + public static async ValueTask>> ResumeStreamAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + StreamingRun result = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false); + + return new(result, runner); + } + /// /// Initiates a non-streaming execution of the workflow with the specified input. /// @@ -60,12 +172,65 @@ public static class InProcessExecution /// A that can be used to cancel the streaming operation. /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - public static ValueTask RunAsync(Workflow workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull + public static ValueTask RunAsync( + Workflow workflow, + TInput input, + CancellationToken cancellation = default) where TInput : notnull { - InProcessRunner runner = new(workflow); + InProcessRunner runner = new(workflow, checkpointManager: null); return runner.RunAsync(input, cancellation); } + /// + /// Initiates a non-streaming execution of the workflow with the specified input, with checkpointing. + /// + /// The workflow will run until its first halt, and the returned will capture + /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The workflow to be executed. Must not be null. + /// The input message to be processed as part of the run. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + public static async ValueTask> RunAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + Run result = await runner.RunAsync(input, cancellation).ConfigureAwait(false); + + await runner.CheckpointAsync(cancellation).ConfigureAwait(false); + + return new(result, runner); + } + + /// + /// Resumes a non-streaming execution of the workflow from a checkpoint. + /// + /// The workflow will run until its first halt, and the returned will capture + /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The workflow to be executed. Must not be null. + /// The corresponding to the checkpoint from which to resume. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + public static async ValueTask> ResumeAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + Run result = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false); + + return new(result, runner); + } + /// /// Initiates a non-streaming execution of the workflow with the specified input. /// @@ -78,9 +243,64 @@ public static class InProcessExecution /// A that can be used to cancel the streaming operation. /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - public static ValueTask> RunAsync(Workflow workflow, TInput input, CancellationToken cancellation = default) where TInput : notnull + public static ValueTask> RunAsync( + Workflow workflow, + TInput input, + CancellationToken cancellation = default) where TInput : notnull { - InProcessRunner runner = new(workflow); + InProcessRunner runner = new(workflow, checkpointManager: null); return runner.RunAsync(input, cancellation); } + + /// + /// Initiates a non-streaming execution of the workflow with the specified input, with checkpointing. + /// + /// The workflow will run until its first halt, and the returned will capture + /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The type of output produced by the workflow. + /// The workflow to be executed. Must not be null. + /// The input message to be processed as part of the run. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + public static async ValueTask>> RunAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + Run result = await runner.RunAsync(input, cancellation).ConfigureAwait(false); + + await runner.CheckpointAsync().ConfigureAwait(false); + + return new(result, runner); + } + + /// + /// Resumes a non-streaming execution of the workflow from a checkpoint. + /// + /// The workflow will run until its first halt, and the returned will capture + /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. + /// The type of input accepted by the workflow. Must be non-nullable. + /// The type of output produced by the workflow. + /// The workflow to be executed. Must not be null. + /// The corresponding to the checkpoint from which to resume. + /// The to use with this run. + /// A that can be used to cancel the streaming operation. + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + public static async ValueTask>> ResumeAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + CancellationToken cancellation = default) where TInput : notnull + { + InProcessRunner runner = new(workflow, checkpointManager); + Run result = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false); + + return new(result, runner); + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Execution/ScopeId.cs b/dotnet/src/Microsoft.Agents.Workflows/ScopeId.cs similarity index 69% rename from dotnet/src/Microsoft.Agents.Workflows/Execution/ScopeId.cs rename to dotnet/src/Microsoft.Agents.Workflows/ScopeId.cs index 3d96b2bfc9..b638bb59f6 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Execution/ScopeId.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/ScopeId.cs @@ -2,25 +2,35 @@ using Microsoft.Shared.Diagnostics; -namespace Microsoft.Agents.Workflows.Execution; +namespace Microsoft.Agents.Workflows; /// /// 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. /// -/// -/// -internal class ScopeId(string executorId, string? scopeName = null) +/// The unique identifier for the executor associated with this ScopeId. +/// The name of the scope, if any. If , this ScopeId +/// corresponds to the Executor's private scope. +public class ScopeId(string executorId, string? scopeName = null) { + /// + /// Gets the unique identifier of the executor. + /// public string ExecutorId { get; } = Throw.IfNullOrEmpty(executorId); + + /// + /// Gets the name of the current scope, if any. + /// public string? ScopeName { get; } = scopeName; + /// public override string ToString() { return $"{this.ExecutorId}/{this.ScopeName ?? "default"}"; } + /// 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; } + /// public override int GetHashCode() { if (this.ScopeName is null) diff --git a/dotnet/src/Microsoft.Agents.Workflows/ScopeKey.cs b/dotnet/src/Microsoft.Agents.Workflows/ScopeKey.cs new file mode 100644 index 0000000000..ce5be0a899 --- /dev/null +++ b/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; + +/// +/// Represents a unique key within a specific scope, combining a scope identifier and a key string. +/// +/// The associated with this key. +/// The unique key within the specified scope. +public class ScopeKey(ScopeId scopeId, string key) +{ + /// + /// The identifier for the scope associated with this key. + /// + public ScopeId ScopeId { get; } = Throw.IfNull(scopeId); + + /// + /// The unique key within the specified scope. + /// + public string Key { get; } = Throw.IfNullOrEmpty(key); + + /// + /// Initializes a new instance of the class. + /// + /// The unique identifier for the executor. + /// The name of the scope, if any. + /// The unique key within the specified scope. + public ScopeKey(string executorId, string? scopeName, string key) + : this(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName), key) + { } + + /// + public override string ToString() + { + return $"{this.ScopeId}/{this.Key}"; + } + + /// + 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; + } + + /// + public override int GetHashCode() + { + return HashCode.Combine(this.ScopeId, this.Key); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs b/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs index 16575b1f95..006287c023 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs +++ b/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(ThreadStateKey).ConfigureAwait(false); + if (threadValue.HasValue) + { + this._thread = await this._agent.DeserializeThreadAsync(threadValue.Value, cancellationToken: cancellation) + .ConfigureAwait(false); + } + + JsonElement? messagesValue = await context.ReadStateAsync(PendingMessagesStateKey).ConfigureAwait(false); + if (messagesValue.HasValue) + { + List 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 agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread()); + IAsyncEnumerable agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread(context)); List updates = new(); await foreach (AgentRunResponseUpdate update in agentStream.ConfigureAwait(false)) diff --git a/dotnet/src/Microsoft.Agents.Workflows/Specialized/IOutputSink.cs b/dotnet/src/Microsoft.Agents.Workflows/Specialized/IOutputSink.cs index 3039b24a4f..c066914122 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Specialized/IOutputSink.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Specialized/IOutputSink.cs @@ -2,7 +2,7 @@ namespace Microsoft.Agents.Workflows.Specialized; -internal interface IOutputSink +internal interface IOutputSink : IIdentified { TResult? Result { get; } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/Specialized/WorkflowJsonUtilities.cs b/dotnet/src/Microsoft.Agents.Workflows/Specialized/WorkflowJsonUtilities.cs new file mode 100644 index 0000000000..9ea4d5c864 --- /dev/null +++ b/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))] + internal sealed partial class WorkflowJsonContext : JsonSerializerContext; + + public static JsonElement SerializeToJson(this List messages) + { + return JsonSerializer.SerializeToElement(messages, Default.ListChatMessage); + } + + public static JsonElement SerializeToJson(this IEnumerable messages) + => messages.ToList().SerializeToJson(); + + public static List DeserializeMessageList(this JsonElement element) + { + return element.Deserialize>(Default.ListChatMessage) ?? []; + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/StreamingAggregators.cs b/dotnet/src/Microsoft.Agents.Workflows/StreamingAggregators.cs index 4a767bb282..e995f273d9 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/StreamingAggregators.cs +++ b/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 { /// /// 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. /// /// 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 /// The type of the result produced by the conversion function. /// A function that converts an input value of type to a result of type . This function is applied to the first input received. - /// The value to return if no input is provided. - /// A that yields the converted result of the first input, or the - /// specified default value if no input is received. - public static StreamingAggregator First(Func conversion, TResult? defaultValue = default) + /// A that yields the converted result of the first input. + public static StreamingAggregator First(Func 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 } /// - /// 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. /// /// The type of the input elements to aggregate. - /// The value to return if the input sequence contains no elements. - /// A that yields the first input element, or if the sequence is empty. - public static StreamingAggregator First(TInput? defaultValue = default) - => First(input => input, defaultValue); + /// A that yields the first input element. + public static StreamingAggregator First() => First(input => input); /// /// 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 /// The type of the input elements to be aggregated. /// The type of the result produced by the conversion function. /// A function that converts each input value to a result. Cannot be null. - /// The initial result value to use before any input is processed. - /// A streaming aggregator that yields the converted value of the last input received, or the specified default - /// value if no input has been processed. - public static StreamingAggregator Last(Func conversion, TResult? defaultValue = default) + /// A streaming aggregator that yields the converted value of the last input received. + public static StreamingAggregator Last(Func conversion) { - TResult? local = defaultValue; + TResult? local = default; return Aggregate; @@ -89,15 +83,11 @@ public static class StreamingAggregators } /// - /// 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. /// /// The type of elements in the input sequence. - /// The value to return if the input sequence contains no elements. - /// A that yields the last element of the sequence, or if the sequence is empty. - public static StreamingAggregator Last(TInput? defaultValue = default) - => Last(input => input, defaultValue); + /// A that yields the last element of the sequence. + public static StreamingAggregator Last() => Last(input => input); /// /// 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. public static StreamingAggregator> Union(Func conversion) { - List results = new(); - return Aggregate; IEnumerable Aggregate(TInput input, IEnumerable? runningResult) { - results.Add(conversion(input)); - return results; + return runningResult != null ? runningResult.Append(conversion(input)) : [conversion(input)]; } } @@ -130,5 +117,12 @@ public static class StreamingAggregators /// A StreamingAggregator that, when applied to multiple input sequences, returns an IEnumerable containing the /// union of all elements from those sequences. public static StreamingAggregator> Union() - => Union(input => input); + { + return Aggregate; + + IEnumerable Aggregate(TInput input, IEnumerable? runningResult) + { + return runningResult != null ? runningResult.Append(input) : new[] { input }; + } + } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/SuperStepCompletedEvent.cs b/dotnet/src/Microsoft.Agents.Workflows/SuperStepCompletedEvent.cs new file mode 100644 index 0000000000..9c1f7a43b0 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/SuperStepCompletedEvent.cs @@ -0,0 +1,16 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.Workflows; + +/// +/// Event triggered when a SuperStep completed. +/// +/// The zero-based index of the SuperStep associated with this event. +/// Debug information about the state of the system on SuperStep completion. +public sealed class SuperStepCompletedEvent(int stepNumber, SuperStepCompletionInfo? completionInfo = null) : SuperStepEvent(stepNumber, data: completionInfo) +{ + /// + /// Gets the debug information about the state of the system on SuperStep completion. + /// + public SuperStepCompletionInfo? CompletionInfo => completionInfo; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/SuperStepCompletionInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/SuperStepCompletionInfo.cs new file mode 100644 index 0000000000..53b235809b --- /dev/null +++ b/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; + +/// +/// Debug information about the SuperStep that finished running. +/// +public sealed class SuperStepCompletionInfo(HashSet activatedExecutors, HashSet? instantiatedExecutors = null) +{ + /// + /// The unique identifiers of instances that processed messages during this SuperStep + /// + public HashSet ActivatedExecutors { get; } = Throw.IfNull(activatedExecutors); + + /// + /// The unique identifiers of instances newly created during this SuperStep + /// + public HashSet InstantiatedExecutors { get; } = instantiatedExecutors ?? []; + + /// + /// 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. + /// + public bool StateUpdated { get; init; } + + /// + /// A flag indicating whether there are messages pending delivery after this SuperStep. + /// + public bool HasPendingMessages { get; init; } + + /// + /// A flag indicating whether there are requests pending delivery after this SuperStep. + /// + public bool HasPendingRequests { get; init; } + + /// + /// Gets the corresponding to the checkpoint created at the end of this SuperStep. + /// if checkpointing was not enabled when the run was started. + /// + public CheckpointInfo? Checkpoint { get; init; } = null; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/SuperStepEvent.cs b/dotnet/src/Microsoft.Agents.Workflows/SuperStepEvent.cs new file mode 100644 index 0000000000..f0b64f2169 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/SuperStepEvent.cs @@ -0,0 +1,25 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.Workflows; + +/// +/// Base class for SuperStep-scoped events, for example, +/// +public class SuperStepEvent(int stepNumber, object? data = null) : WorkflowEvent(data) +{ + /// + /// The zero-based index of the SuperStep associated with this event. + /// + public int StepNumber => stepNumber; + + /// + 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})"; + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/SuperStepStartInfo.cs b/dotnet/src/Microsoft.Agents.Workflows/SuperStepStartInfo.cs new file mode 100644 index 0000000000..48a1de1771 --- /dev/null +++ b/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; + +/// +/// Debug information about the SuperStep starting to run. +/// +public sealed class SuperStepStartInfo(HashSet? sendingExecutors = null) +{ + /// + /// The unique identifiers of instances that sent messages during the previous SuperStep. + /// + public HashSet SendingExecutors { get; } = sendingExecutors ?? []; + + /// + /// Gets a value indicating whether there are any external messages queued during the previous SuperStep. + /// + public bool HasExternalMessages { get; init; } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/SuperStepStartedEvent.cs b/dotnet/src/Microsoft.Agents.Workflows/SuperStepStartedEvent.cs new file mode 100644 index 0000000000..9a5d1fb18e --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/SuperStepStartedEvent.cs @@ -0,0 +1,16 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.Workflows; + +/// +/// Event triggered when a SuperStep started. +/// +/// The zero-based index of the SuperStep associated with this event. +/// Debug information about the state of the system on SuperStep start. +public sealed class SuperStepStartedEvent(int stepNumber, SuperStepStartInfo? startInfo = null) : SuperStepEvent(stepNumber, data: startInfo) +{ + /// + /// Gets the debug information about the state of the system on SuperStep start. + /// + public SuperStepStartInfo? StartInfo => startInfo; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs index c216605e10..2b1549c5bd 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Workflow.cs @@ -15,7 +15,7 @@ public class Workflow /// /// A dictionary of executor providers, keyed by executor ID. /// - public Dictionary> ExecutorProviders { get; internal init; } = new(); + internal Dictionary Registrations { get; init; } = new(); /// /// 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. /// /// - /// Each port has a corresponding entry in the dictionary. + /// Each port has a corresponding entry in the dictionary. /// public Dictionary Ports { get; internal init; } = new(); @@ -73,7 +73,7 @@ public class Workflow : Workflow return new Workflow(this.StartExecutorId, outputSource) { - ExecutorProviders = this.ExecutorProviders, + Registrations = this.Registrations, Edges = this.Edges, Ports = this.Ports }; @@ -96,6 +96,11 @@ public class Workflow : Workflow this._output = Throw.IfNull(outputSource); } + /// + /// Gets the unique identifier of the output collector. + /// + public string OutputCollectorId => this._output.Id; + /// /// The running (partial) output of the workflow, if any. /// diff --git a/dotnet/src/Microsoft.Agents.Workflows/WorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.Workflows/WorkflowBuilder.cs index eda20940b9..4e61c0e3a8 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/WorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/WorkflowBuilder.cs @@ -7,14 +7,6 @@ using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.Workflows; -/// -/// A factory method that produces an executor instance. -/// -/// The executor type. -/// A new instance. -public delegate TExecutor ExecutorProvider() - where TExecutor : Executor; - /// /// 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> _executors = new(); + private readonly Dictionary _executors = new(); private readonly Dictionary> _edges = new(); private readonly HashSet _unboundExecutors = new(); private readonly HashSet _conditionlessEdges = new(); @@ -49,20 +41,17 @@ public class WorkflowBuilder private ExecutorIsh Track(ExecutorIsh executorish) { - ExecutorProvider 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 provider) - { - this._executors[id] = provider; - } - /// /// Binds the specified executor to the workflow, allowing it to participate in workflow execution. /// @@ -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? 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(this._startExecutorId) // Why does it not see the default ctor? { - ExecutorProviders = this._executors, + Registrations = this._executors, Edges = this._edges, Ports = this._inputPorts }; diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/RepresentationTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/RepresentationTests.cs new file mode 100644 index 0000000000..1809d0d908 --- /dev/null +++ b/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 RunAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public override IAsyncEnumerable RunStreamingAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + } + + private static InputPort TestInputPort => + InputPort.Create("ExternalFunction"); + + private static List ListAggregator(List? 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> outputCollector = new(StreamingAggregators.Union()); + 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 Condition() => Condition(); + private static Func Condition() => _ => true; + + private static Func> EdgeAssigner() => EdgeAssigner(); + private static Func> EdgeAssigner() => (_, _) => []; + + [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(Workflow workflow, Workflow? comparator = null, bool expect = true) + { + comparator ??= workflow; + + WorkflowInfo info = workflow.ToWorkflowInfo(); + info.IsMatch(comparator).Should().Be(expect); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs index 9030dce3e0..229e6c3b39 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs +++ b/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 WorkflowInstance + { + get + { + UppercaseExecutor uppercase = new(); + ReverseTextExecutor reverse = new(); + + WorkflowBuilder builder = new(uppercase); + builder.AddEdge(uppercase, reverse); + + return builder.Build(); + } + } + public static async ValueTask RunAsync(TextWriter writer) { - UppercaseExecutor uppercase = new(); - ReverseTextExecutor reverse = new(); - - WorkflowBuilder builder = new(uppercase); - builder.AddEdge(uppercase, reverse); - - Workflow workflow = builder.Build(); - 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)) { diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs index 1475c078d6..cf78998bf5 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs +++ b/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 workflow = builder.Build(); - - 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); diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs index 408a8954ce..25a726c893 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs +++ b/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 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(); + } + } + public static async ValueTask 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 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(); - - 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) diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs index 45bcc208ee..d886f2a6c7 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs @@ -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 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(); + } + } + public static async ValueTask RunAsync(TextWriter writer) { - GuessNumberExecutor guessNumber = new(1, 100); - JudgeExecutor judge = new(42); // Let's say the target number is 42 - - Workflow workflow = new WorkflowBuilder(guessNumber) - .AddEdge(guessNumber, judge) - .AddEdge(judge, guessNumber) - .Build(); - - 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, 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, IMessag public async ValueTask 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, 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("TryCount").ConfigureAwait(false); + } } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_ExternalRequest.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs similarity index 87% rename from dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_ExternalRequest.cs rename to dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs index a9593a38e9..1f078ecd29 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_ExternalRequest.cs +++ b/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 RunAsync(TextWriter writer, Func userGuessCallback) + public static Workflow CreateWorkflowInstance(out JudgeExecutor judge) { InputPort guessNumber = InputPort.Create("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 workflow = new WorkflowBuilder(guessNumber) + return new WorkflowBuilder(guessNumber) .AddEdge(guessNumber, judge) .AddEdge(judge, guessNumber, (message) => message is NumberSignal signal && signal != NumberSignal.Matched) .BuildWithOutput(judge, ComputeStreamingOutput, (NumberSignal s, string? _) => s == NumberSignal.Matched); + } + public static Workflow WorkflowInstance + { + get + { + return CreateWorkflowInstance(out _); + } + } + + public static async ValueTask RunAsync(TextWriter writer, Func userGuessCallback) + { + Workflow workflow = WorkflowInstance; StreamingRun handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false)) diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs new file mode 100644 index 0000000000..f0755c064a --- /dev/null +++ b/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 RunAsync(TextWriter writer, Func userGuessCallback) + { + Workflow workflow = Step4EntryPoint.CreateWorkflowInstance(out JudgeExecutor judge); + Checkpointed> checkpointed = + await InProcessExecution.StreamAsync(workflow, NumberSignal.Init, CheckpointManager) + .ConfigureAwait(false); + + List checkpoints = new(); + CancellationTokenSource cancellationSource = new(); + + StreamingRun 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 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 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); + } + + /// + /// This converts the incoming from the judge to a status text that can be displayed + /// to the user. + /// + /// + /// This works correctly timing-wise because both the and the + /// are one edge from the (see the workflow definition in the + /// method). That means they will get the at the same time (one + /// SuperStep after the Judge has generated it.) + /// + /// + /// + /// + 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 + }; + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs index d631570902..0bcf319562 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs +++ b/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 { diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs index 9bff976f35..a46f1b6660 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs +++ b/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 { diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs index 0ac7635533..9e3277dfaf 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs +++ b/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 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); } diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/StateSmokeTest.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/StateSmokeTest.cs index 1aeff7da3c..32054aae5d 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/StateSmokeTest.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/StateSmokeTest.cs @@ -75,7 +75,7 @@ public class StateSmokeTest Assert.Null(await manager.ReadStateAsync(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(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) diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/StreamingAggregatorsTests.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/StreamingAggregatorsTests.cs new file mode 100644 index 0000000000..337c4681f0 --- /dev/null +++ b/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( + StreamingAggregator aggregator, + IEnumerable inputs, + TResult? runningResult = default) + { + foreach (TInput input in inputs) + { + runningResult = aggregator(input, runningResult); + } + + return runningResult!; + } + + [Fact] + public void Test_StreamingAggregators_First() + { + IEnumerable inputs = [1, 2, 3]; + StreamingAggregator aggregator = StreamingAggregators.First(); + + 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 inputs = [2, 4, 6]; + StreamingAggregator aggregator = StreamingAggregators.First(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 inputs = [1, 2, 3]; + StreamingAggregator aggregator = StreamingAggregators.Last(); + + 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 inputs = [2, 4, 6]; + StreamingAggregator aggregator = StreamingAggregators.Last(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 inputs = [1, 2, 3]; + StreamingAggregator> aggregator = StreamingAggregators.Union(); + + IEnumerable? 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 inputs = [2, 4, 6]; + StreamingAggregator> aggregator = StreamingAggregators.Union(input => input / 2); + + IEnumerable? 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"); + } +}