From 4fd61924fca662128ff951f2d100cddec3707265 Mon Sep 17 00:00:00 2001 From: Stephen Toub Date: Tue, 30 Sep 2025 09:44:53 -0400 Subject: [PATCH] .NET: Clean up some CancellationToken usage (#996) * Clean up some CancellationToken usage * Unused usings --- .../AgentWebChat.AgentHost/Log.cs | 2 +- .../CheckpointAndRehydrate/WorkflowHelper.cs | 8 +- .../CheckpointAndResume/WorkflowHelper.cs | 8 +- .../WorkflowHelper.cs | 4 +- .../Microsoft.Agents.AI.A2A/A2AHostAgent.cs | 6 +- .../Extensions/A2ACardResolverExtensions.cs | 2 +- .../AgentCatalog.cs | 2 +- .../ActorResponseHandle.cs | 4 +- .../IActor.cs | 2 +- .../IActorClient.cs | 4 +- .../IActorRuntimeContext.cs | 6 +- .../IActorStateStorage.cs | 4 +- .../LazyCosmosContainer.cs | 2 +- .../Interpreter/DeclarativeActionExecutor.cs | 4 +- .../WorkflowAgentProvider.cs | 10 +- .../AgentWorkflowBuilder.cs | 10 +- .../AggregatingExecutor.cs | 8 +- .../ChatProtocolExecutor.cs | 6 +- .../Checkpointed.cs | 4 +- .../Checkpointing/ICheckpointingRunner.cs | 2 +- .../Execution/ISuperStepRunner.cs | 2 +- .../Microsoft.Agents.AI.Workflows/Executor.cs | 8 +- .../FunctionExecutor.cs | 8 +- .../InProc/InProcessRunner.cs | 46 +++---- .../InProc/InProcessRunnerContext.cs | 10 +- .../InProcessExecution.cs | 124 +++++++++--------- .../src/Microsoft.Agents.AI.Workflows/Run.cs | 22 ++-- .../Specialized/AIAgentHostExecutor.cs | 12 +- .../StreamingRun.cs | 22 ++-- .../WorkflowHostAgent.cs | 12 +- .../ChatClient/ChatClientAgent.cs | 2 +- .../RepresentationTests.cs | 2 +- .../Sample/03_Simple_Workflow_Loop.cs | 4 +- .../05_Simple_Workflow_Checkpointing.cs | 2 +- .../TestingExecutor.cs | 8 +- 35 files changed, 191 insertions(+), 191 deletions(-) diff --git a/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Log.cs b/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Log.cs index 3f98f0bbda..469fcea62b 100644 --- a/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Log.cs +++ b/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Log.cs @@ -44,7 +44,7 @@ internal static partial class Log [LoggerMessage( Level = LogLevel.Warning, Message = "SSE streaming cancelled: RequestId={RequestId}")] - public static partial void SseStreamingCancelled(ILogger logger, string requestId); + public static partial void SseStreamingCanceled(ILogger logger, string requestId); [LoggerMessage( Level = LogLevel.Error, diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs index 48ef7f38d9..3957ab97e5 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs @@ -93,14 +93,14 @@ internal sealed class GuessNumberExecutor() : ReflectingExecutor - protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => context.QueueStateUpdateAsync(StateKey, (this.LowerBound, this.UpperBound)); /// /// Restore the state of the executor from a checkpoint. /// This must be overridden to restore any state that was saved during checkpointing. /// - protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => (this.LowerBound, this.UpperBound) = await context.ReadStateAsync<(int, int)>(StateKey).ConfigureAwait(false); } @@ -143,13 +143,13 @@ internal sealed class JudgeExecutor() : ReflectingExecutor("Judge /// Checkpoint the current state of the executor. /// This must be overridden to save any state that is needed to resume the executor. /// - protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => context.QueueStateUpdateAsync(StateKey, this._tries); /// /// Restore the state of the executor from a checkpoint. /// This must be overridden to restore any state that was saved during checkpointing. /// - protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => this._tries = await context.ReadStateAsync(StateKey).ConfigureAwait(false); } diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs index f19a105b43..f5a22a89e3 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs @@ -93,14 +93,14 @@ internal sealed class GuessNumberExecutor() : ReflectingExecutor - protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => context.QueueStateUpdateAsync(StateKey, (this.LowerBound, this.UpperBound)); /// /// Restore the state of the executor from a checkpoint. /// This must be overridden to restore any state that was saved during checkpointing. /// - protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => (this.LowerBound, this.UpperBound) = await context.ReadStateAsync<(int, int)>(StateKey).ConfigureAwait(false); } @@ -143,13 +143,13 @@ internal sealed class JudgeExecutor() : ReflectingExecutor("Judge /// Checkpoint the current state of the executor. /// This must be overridden to save any state that is needed to resume the executor. /// - protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => context.QueueStateUpdateAsync(StateKey, this._tries); /// /// Restore the state of the executor from a checkpoint. /// This must be overridden to restore any state that was saved during checkpointing. /// - protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => this._tries = await context.ReadStateAsync(StateKey).ConfigureAwait(false); } diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs index 3ea1735284..2c3eb44747 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs @@ -93,13 +93,13 @@ internal sealed class JudgeExecutor() : ReflectingExecutor("Judge /// Checkpoint the current state of the executor. /// This must be overridden to save any state that is needed to resume the executor. /// - protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => context.QueueStateUpdateAsync(StateKey, this._tries); /// /// Restore the state of the executor from a checkpoint. /// This must be overridden to restore any state that was saved during checkpointing. /// - protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => + protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => this._tries = await context.ReadStateAsync(StateKey).ConfigureAwait(false); } diff --git a/dotnet/src/Microsoft.Agents.AI.A2A/A2AHostAgent.cs b/dotnet/src/Microsoft.Agents.AI.A2A/A2AHostAgent.cs index 718b027aec..446d0f65e0 100644 --- a/dotnet/src/Microsoft.Agents.AI.A2A/A2AHostAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI.A2A/A2AHostAgent.cs @@ -62,7 +62,7 @@ public sealed class A2AHostAgent /// Handle a received message /// /// The to handle - /// The to cancel the operation + /// The to monitor for cancellation requests. The default is . public async Task OnMessageReceivedAsync(MessageSendParams messageSend, CancellationToken cancellationToken = default) { Throw.IfNull(messageSend); @@ -92,8 +92,8 @@ public sealed class A2AHostAgent /// Return the associated with this hosted agent. /// /// Current URL for the agent - /// The to cancel the operation - public Task GetAgentCardAsync(string agentUrl, CancellationToken cancellationToken) + /// The to monitor for cancellation requests. The default is . + public Task GetAgentCardAsync(string agentUrl, CancellationToken cancellationToken = default) { // Ensure the URL is in the correct format Uri uri = new(agentUrl); diff --git a/dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2ACardResolverExtensions.cs b/dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2ACardResolverExtensions.cs index 4dd071de59..df4f6afcb3 100644 --- a/dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2ACardResolverExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2ACardResolverExtensions.cs @@ -38,7 +38,7 @@ public static class A2ACardResolverExtensions /// The to use for the agent creation. /// The to use for HTTP requests. /// The logger factory for enabling logging within the agent. - /// The to use when retrieving the agent card. + /// The to monitor for cancellation requests. The default is . /// An instance backed by the A2A agent. public static async Task GetAIAgentAsync(this A2ACardResolver resolver, HttpClient? httpClient = null, ILoggerFactory? loggerFactory = null, CancellationToken cancellationToken = default) { diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/AgentCatalog.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/AgentCatalog.cs index dc883d1416..0d2ef69640 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/AgentCatalog.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/AgentCatalog.cs @@ -24,7 +24,7 @@ public abstract class AgentCatalog /// /// Asynchronously retrieves all registered AI agents from the catalog. /// - /// A cancellation token that can be used to cancel the enumeration operation. + /// The to monitor for cancellation requests. The default is . /// /// An asynchronous enumerable of instances representing all registered agents. /// The enumeration will only include agents that are successfully resolved from the service provider. diff --git a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/ActorResponseHandle.cs b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/ActorResponseHandle.cs index b02327e4ce..4ed4a24fca 100644 --- a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/ActorResponseHandle.cs +++ b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/ActorResponseHandle.cs @@ -28,7 +28,7 @@ public abstract class ActorResponseHandle : IDisposable /// /// Gets the response from the completed request. /// - /// A token to cancel the wait operation. + /// The to monitor for cancellation requests. The default is . /// A task that completes when the request is finished. public abstract ValueTask GetResponseAsync(CancellationToken cancellationToken); @@ -41,7 +41,7 @@ public abstract class ActorResponseHandle : IDisposable /// /// Watches for status and data updates to the request. /// - /// A token to cancel the watch operation. + /// The to monitor for cancellation requests. The default is . /// An asynchronous enumerable of request updates. public abstract IAsyncEnumerable WatchUpdatesAsync(CancellationToken cancellationToken); diff --git a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActor.cs b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActor.cs index a7fc8d4b9b..df7de4c3ef 100644 --- a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActor.cs @@ -17,7 +17,7 @@ public interface IActor : IAsyncDisposable /// When the value returned from this method completes, the actor is considered stopped. /// IActor is expected to call IActorContext.WatchMessagesAsync() to receive messages. /// - /// A token to cancel the start operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the start operation. ValueTask RunAsync(CancellationToken cancellationToken); } diff --git a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorClient.cs b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorClient.cs index ee0b9fa29e..87ad973dfe 100644 --- a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorClient.cs +++ b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorClient.cs @@ -15,7 +15,7 @@ public interface IActorClient /// This method is idempotent: if the request is already in progress, it will return the existing response. /// /// The request to send to the actor. - /// A token to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the actor response handle. ValueTask SendRequestAsync(ActorRequest request, CancellationToken cancellationToken); @@ -24,7 +24,7 @@ public interface IActorClient /// /// The identifier of the actor processing the request. /// The unique identifier of the request message. - /// A token to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the actor response handle. ValueTask GetResponseAsync(ActorId actorId, string messageId, CancellationToken cancellationToken); } diff --git a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorRuntimeContext.cs b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorRuntimeContext.cs index cbd91a84ed..61ed7638a3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorRuntimeContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorRuntimeContext.cs @@ -20,7 +20,7 @@ public interface IActorRuntimeContext /// /// Watches for incoming requests and responses in the actor's inbox and outbox. /// - /// A token to cancel the watch operation. + /// The to monitor for cancellation requests. The default is . /// An asynchronous enumerable of actor notifications. IAsyncEnumerable WatchMessagesAsync(CancellationToken cancellationToken = default); @@ -28,7 +28,7 @@ public interface IActorRuntimeContext /// Performs a batch of write operations atomically. /// /// The batch of write operations to perform. - /// A token to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the write response. ValueTask WriteAsync(ActorWriteOperationBatch operations, CancellationToken cancellationToken = default); @@ -36,7 +36,7 @@ public interface IActorRuntimeContext /// Performs a batch of read operations. /// /// The batch of read operations to perform. - /// A token to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the read response. ValueTask ReadAsync(ActorReadOperationBatch operations, CancellationToken cancellationToken = default); diff --git a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorStateStorage.cs b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorStateStorage.cs index 34ba185518..e8bd011d25 100644 --- a/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorStateStorage.cs +++ b/dotnet/src/Microsoft.Agents.AI.Runtime.Abstractions/IActorStateStorage.cs @@ -17,7 +17,7 @@ public interface IActorStateStorage /// The identifier of the actor whose state is being modified. /// The collection of write operations to perform. /// The expected ETag for optimistic concurrency control. - /// A token to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the write response with success status and updated ETag. ValueTask WriteStateAsync(ActorId actorId, IReadOnlyCollection operations, string etag, CancellationToken cancellationToken = default); @@ -26,7 +26,7 @@ public interface IActorStateStorage /// /// The identifier of the actor whose state is being read. /// The collection of read operations to perform. - /// A token to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A task representing the read response with results and current ETag. ValueTask ReadStateAsync(ActorId actorId, IReadOnlyCollection operations, CancellationToken cancellationToken = default); } diff --git a/dotnet/src/Microsoft.Agents.AI.Runtime.Storage.CosmosDB/LazyCosmosContainer.cs b/dotnet/src/Microsoft.Agents.AI.Runtime.Storage.CosmosDB/LazyCosmosContainer.cs index d68fb1eca4..78db567e95 100644 --- a/dotnet/src/Microsoft.Agents.AI.Runtime.Storage.CosmosDB/LazyCosmosContainer.cs +++ b/dotnet/src/Microsoft.Agents.AI.Runtime.Storage.CosmosDB/LazyCosmosContainer.cs @@ -136,7 +136,7 @@ internal sealed class LazyCosmosContainer : IAsyncDisposable HttpStatusCode.RequestTimeout => true, // 408 - Request timeout _ => false }, - TaskCanceledException or OperationCanceledException or ArgumentException => false, + OperationCanceledException or ArgumentException => false, _ => true // Retry other exceptions (network issues, etc.) }; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs index 21c3bf5e53..aa3da84f40 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs @@ -99,8 +99,8 @@ internal abstract class DeclarativeActionExecutor : Executor - protected override ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) => - this._state.RestoreAsync(context, cancellation); + protected override ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => + this._state.RestoreAsync(context, cancellationToken); protected async ValueTask AssignAsync(PropertyPath? targetPath, FormulaValue result, IWorkflowContext context) { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/WorkflowAgentProvider.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/WorkflowAgentProvider.cs index 39497eb72a..6eb46c0f39 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/WorkflowAgentProvider.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/WorkflowAgentProvider.cs @@ -16,14 +16,14 @@ public abstract class WorkflowAgentProvider /// Asynchronously retrieves an AI agent by its unique identifier. /// /// The unique identifier of the AI agent to retrieve. Cannot be null or empty. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// The task result contains the associated. public abstract Task GetAgentAsync(string agentId, CancellationToken cancellationToken = default); /// /// Asynchronously creates a new conversation and returns its unique identifier. /// - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// The conversation identifier public abstract Task CreateConversationAsync(CancellationToken cancellationToken = default); @@ -32,7 +32,7 @@ public abstract class WorkflowAgentProvider /// /// The identifier of the target conversation. /// The message being added. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . public abstract Task CreateMessageAsync(string conversationId, ChatMessage conversationMessage, CancellationToken cancellationToken = default); /// @@ -40,7 +40,7 @@ public abstract class WorkflowAgentProvider /// /// The identifier of the target conversation. /// The identifier of the target message. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// The requested message public abstract Task GetMessageAsync(string conversationId, string messageId, CancellationToken cancellationToken = default); @@ -52,7 +52,7 @@ public abstract class WorkflowAgentProvider /// A cursor for use in pagination. after is an object ID that defines your place in the list. /// A cursor for use in pagination. before is an object ID that defines your place in the list. /// Provide records in descending order when true. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// The requested messages public abstract IAsyncEnumerable GetMessagesAsync( string conversationId, diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs index 3d60d993d7..2705fb111a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs @@ -198,7 +198,7 @@ public static partial class AgentWorkflowBuilder /// private sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages"), IResettableExecutor { - protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default) + protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) => context.YieldOutputAsync(messages); ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync(); @@ -223,7 +223,7 @@ public static partial class AgentWorkflowBuilder /// private sealed class BatchChatMessagesToListExecutor(string id) : ChatProtocolExecutor(id), IResettableExecutor { - protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default) + protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) => context.SendMessageAsync(messages); ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync(); @@ -623,7 +623,7 @@ public static partial class AgentWorkflowBuilder /// Selects the next agent to participate in the group chat based on the provided chat history and team. /// /// The chat history to consider. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// The next to speak. This agent must be part of the chat. protected internal abstract ValueTask SelectNextAgentAsync( IReadOnlyList history, @@ -633,7 +633,7 @@ public static partial class AgentWorkflowBuilder /// Filters the chat history before it's passed to the next agent. /// /// The chat history to filter. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// The filtered chat history. protected internal virtual ValueTask> UpdateHistoryAsync( IReadOnlyList history, @@ -644,7 +644,7 @@ public static partial class AgentWorkflowBuilder /// Determines whether the group chat should be terminated based on the provided chat history and iteration count. /// /// The chat history to consider. - /// A cancellation token that can be used to cancel the operation. + /// The to monitor for cancellation requests. The default is . /// A indicating whether the chat should be terminated. protected internal virtual ValueTask ShouldTerminateAsync( IReadOnlyList history, diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/AggregatingExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/AggregatingExecutor.cs index a9d3f9c939..c6cb36abe3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/AggregatingExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/AggregatingExecutor.cs @@ -35,17 +35,17 @@ public class AggregatingExecutor(string id, } /// - protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { await context.QueueStateUpdateAsync(AggregateStateKey, this._runningAggregate).ConfigureAwait(false); - await base.OnCheckpointingAsync(context, cancellation).ConfigureAwait(false); + await base.OnCheckpointingAsync(context, cancellationToken).ConfigureAwait(false); } /// - protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { - await base.OnCheckpointRestoredAsync(context, cancellation).ConfigureAwait(false); + await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false); this._runningAggregate = await context.ReadStateAsync(AggregateStateKey).ConfigureAwait(false); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs index 09e791b034..afcc7e52cc 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs @@ -45,10 +45,10 @@ internal abstract class ChatProtocolExecutor(string id, ChatProtocolExecutorOpti await context.SendMessageAsync(token).ConfigureAwait(false); } - protected abstract ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default); + protected abstract ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default); private const string PendingMessagesStateKey = nameof(_pendingMessages); - protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { Task messagesTask = Task.CompletedTask; if (this._pendingMessages.Count > 0) @@ -60,7 +60,7 @@ internal abstract class ChatProtocolExecutor(string id, ChatProtocolExecutorOpti await messagesTask.ConfigureAwait(false); } - protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { JsonElement? messagesValue = await context.ReadStateAsync(PendingMessagesStateKey).ConfigureAwait(false); if (messagesValue.HasValue) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs index e1579d85ba..960b14cc54 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs @@ -47,6 +47,6 @@ public class Checkpointed } /// - public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default) - => this._runner.RestoreCheckpointAsync(checkpointInfo, cancellation); + public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default) + => this._runner.RestoreCheckpointAsync(checkpointInfo, cancellationToken); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointingRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointingRunner.cs index d0a8314b04..05a81115a4 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointingRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointingRunner.cs @@ -11,5 +11,5 @@ 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); + ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs index 65d34ee88f..6c317aa32d 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs @@ -18,7 +18,7 @@ internal interface ISuperStepRunner event EventHandler? WorkflowEvent; - ValueTask RunSuperStepAsync(CancellationToken cancellation); + ValueTask RunSuperStepAsync(CancellationToken cancellationToken); ValueTask RequestEndRunAsync(); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs index 66cece7b01..65bc6e8732 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs @@ -139,16 +139,16 @@ public abstract class Executor : IIdentified /// /// The workflow context. /// A ValueTask representing the asynchronous operation. - /// - protected internal virtual ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) => default; + /// The to monitor for cancellation requests. The default is . + protected internal virtual ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = 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; + /// The to monitor for cancellation requests. The default is . + protected internal virtual ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => default; /// /// A set of s, representing the messages this executor can handle. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/FunctionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/FunctionExecutor.cs index 92920f2e14..0db15f42bf 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/FunctionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/FunctionExecutor.cs @@ -21,9 +21,9 @@ public class FunctionExecutor(string id, { return RunActionAsync; - ValueTask RunActionAsync(TInput input, IWorkflowContext workflowContext, CancellationToken cancellation) + ValueTask RunActionAsync(TInput input, IWorkflowContext workflowContext, CancellationToken cancellationToken) { - handlerSync(input, workflowContext, cancellation); + handlerSync(input, workflowContext, cancellationToken); return default; } } @@ -57,9 +57,9 @@ public class FunctionExecutor(string id, { return RunFuncAsync; - ValueTask RunFuncAsync(TInput input, IWorkflowContext workflowContext, CancellationToken cancellation) + ValueTask RunFuncAsync(TInput input, IWorkflowContext workflowContext, CancellationToken cancellationToken) { - TOutput result = handlerSync(input, workflowContext, cancellation); + TOutput result = handlerSync(input, workflowContext, cancellationToken); return new ValueTask(result); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs index ca2184cb87..e82292b753 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs @@ -108,7 +108,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner this.WorkflowEvent?.Invoke(this, workflowEvent); } - public async ValueTask ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default) + public async ValueTask ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); Throw.IfNull(checkpoint); @@ -117,12 +117,12 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner throw new InvalidOperationException("This runner was not configured with a CheckpointManager, so it cannot restore checkpoints."); } - await this.RestoreCheckpointAsync(checkpoint, cancellation).ConfigureAwait(false); + await this.RestoreCheckpointAsync(checkpoint, cancellationToken).ConfigureAwait(false); return new StreamingRun(this); } - public async ValueTask StreamAsync(object input, CancellationToken cancellation = default) + public async ValueTask StreamAsync(object input, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); await this.EnqueueMessageAsync(input).ConfigureAwait(false); @@ -130,7 +130,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner return new StreamingRun(this); } - public async ValueTask StreamAsync(TInput input, CancellationToken cancellation = default) + public async ValueTask StreamAsync(TInput input, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); await this.EnqueueMessageAsync(input).ConfigureAwait(false); @@ -138,31 +138,31 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner return new StreamingRun(this); } - internal async ValueTask ResumeAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default) + internal async ValueTask ResumeAsync(CheckpointInfo checkpoint, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); - StreamingRun streamingRun = await this.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false); - cancellation.ThrowIfCancellationRequested(); + StreamingRun streamingRun = await this.ResumeStreamAsync(checkpoint, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); - return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false); + return await Run.CaptureStreamAsync(streamingRun, cancellationToken).ConfigureAwait(false); } - public async ValueTask RunAsync(object input, CancellationToken cancellation = default) + public async ValueTask RunAsync(object input, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); - StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false); - cancellation.ThrowIfCancellationRequested(); + StreamingRun streamingRun = await this.StreamAsync(input, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); - return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false); + return await Run.CaptureStreamAsync(streamingRun, cancellationToken).ConfigureAwait(false); } - public async ValueTask RunAsync(TInput input, CancellationToken cancellation = default) + public async ValueTask RunAsync(TInput input, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); - StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false); - cancellation.ThrowIfCancellationRequested(); + StreamingRun streamingRun = await this.StreamAsync(input, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); - return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false); + return await Run.CaptureStreamAsync(streamingRun, cancellationToken).ConfigureAwait(false); } bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests; @@ -170,10 +170,10 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner public IReadOnlyList Checkpoints => this._checkpoints; - async ValueTask ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellation) + async ValueTask ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellationToken) { this.RunContext.CheckEnded(); - if (cancellation.IsCancellationRequested) + if (cancellationToken.IsCancellationRequested) { return false; } @@ -239,7 +239,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner private WorkflowInfo? _workflowInfoCache; private readonly List _checkpoints = []; - internal async ValueTask CheckpointAsync(CancellationToken cancellation = default) + internal async ValueTask CheckpointAsync(CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); if (this.CheckpointManager is null) @@ -250,7 +250,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner } // Notify all the executors that they should prepare for checkpointing. - Task prepareTask = this.RunContext.PrepareForCheckpointAsync(cancellation); + Task prepareTask = this.RunContext.PrepareForCheckpointAsync(cancellationToken); // Create a representation of the current workflow if it does not already exist. this._workflowInfoCache ??= this.Workflow.ToWorkflowInfo(); @@ -269,7 +269,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner this._checkpoints.Add(checkpointInfo); } - public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default) + public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); Throw.IfNull(checkpointInfo); @@ -291,8 +291,8 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner await this.RunContext.StateManager.ImportStateAsync(checkpoint).ConfigureAwait(false); await this.RunContext.ImportStateAsync(checkpoint).ConfigureAwait(false); - Task executorNotifyTask = this.RunContext.NotifyCheckpointLoadedAsync(cancellation); - ValueTask republishRequestsTask = this.RunContext.RepublishUnservicedRequestsAsync(cancellation); + Task executorNotifyTask = this.RunContext.NotifyCheckpointLoadedAsync(cancellationToken); + ValueTask republishRequestsTask = this.RunContext.RepublishUnservicedRequestsAsync(cancellationToken); await this.EdgeMap.ImportStateAsync(checkpoint).ConfigureAwait(false); await Task.WhenAll(executorNotifyTask, republishRequestsTask.AsTask()).ConfigureAwait(false); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index 4d2521ed48..49741bb85f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -210,7 +210,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext => RunnerContext.StateManager.ClearStateAsync(ExecutorId, scopeName); } - internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default) + internal Task PrepareForCheckpointAsync(CancellationToken cancellationToken = default) { this.CheckEnded(); @@ -219,11 +219,11 @@ internal sealed class InProcessRunnerContext : IRunnerContext async Task InvokeCheckpointingAsync(Task executorTask) { Executor executor = await executorTask.ConfigureAwait(false); - await executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellation).ConfigureAwait(false); + await executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellationToken).ConfigureAwait(false); } } - internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellation = default) + internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellationToken = default) { this.CheckEnded(); @@ -232,7 +232,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext async Task InvokeCheckpointRestoredAsync(Task executorTask) { Executor executor = await executorTask.ConfigureAwait(false); - await executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellation).ConfigureAwait(false); + await executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellationToken).ConfigureAwait(false); } } @@ -253,7 +253,7 @@ internal sealed class InProcessRunnerContext : IRunnerContext return new(result); } - internal async ValueTask RepublishUnservicedRequestsAsync(CancellationToken cancellation = default) + internal async ValueTask RepublishUnservicedRequestsAsync(CancellationToken cancellationToken = default) { this.CheckEnded(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs index 2b94548ffd..806e3567dd 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs @@ -19,61 +19,61 @@ public static class InProcessExecution where TInput : notnull => new(checkedWorkflow, checkpointManager, runId, [typeof(TInput)]); - private static ValueTask StreamAsync(InProcessRunner runner, TInput input, CancellationToken cancellation = default) - => runner.StreamAsync(input, cancellation); + private static ValueTask StreamAsync(InProcessRunner runner, TInput input, CancellationToken cancellationToken = default) + => runner.StreamAsync(input, cancellationToken); - private static ValueTask StreamAsync(InProcessRunner runner, object input, CancellationToken cancellation = default) - => runner.StreamAsync(input, cancellation); + private static ValueTask StreamAsync(InProcessRunner runner, object input, CancellationToken cancellationToken = default) + => runner.StreamAsync(input, cancellationToken); - private static ValueTask RunAsync(InProcessRunner runner, TInput input, CancellationToken cancellation = default) - => runner.RunAsync(input, cancellation); + private static ValueTask RunAsync(InProcessRunner runner, TInput input, CancellationToken cancellationToken = default) + => runner.RunAsync(input, cancellationToken); - private static ValueTask RunAsync(InProcessRunner runner, object input, CancellationToken cancellation = default) - => runner.RunAsync(input, cancellation); + private static ValueTask RunAsync(InProcessRunner runner, object input, CancellationToken cancellationToken = default) + => runner.RunAsync(input, cancellationToken); - private static async ValueTask> StreamCheckpointedAsync(InProcessRunner runner, TInput input, CancellationToken cancellation = default) + private static async ValueTask> StreamCheckpointedAsync(InProcessRunner runner, TInput input, CancellationToken cancellationToken = default) where TInput : notnull { - StreamingRun run = await StreamAsync(runner, input, cancellation).ConfigureAwait(false); - await runner.CheckpointAsync(cancellation).ConfigureAwait(false); + StreamingRun run = await StreamAsync(runner, input, cancellationToken).ConfigureAwait(false); + await runner.CheckpointAsync(cancellationToken).ConfigureAwait(false); return new(run, runner); } - private static async ValueTask> StreamCheckpointedAsync(InProcessRunner runner, object input, CancellationToken cancellation = default) + private static async ValueTask> StreamCheckpointedAsync(InProcessRunner runner, object input, CancellationToken cancellationToken = default) { - StreamingRun run = await StreamAsync(runner, input, cancellation).ConfigureAwait(false); - await runner.CheckpointAsync(cancellation).ConfigureAwait(false); + StreamingRun run = await StreamAsync(runner, input, cancellationToken).ConfigureAwait(false); + await runner.CheckpointAsync(cancellationToken).ConfigureAwait(false); return new(run, runner); } - private static async ValueTask> RunCheckpointedAsync(InProcessRunner runner, TInput input, CancellationToken cancellation = default) + private static async ValueTask> RunCheckpointedAsync(InProcessRunner runner, TInput input, CancellationToken cancellationToken = default) where TInput : notnull { - Run run = await RunAsync(runner, input, cancellation).ConfigureAwait(false); - await runner.CheckpointAsync(cancellation).ConfigureAwait(false); + Run run = await RunAsync(runner, input, cancellationToken).ConfigureAwait(false); + await runner.CheckpointAsync(cancellationToken).ConfigureAwait(false); return new(run, runner); } - private static async ValueTask> RunCheckpointedAsync(InProcessRunner runner, object input, CancellationToken cancellation = default) + private static async ValueTask> RunCheckpointedAsync(InProcessRunner runner, object input, CancellationToken cancellationToken = default) { - Run run = await RunAsync(runner, input, cancellation).ConfigureAwait(false); - await runner.CheckpointAsync(cancellation).ConfigureAwait(false); + Run run = await RunAsync(runner, input, cancellationToken).ConfigureAwait(false); + await runner.CheckpointAsync(cancellationToken).ConfigureAwait(false); return new(run, runner); } - private static async ValueTask> ResumeStreamCheckpointedAsync(InProcessRunner runner, CheckpointInfo fromCheckpoint, CancellationToken cancellation = default) + private static async ValueTask> ResumeStreamCheckpointedAsync(InProcessRunner runner, CheckpointInfo fromCheckpoint, CancellationToken cancellationToken = default) { - StreamingRun run = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false); + StreamingRun run = await runner.ResumeStreamAsync(fromCheckpoint, cancellationToken).ConfigureAwait(false); return new(run, runner); } - private static async ValueTask> ResumeRunCheckpointedAsync(InProcessRunner runner, CheckpointInfo fromCheckpoint, CancellationToken cancellation = default) + private static async ValueTask> ResumeRunCheckpointedAsync(InProcessRunner runner, CheckpointInfo fromCheckpoint, CancellationToken cancellationToken = default) { - Run run = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false); + Run run = await runner.ResumeAsync(fromCheckpoint, cancellationToken).ConfigureAwait(false); return new(run, runner); } @@ -87,17 +87,17 @@ public static class InProcessExecution /// The workflow to be executed. Must not be null. /// The input message to be processed as part of the streaming run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// 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, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId); - return StreamAsync(runner, (object)input, cancellation); + return StreamAsync(runner, (object)input, cancellationToken); } /// @@ -110,17 +110,17 @@ public static class InProcessExecution /// The workflow to be executed. Must not be null. /// The input message to be processed as part of the streaming run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// 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, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId); - return StreamAsync(runner, input, cancellation); + return StreamAsync(runner, input, cancellationToken); } /// @@ -134,7 +134,7 @@ public static class InProcessExecution /// The input message to be processed as part of the streaming run. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. public static ValueTask> StreamAsync( @@ -142,10 +142,10 @@ public static class InProcessExecution TInput input, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId); - return StreamCheckpointedAsync(runner, (object)input, cancellation); + return StreamCheckpointedAsync(runner, (object)input, cancellationToken); } /// @@ -159,7 +159,7 @@ public static class InProcessExecution /// The input message to be processed as part of the streaming run. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. public static ValueTask> StreamAsync( @@ -167,55 +167,55 @@ public static class InProcessExecution TInput input, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId); - return StreamCheckpointedAsync(runner, input, cancellation); + return StreamCheckpointedAsync(runner, input, cancellationToken); } /// /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. /// - /// If the operation is cancelled via the token, the streaming execution will + /// If the operation is cancelled via the token, the streaming execution will /// be terminated. /// The workflow to be executed. Must not be null. /// The corresponding to the checkpoint from which to resume. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that provides access to the results of the streaming run. public static ValueTask> ResumeStreamAsync( Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) + CancellationToken cancellationToken = default) { InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId); - return ResumeStreamCheckpointedAsync(runner, fromCheckpoint, cancellation); + return ResumeStreamCheckpointedAsync(runner, fromCheckpoint, cancellationToken); } /// /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. /// - /// If the operation is cancelled via the token, the streaming execution will + /// 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. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that provides access to the results of the streaming run. public static ValueTask> ResumeStreamAsync( Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId); - return ResumeStreamCheckpointedAsync(runner, fromCheckpoint, cancellation); + return ResumeStreamCheckpointedAsync(runner, fromCheckpoint, cancellationToken); } /// @@ -227,17 +227,17 @@ public static class InProcessExecution /// The workflow to be executed. Must not be null. /// The input message to be processed as part of the run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// 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, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId); - return RunAsync(runner, (object)input, cancellation); + return RunAsync(runner, (object)input, cancellationToken); } /// @@ -249,17 +249,17 @@ public static class InProcessExecution /// The workflow to be executed. Must not be null. /// The input message to be processed as part of the run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// 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, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId); - return RunAsync(runner, input, cancellation); + return RunAsync(runner, input, cancellationToken); } /// @@ -272,7 +272,7 @@ public static class InProcessExecution /// The input message to be processed as part of the run. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. public static ValueTask> RunAsync( @@ -280,10 +280,10 @@ public static class InProcessExecution TInput input, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager: checkpointManager, runId); - return RunCheckpointedAsync(runner, (object)input, cancellation); + return RunCheckpointedAsync(runner, (object)input, cancellationToken); } /// @@ -296,7 +296,7 @@ public static class InProcessExecution /// The input message to be processed as part of the run. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. public static ValueTask> RunAsync( @@ -304,10 +304,10 @@ public static class InProcessExecution TInput input, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager: checkpointManager, runId); - return RunCheckpointedAsync(runner, input, cancellation); + return RunCheckpointedAsync(runner, input, cancellationToken); } /// @@ -319,7 +319,7 @@ public static class InProcessExecution /// The corresponding to the checkpoint from which to resume. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. public static ValueTask> ResumeAsync( @@ -327,10 +327,10 @@ public static class InProcessExecution CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) + CancellationToken cancellationToken = default) { InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId); - return ResumeRunCheckpointedAsync(runner, fromCheckpoint, cancellation); + return ResumeRunCheckpointedAsync(runner, fromCheckpoint, cancellationToken); } /// @@ -342,7 +342,7 @@ public static class InProcessExecution /// The corresponding to the checkpoint from which to resume. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// A that can be used to cancel the streaming operation. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. public static ValueTask> ResumeAsync( @@ -350,9 +350,9 @@ public static class InProcessExecution CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellation = default) where TInput : notnull + CancellationToken cancellationToken = default) where TInput : notnull { InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId); - return ResumeRunCheckpointedAsync(runner, fromCheckpoint, cancellation); + return ResumeRunCheckpointedAsync(runner, fromCheckpoint, cancellationToken); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs index 113a47028d..5773e1e243 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs @@ -40,10 +40,10 @@ public enum RunStatus /// public sealed class Run { - internal static async ValueTask CaptureStreamAsync(StreamingRun run, CancellationToken cancellation = default) + internal static async ValueTask CaptureStreamAsync(StreamingRun run, CancellationToken cancellationToken = default) { Run result = new(run); - await result.RunToNextHaltAsync(cancellation).ConfigureAwait(false); + await result.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); return result; } @@ -54,11 +54,11 @@ public sealed class Run this._streamingRun = streamingRun; } - internal async ValueTask RunToNextHaltAsync(CancellationToken cancellation = default) + internal async ValueTask RunToNextHaltAsync(CancellationToken cancellationToken = default) { bool hadEvents = false; this.Status = RunStatus.Running; - await foreach (WorkflowEvent evt in this._streamingRun.WatchStreamAsync(blockOnPendingRequest: false, cancellation).ConfigureAwait(false)) + await foreach (WorkflowEvent evt in this._streamingRun.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken).ConfigureAwait(false)) { hadEvents = true; this._eventSink.Add(evt); @@ -113,31 +113,31 @@ public sealed class Run /// /// Resume execution of the workflow with the provided external responses. /// - /// A that can be used to cancel the workflow execution. /// An array of objects to send to the workflow. + /// The to monitor for cancellation requests. The default is . /// true if the workflow had any output events, false otherwise. - public async ValueTask ResumeAsync(CancellationToken cancellation = default, params IEnumerable responses) + public async ValueTask ResumeAsync(IEnumerable responses, CancellationToken cancellationToken = default) { foreach (ExternalResponse response in responses) { await this._streamingRun.SendResponseAsync(response).ConfigureAwait(false); } - return await this.RunToNextHaltAsync(cancellation).ConfigureAwait(false); + return await this.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); } /// /// Resume execution of the workflow with the provided external responses. /// - /// A that can be used to cancel the workflow execution. /// An array of messages to send to the workflow. Messages will only be sent if they are valid /// input types to the starting executor or a . + /// The to monitor for cancellation requests. The default is . /// true if the workflow had any output events, false otherwise. - public async ValueTask ResumeAsync(CancellationToken cancellation = default, params IEnumerable messages) + public async ValueTask ResumeAsync(IEnumerable messages, CancellationToken cancellationToken = default) { if (messages is IEnumerable responses) { - return await this.ResumeAsync(cancellation, responses).ConfigureAwait(false); + return await this.ResumeAsync(responses, cancellationToken).ConfigureAwait(false); } foreach (T message in messages) @@ -145,7 +145,7 @@ public sealed class Run await this._streamingRun.TrySendMessageAsync(message).ConfigureAwait(false); } - return await this.RunToNextHaltAsync(cancellation).ConfigureAwait(false); + return await this.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); } /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs index 0d6d18cf8d..e50e0f803a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs @@ -24,7 +24,7 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor this._thread ??= this._agent.GetNewThread(); private const string ThreadStateKey = nameof(_thread); - protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { Task threadTask = Task.CompletedTask; if (this._thread is not null) @@ -33,12 +33,12 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor threadTask = context.QueueStateUpdateAsync(ThreadStateKey, threadValue).AsTask(); } - Task baseTask = base.OnCheckpointingAsync(context, cancellation).AsTask(); + Task baseTask = base.OnCheckpointingAsync(context, cancellationToken).AsTask(); await Task.WhenAll(threadTask, baseTask).ConfigureAwait(false); } - protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { JsonElement? threadValue = await context.ReadStateAsync(ThreadStateKey).ConfigureAwait(false); if (threadValue.HasValue) @@ -46,13 +46,13 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor this._thread = this._agent.DeserializeThread(threadValue.Value); } - await base.OnCheckpointRestoredAsync(context, cancellation).ConfigureAwait(false); + await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false); } - protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default) + protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) { emitEvents ??= this._emitEvents; - IAsyncEnumerable agentStream = this._agent.RunStreamingAsync(messages, this.EnsureThread(context), cancellationToken: cancellation); + IAsyncEnumerable agentStream = this._agent.RunStreamingAsync(messages, this.EnsureThread(context), cancellationToken: cancellationToken); List updates = []; ChatMessage? currentStreamingMessage = null; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs index cdcef89ffd..5b325d8e8b 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs @@ -79,17 +79,17 @@ public sealed class StreamingRun /// This method yields instances in real time as the workflow /// progresses. The stream completes when a is encountered. Events are /// delivered in the order they are raised. - /// A that can be used to cancel the streaming operation. If cancellation is + /// A that can be used to cancel the streaming operation. If cancellation is /// requested, the stream will end and no further events will be yielded. /// An asynchronous stream of objects representing significant workflow state changes. /// The stream ends when the workflow completes or when cancellation is requested. public IAsyncEnumerable WatchStreamAsync( - CancellationToken cancellation = default) - => this.WatchStreamAsync(blockOnPendingRequest: true, cancellation); + CancellationToken cancellationToken = default) + => this.WatchStreamAsync(blockOnPendingRequest: true, cancellationToken); internal async IAsyncEnumerable WatchStreamAsync( bool blockOnPendingRequest, - [EnumeratorCancellation] CancellationToken cancellation = default) + [EnumeratorCancellation] CancellationToken cancellationToken = default) { List eventSink = []; @@ -100,8 +100,8 @@ public sealed class StreamingRun do { // Drain SuperSteps while there are steps to run - await this._stepRunner.RunSuperStepAsync(cancellation).ConfigureAwait(false); - if (cancellation.IsCancellationRequested) + await this._stepRunner.RunSuperStepAsync(cancellationToken).ConfigureAwait(false); + if (cancellationToken.IsCancellationRequested) { yield break; // Exit if cancellation is requested } @@ -109,7 +109,7 @@ public sealed class StreamingRun bool hadCompletionEvent = false; foreach (WorkflowEvent raisedEvent in Interlocked.Exchange(ref eventSink, [])) { - if (cancellation.IsCancellationRequested) + if (cancellationToken.IsCancellationRequested) { yield break; // Exit if cancellation is requested } @@ -140,7 +140,7 @@ public sealed class StreamingRun { this._waitForResponseSource ??= new(); - using CancellationTokenRegistration registration = cancellation.Register(() => this._waitForResponseSource?.SetResult(new())); + using CancellationTokenRegistration registration = cancellationToken.Register(() => this._waitForResponseSource?.SetResult(new())); await this._waitForResponseSource.Task.ConfigureAwait(false); this._waitForResponseSource = null; @@ -182,14 +182,14 @@ public static class StreamingRunExtensions /// An optional callback function invoked for each received from the stream. /// The callback can return a response object to be sent back to the workflow, or if no response /// is required. - /// A to observe while waiting for events. + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The task completes when the workflow /// execution stream is fully processed. - public static async ValueTask RunToCompletionAsync(this StreamingRun handle, Func? eventCallback = null, CancellationToken cancellation = default) + public static async ValueTask RunToCompletionAsync(this StreamingRun handle, Func? eventCallback = null, CancellationToken cancellationToken = default) { Throw.IfNull(handle); - await foreach (WorkflowEvent @event in handle.WatchStreamAsync(cancellation).ConfigureAwait(false)) + await foreach (WorkflowEvent @event in handle.WatchStreamAsync(cancellationToken).ConfigureAwait(false)) { ExternalResponse? maybeResponse = eventCallback?.Invoke(@event); if (maybeResponse is not null) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs index c5625a12b6..7add0138a5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs @@ -53,7 +53,7 @@ internal sealed class WorkflowHostAgent : AIAgent private async IAsyncEnumerable InvokeStageAsync( WorkflowThread conversation, - [EnumeratorCancellation] CancellationToken cancellation = default) + [EnumeratorCancellation] CancellationToken cancellationToken = default) { string runId = conversation.RunId; List messages = conversation.MessageStore.GetFromBookmark().ToList(); @@ -64,7 +64,7 @@ internal sealed class WorkflowHostAgent : AIAgent // in the case of new threads. if (!this._runningWorkflows.TryGetValue(runId, out StreamingRun? run)) { - run = await InProcessExecution.StreamAsync(this._workflow, messages, cancellation: cancellation) + run = await InProcessExecution.StreamAsync(this._workflow, messages, cancellationToken: cancellationToken) .ConfigureAwait(false); this._runningWorkflows[runId] = run; } @@ -75,9 +75,9 @@ internal sealed class WorkflowHostAgent : AIAgent } await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); - await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellation) + await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken) .ConfigureAwait(false) - .WithCancellation(cancellation)) + .WithCancellation(cancellationToken)) { switch (evt) { @@ -99,7 +99,7 @@ internal sealed class WorkflowHostAgent : AIAgent } } - private async ValueTask UpdateThreadAsync(IEnumerable messages, AgentThread? thread = null, CancellationToken cancellation = default) + private async ValueTask UpdateThreadAsync(IEnumerable messages, AgentThread? thread = null, CancellationToken cancellationToken = default) { thread ??= this.GetNewThread(); @@ -108,7 +108,7 @@ internal sealed class WorkflowHostAgent : AIAgent throw new ArgumentException($"Incompatible thread type: {thread.GetType()} (expecting {typeof(WorkflowThread)})", nameof(thread)); } - await workflowThread.MessageStore.AddMessagesAsync(messages, cancellation).ConfigureAwait(false); + await workflowThread.MessageStore.AddMessagesAsync(messages, cancellationToken).ConfigureAwait(false); return workflowThread; } diff --git a/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs b/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs index cc5ddf31c3..a357a4e8bd 100644 --- a/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs @@ -467,7 +467,7 @@ public sealed class ChatClientAgent : AIAgent /// The conversation thread to use or create. /// The input messages to use. /// Optional parameters for agent invocation. - /// The cancellation token. + /// The to monitor for cancellation requests. The default is . /// A tuple containing the thread, chat options, and thread messages. private async Task<(ChatClientAgentThread AgentThread, ChatOptions? ChatOptions, List ThreadMessages)> PrepareThreadAndMessagesAsync( AgentThread? thread, diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs index 088f2af816..656ce96aa3 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs @@ -68,7 +68,7 @@ public class RepresentationTests testsRun++; } - async ValueTask MessageHandlerAsync(int message, IWorkflowContext workflowContext, CancellationToken cancellation = default) + async ValueTask MessageHandlerAsync(int message, IWorkflowContext workflowContext, CancellationToken cancellationToken = default) { } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs index b77cb45dfc..4da5e82407 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/03_Simple_Workflow_Loop.cs @@ -113,12 +113,12 @@ internal sealed class JudgeExecutor : ReflectingExecutor, IMessag NumberSignal.Above; } - protected internal override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { return context.QueueStateUpdateAsync("TryCount", this.Tries); } - protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default) + protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { this.Tries = await context.ReadStateAsync("TryCount").ConfigureAwait(false) ?? 0; } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs index 5f94f78a24..df14d6e892 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/05_Simple_Workflow_Checkpointing.cs @@ -41,7 +41,7 @@ internal static class Step5EntryPoint { await handle.EndRunAsync().ConfigureAwait(false); - checkpointed = await InProcessExecution.ResumeStreamAsync(workflow, targetCheckpoint, checkpointManager, runId: handle.RunId, cancellation: CancellationToken.None) + checkpointed = await InProcessExecution.ResumeStreamAsync(workflow, targetCheckpoint, checkpointManager, runId: handle.RunId, cancellationToken: CancellationToken.None) .ConfigureAwait(false); handle = checkpointed.Run; } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestingExecutor.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestingExecutor.cs index 4396b2d849..210f3aa89b 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestingExecutor.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestingExecutor.cs @@ -25,12 +25,12 @@ internal abstract class TestingExecutor : Executor, IDisposable this._actions = actions; } - public void UnlinkCancellation(CancellationToken token) => - this._linkedTokens.Remove(token); + public void UnlinkCancellation(CancellationToken cancellationToken) => + this._linkedTokens.Remove(cancellationToken); - public void LinkCancellation(CancellationToken token) + public void LinkCancellation(CancellationToken cancellationToken) { - this._linkedTokens.Add(token); + this._linkedTokens.Add(cancellationToken); CancellationTokenSource tokenSource = CancellationTokenSource.CreateLinkedTokenSource(this._linkedTokens.ToArray()); tokenSource = Interlocked.Exchange(ref this._internalCts, tokenSource); tokenSource.Dispose();