diff --git a/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs b/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs index 17ba3602f8..600182974c 100644 --- a/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs @@ -49,7 +49,7 @@ public static class Program .Build(); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Create a slogan for a new electric SUV that is affordable and fun to drive."); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Create a slogan for a new electric SUV that is affordable and fun to drive."); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { if (evt is SloganGeneratedEvent or FeedbackEvent) diff --git a/dotnet/samples/GettingStarted/Workflows/Agents/FoundryAgent/Program.cs b/dotnet/samples/GettingStarted/Workflows/Agents/FoundryAgent/Program.cs index 3eadeea732..9a9005e79d 100644 --- a/dotnet/samples/GettingStarted/Workflows/Agents/FoundryAgent/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Agents/FoundryAgent/Program.cs @@ -38,7 +38,7 @@ public static class Program .Build(); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!")); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!")); // Must send the turn token to trigger the agents. // The agents are wrapped as executors. When they receive messages, // they will cache the messages and only start processing when they receive a TurnToken. diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs index 98bf11f172..e36d59f6d6 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs @@ -32,7 +32,7 @@ public static class Program var checkpoints = new List(); // Execute the workflow and save checkpoints - Checkpointed checkpointedRun = await InProcessExecution + await using Checkpointed checkpointedRun = await InProcessExecution .StreamAsync(workflow, NumberSignal.Init, checkpointManager) .ConfigureAwait(false); await foreach (WorkflowEvent evt in checkpointedRun.Run.WatchStreamAsync().ConfigureAwait(false)) @@ -72,7 +72,7 @@ public static class Program Console.WriteLine($"\n\nHydrating a new workflow instance from the {CheckpointIndex + 1}th checkpoint."); CheckpointInfo savedCheckpoint = checkpoints[CheckpointIndex]; - Checkpointed newCheckpointedRun = + await using Checkpointed newCheckpointedRun = await InProcessExecution.ResumeStreamAsync(newWorkflow, savedCheckpoint, checkpointManager, checkpointedRun.Run.RunId) .ConfigureAwait(false); diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs index 376d669671..f7fb178c8b 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs @@ -31,7 +31,7 @@ public static class Program var checkpoints = new List(); // Execute the workflow and save checkpoints - Checkpointed checkpointedRun = await InProcessExecution + await using Checkpointed checkpointedRun = await InProcessExecution .StreamAsync(workflow, NumberSignal.Init, checkpointManager) .ConfigureAwait(false); await foreach (WorkflowEvent evt in checkpointedRun.Run.WatchStreamAsync().ConfigureAwait(false)) diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs index b38b5167bb..48903425a9 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs @@ -34,7 +34,7 @@ public static class Program var checkpoints = new List(); // Execute the workflow and save checkpoints - Checkpointed checkpointedRun = await InProcessExecution + await using Checkpointed checkpointedRun = await InProcessExecution .StreamAsync(workflow, new SignalWithNumber(NumberSignal.Init), checkpointManager) .ConfigureAwait(false); await foreach (WorkflowEvent evt in checkpointedRun.Run.WatchStreamAsync().ConfigureAwait(false)) diff --git a/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs b/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs index d332e4e5cc..d43474ece0 100644 --- a/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs @@ -59,7 +59,7 @@ public static class Program .Build(); // Execute the workflow in streaming mode - StreamingRun run = await InProcessExecution.StreamAsync(workflow, "What is temperature?"); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "What is temperature?"); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { if (evt is WorkflowOutputEvent output) diff --git a/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs b/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs index 791fb074a7..a0f059a00e 100644 --- a/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs @@ -99,7 +99,7 @@ public static class Program // Step 2: Run the workflow Console.WriteLine("\n=== RUNNING WORKFLOW ===\n"); - StreamingRun run = await InProcessExecution.StreamAsync(workflow, rawText); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, rawText); await foreach (WorkflowEvent evt in run.WatchStreamAsync()) { Console.WriteLine($"Event: {evt}"); diff --git a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/01_EdgeCondition/Program.cs b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/01_EdgeCondition/Program.cs index bb76360983..65ac2aa852 100644 --- a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/01_EdgeCondition/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/01_EdgeCondition/Program.cs @@ -62,7 +62,7 @@ public static class Program string email = Resources.Read("spam.txt"); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, email)); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, email)); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/02_SwitchCase/Program.cs b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/02_SwitchCase/Program.cs index 216088b74f..4f985f47a2 100644 --- a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/02_SwitchCase/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/02_SwitchCase/Program.cs @@ -78,7 +78,7 @@ public static class Program string email = Resources.Read("ambiguous_email.txt"); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, email)); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, email)); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs index a87962a32d..987dd7ffd6 100644 --- a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs @@ -86,7 +86,7 @@ public static class Program string email = Resources.Read("email.txt"); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, email)); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, email)); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs b/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs index 5308bb06b8..f9754ab22e 100644 --- a/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs @@ -64,7 +64,7 @@ internal sealed class Program InputResponse? response = null; do { - ExternalRequest? inputRequest = await this.MonitorWorkflowRunAsync(run, response); + ExternalRequest? inputRequest = await this.MonitorAndDisposeWorkflowRunAsync(run, response); if (inputRequest is not null) { Notify("\nWORKFLOW: Yield"); @@ -83,6 +83,7 @@ internal sealed class Program // Restore the latest checkpoint. Debug.WriteLine($"RESTORE #{this.LastCheckpoint.CheckpointId}"); Notify("\nWORKFLOW: Restore"); + run = await InProcessExecution.ResumeStreamAsync(workflow, this.LastCheckpoint, checkpointManager, run.Run.RunId); } else @@ -132,8 +133,10 @@ internal sealed class Program this.FoundryClient = new PersistentAgentsClient(this.FoundryEndpoint, new AzureCliCredential()); } - private async Task MonitorWorkflowRunAsync(Checkpointed run, InputResponse? response = null) + private async Task MonitorAndDisposeWorkflowRunAsync(Checkpointed run, InputResponse? response = null) { + await using IAsyncDisposable disposeRun = run; + string? messageId = null; await foreach (WorkflowEvent workflowEvent in run.Run.WatchStreamAsync().ConfigureAwait(false)) @@ -175,7 +178,7 @@ internal sealed class Program } else { - await run.Run.EndRunAsync().ConfigureAwait(false); + await run.Run.DisposeAsync().ConfigureAwait(false); return requestInfo.Request; } break; diff --git a/dotnet/samples/GettingStarted/Workflows/DeclarativeCode/Program.cs b/dotnet/samples/GettingStarted/Workflows/DeclarativeCode/Program.cs index f2bea10bd8..ca8032f808 100644 --- a/dotnet/samples/GettingStarted/Workflows/DeclarativeCode/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/DeclarativeCode/Program.cs @@ -45,7 +45,7 @@ internal sealed class Program // Run the workflow, just like any other workflow string input = this.GetWorkflowInput(); StreamingRun run = await InProcessExecution.StreamAsync(workflow, input); - await this.MonitorWorkflowRunAsync(run); + await this.MonitorAndDisposeWorkflowRunAsync(run); Notify("\nWORKFLOW: Done!"); } @@ -70,8 +70,10 @@ internal sealed class Program this.FoundryClient = new PersistentAgentsClient(this.FoundryEndpoint, new AzureCliCredential()); } - private async Task MonitorWorkflowRunAsync(StreamingRun run) + private async Task MonitorAndDisposeWorkflowRunAsync(StreamingRun run) { + await using IAsyncDisposable disposeRun = run; + string? messageId = null; await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) diff --git a/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs b/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs index 1da9fa37ae..e3628b11e5 100644 --- a/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs @@ -27,7 +27,7 @@ public static class Program var workflow = await WorkflowHelper.GetWorkflowAsync().ConfigureAwait(false); // Execute the workflow - StreamingRun handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); + await using StreamingRun handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false)) { switch (evt) diff --git a/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs b/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs index 69d66cc8e6..020c193ca0 100644 --- a/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs @@ -33,7 +33,7 @@ public static class Program .BuildAsync(); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { if (evt is WorkflowOutputEvent outputEvent) diff --git a/dotnet/samples/GettingStarted/Workflows/Observability/AspireDashboard/Program.cs b/dotnet/samples/GettingStarted/Workflows/Observability/AspireDashboard/Program.cs index f091e46683..c02d6bc4a0 100644 --- a/dotnet/samples/GettingStarted/Workflows/Observability/AspireDashboard/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Observability/AspireDashboard/Program.cs @@ -57,7 +57,7 @@ public static class Program .Build(); // Execute the workflow with input data - Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!"); + await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!"); foreach (WorkflowEvent evt in run.NewEvents) { if (evt is ExecutorCompletedEvent executorComplete) diff --git a/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs b/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs index 0516513690..8812d2689e 100644 --- a/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs @@ -33,7 +33,7 @@ public static class Program .Build(); // Execute the workflow with input data - Run run = await InProcessExecution.RunAsync(workflow, "Lorem_Ipsum.txt"); + await using Run run = await InProcessExecution.RunAsync(workflow, "Lorem_Ipsum.txt"); foreach (WorkflowEvent evt in run.NewEvents) { if (evt is WorkflowOutputEvent outputEvent) diff --git a/dotnet/samples/GettingStarted/Workflows/_Foundational/01_ExecutorsAndEdges/Program.cs b/dotnet/samples/GettingStarted/Workflows/_Foundational/01_ExecutorsAndEdges/Program.cs index b64e1bad6b..b3712eed11 100644 --- a/dotnet/samples/GettingStarted/Workflows/_Foundational/01_ExecutorsAndEdges/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/_Foundational/01_ExecutorsAndEdges/Program.cs @@ -30,7 +30,7 @@ public static class Program var workflow = builder.Build(); // Execute the workflow with input data - Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!"); + await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!"); foreach (WorkflowEvent evt in run.NewEvents) { if (evt is ExecutorCompletedEvent executorComplete) diff --git a/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs b/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs index 4544ac4212..c3c1370b84 100644 --- a/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs @@ -29,7 +29,7 @@ public static class Program var workflow = builder.Build(); // Execute the workflow in streaming mode - StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Hello, World!"); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Hello, World!"); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { if (evt is ExecutorCompletedEvent executorCompleted) diff --git a/dotnet/samples/GettingStarted/Workflows/_Foundational/03_AgentsInWorkflows/Program.cs b/dotnet/samples/GettingStarted/Workflows/_Foundational/03_AgentsInWorkflows/Program.cs index 2a3058c611..2b30f3f5a1 100644 --- a/dotnet/samples/GettingStarted/Workflows/_Foundational/03_AgentsInWorkflows/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/_Foundational/03_AgentsInWorkflows/Program.cs @@ -44,7 +44,7 @@ public static class Program .Build(); // Execute the workflow - StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!")); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!")); // Must send the turn token to trigger the agents. // The agents are wrapped as executors. When they receive messages, diff --git a/dotnet/samples/GettingStarted/Workflows/_Foundational/04_AgentWorkflowPatterns/Program.cs b/dotnet/samples/GettingStarted/Workflows/_Foundational/04_AgentWorkflowPatterns/Program.cs index 70a0ddc8a0..aeb06cfb64 100644 --- a/dotnet/samples/GettingStarted/Workflows/_Foundational/04_AgentWorkflowPatterns/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/_Foundational/04_AgentWorkflowPatterns/Program.cs @@ -84,7 +84,7 @@ public static class Program { string? lastExecutorId = null; - StreamingRun run = await InProcessExecution.StreamAsync(workflow, messages); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, messages); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs index b481113a2a..f61540c89c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointed.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -14,7 +15,7 @@ namespace Microsoft.Agents.AI.Workflows; /// The type of the underlying workflow run handle. /// /// -public class Checkpointed +public sealed class Checkpointed : IAsyncDisposable { private readonly ICheckpointingHandle _runner; @@ -46,6 +47,19 @@ public class Checkpointed } } + /// + public async ValueTask DisposeAsync() + { + if (this.Run is IAsyncDisposable asyncDisposable) + { + await asyncDisposable.DisposeAsync().ConfigureAwait(false); + } + else if (this.Run is IDisposable disposable) + { + disposable.Dispose(); + } + } + /// public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default) => this._runner.RestoreCheckpointAsync(checkpointInfo, cancellationToken); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs index a983126c44..8fe6edce2a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs @@ -11,13 +11,12 @@ using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.AI.Workflows.Execution; -internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, IInputCoordinator +internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable { - private readonly AsyncCoordinator _waitForResponseCoordinator = new(); private readonly ISuperStepRunner _stepRunner; private readonly ICheckpointingHandle _checkpointingHandle; - private readonly LockstepRunEventStream _eventStream; + private readonly IRunEventStream _eventStream; private readonly CancellationTokenSource _endRunSource = new(); private int _isDisposed; private int _isEventStreamTaken; @@ -29,17 +28,26 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I this._eventStream = mode switch { - //ExecutionMode.OffThread => Not supported yet + ExecutionMode.OffThread => new StreamingRunEventStream(stepRunner), + ExecutionMode.Subworkflow => new StreamingRunEventStream(stepRunner, disableRunLoop: true), ExecutionMode.Lockstep => new LockstepRunEventStream(stepRunner), _ => throw new ArgumentOutOfRangeException(nameof(mode), $"Unknown execution mode {mode}") }; + this._eventStream.Start(); + + // If there are already unprocessed messages (e.g., from a checkpoint restore that happened + // before this handle was created), signal the run loop to start processing them + if (stepRunner.HasUnprocessedMessages) + { + this.SignalInputToRunLoop(); + } } - public ValueTask WaitForNextInputAsync(CancellationToken cancellation = default) - => this._waitForResponseCoordinator.WaitForCoordinationAsync(cancellation); + //private readonly AsyncCoordinator _waitForResponseCoordinator = new(); - public void ReleaseResponseWaiter() => this._waitForResponseCoordinator.MarkCoordinationPoint(); + //public ValueTask WaitForNextInputAsync(CancellationToken cancellation = default) + // => this._waitForResponseCoordinator.WaitForCoordinationAsync(cancellation); public string RunId => this._stepRunner.RunId; @@ -48,28 +56,39 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I public ValueTask GetStatusAsync(CancellationToken cancellation = default) => this._eventStream.GetStatusAsync(cancellation); - public async IAsyncEnumerable TakeEventStreamAsync(bool breakOnHalt, [EnumeratorCancellation] CancellationToken cancellation = default) + public async IAsyncEnumerable TakeEventStreamAsync(bool blockOnPendingRequest, [EnumeratorCancellation] CancellationToken cancellation = default) { - // Create a linked cancellation token that combines the provided token with the end-run token - using var linkedSource = CancellationTokenSource.CreateLinkedTokenSource(cancellation, this._endRunSource.Token); - - // Only one enumerator of this is allowed at a time + //Debug.Assert(breakOnHalt); + // Enforce single active enumerator (this runs when enumeration begins) if (Interlocked.CompareExchange(ref this._isEventStreamTaken, 1, 0) != 0) { throw new InvalidOperationException("The event stream has already been taken. Only one enumerator is allowed at a time."); } + CancellationTokenSource? linked = null; try { - await foreach (WorkflowEvent @event in this._eventStream.TakeEventStreamAsync(linkedSource.Token) - .ConfigureAwait(false)) + linked = CancellationTokenSource.CreateLinkedTokenSource(cancellation, this._endRunSource.Token); + var token = linked.Token; + + // Build the inner stream before the loop so synchronous exceptions still release the gate + var inner = this._eventStream.TakeEventStreamAsync(blockOnPendingRequest, token); + + await foreach (var ev in inner.WithCancellation(token).ConfigureAwait(false)) { - yield return @event; + // Filter out the RequestHaltEvent, since it is an internal signalling event. + if (ev is RequestHaltEvent) + { + yield break; + } + + yield return ev; } } finally { - Volatile.Write(ref this._isEventStreamTaken, 0); + linked?.Dispose(); + Interlocked.Exchange(ref this._isEventStreamTaken, 0); } } @@ -80,7 +99,7 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I { if (message is ExternalResponse response) { - // EnqueueResponseAsync marks the coordination point itself + // EnqueueResponseAsync handles signaling await this.EnqueueResponseAsync(response, cancellation) .ConfigureAwait(false); @@ -90,7 +109,8 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I bool result = await this._stepRunner.EnqueueMessageAsync(message, cancellation) .ConfigureAwait(false); - this._waitForResponseCoordinator.MarkCoordinationPoint(); + // Signal the run loop that new input is available + this.SignalInputToRunLoop(); return result; } @@ -104,7 +124,7 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I if (declaredType != null && typeof(ExternalResponse).IsAssignableFrom(declaredType)) { - // EnqueueResponseAsync marks the coordination point itself + // EnqueueResponseAsync handles signaling await this.EnqueueResponseAsync((ExternalResponse)message, cancellation) .ConfigureAwait(false); @@ -112,7 +132,7 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I } else if (declaredType == null && message is ExternalResponse response) { - // EnqueueResponseAsync marks the coordination point itself + // EnqueueResponseAsync handles signaling await this.EnqueueResponseAsync(response, cancellation) .ConfigureAwait(false); @@ -122,7 +142,8 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I bool result = await this._stepRunner.EnqueueMessageUntypedAsync(message, declaredType ?? message.GetType(), cancellation) .ConfigureAwait(false); - this._waitForResponseCoordinator.MarkCoordinationPoint(); + // Signal the run loop that new input is available + this.SignalInputToRunLoop(); return result; } @@ -131,13 +152,13 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I { await this._stepRunner.EnqueueResponseAsync(response, cancellation).ConfigureAwait(false); - this._waitForResponseCoordinator.MarkCoordinationPoint(); + // Signal the run loop that new input is available + this.SignalInputToRunLoop(); } - public ValueTask RequestEndRunAsync() + private void SignalInputToRunLoop() { - this._endRunSource.Cancel(); - return this._stepRunner.RequestEndRunAsync(); + this._eventStream.SignalInput(); } public async ValueTask DisposeAsync() @@ -145,13 +166,32 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable, I if (Interlocked.Exchange(ref this._isDisposed, 1) == 0) { this._endRunSource.Cancel(); - await this.RequestEndRunAsync().ConfigureAwait(false); + + await this._eventStream.StopAsync().ConfigureAwait(false); + await this._stepRunner.RequestEndRunAsync().ConfigureAwait(false); + this._endRunSource.Dispose(); await this._eventStream.DisposeAsync().ConfigureAwait(false); } } - public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default) - => this._checkpointingHandle.RestoreCheckpointAsync(checkpointInfo, cancellationToken); + public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default) + { + // Clear buffered events from the channel BEFORE restoring to discard stale events from supersteps + // that occurred after the checkpoint we're restoring to + // This must happen BEFORE the restore so that events republished during restore aren't cleared + if (this._eventStream is StreamingRunEventStream streamingEventStream) + { + streamingEventStream.ClearBufferedEvents(); + } + + // Restore the workflow state - this will republish unserviced requests as new events + await this._checkpointingHandle.RestoreCheckpointAsync(checkpointInfo, cancellationToken).ConfigureAwait(false); + + // After restore, signal the run loop to process any restored messages + // This is necessary because ClearBufferedEvents() doesn't signal, and the restored + // queued messages won't automatically wake up the run loop + this.SignalInputToRunLoop(); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/IRunEventStream.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/IRunEventStream.cs index 6b5a82868c..5ffce6ce36 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/IRunEventStream.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/IRunEventStream.cs @@ -10,6 +10,12 @@ namespace Microsoft.Agents.AI.Workflows.Execution; internal interface IRunEventStream : IAsyncDisposable { void Start(); + void SignalInput(); + + // this cannot be cancelled + ValueTask StopAsync(); + ValueTask GetStatusAsync(CancellationToken cancellation = default); - IAsyncEnumerable TakeEventStreamAsync(CancellationToken cancellation = default); + + IAsyncEnumerable TakeEventStreamAsync(bool blockOnPendingRequest, CancellationToken cancellation = default); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/InputWaiter.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/InputWaiter.cs new file mode 100644 index 0000000000..f86f3721bb --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/InputWaiter.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows.Execution; + +internal sealed class InputWaiter : IDisposable +{ + private readonly SemaphoreSlim _inputSignal = new(initialCount: 0, 1); + + public void Dispose() + { + this._inputSignal.Dispose(); + } + + /// + /// Signals that new input has been provided and the waiter should continue processing. + /// Called by AsyncRunHandle when the user enqueues a message or response. + /// + public void SignalInput() + { + // Release the run loop to process more work + // Only release if not already signaled (binary semaphore behavior) + try + { + this._inputSignal.Release(); + } + catch (SemaphoreFullException) + { + // Swallow for now + } + } + + public Task WaitForInputAsync(CancellationToken cancellation = default) => this.WaitForInputAsync(null, cancellation); + + public async Task WaitForInputAsync(TimeSpan? timeout = null, CancellationToken cancellation = default) + { + await this._inputSignal.WaitAsync(timeout ?? TimeSpan.FromMilliseconds(-1), cancellation).ConfigureAwait(false); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs index baec8236db..a66f0af978 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs @@ -16,29 +16,45 @@ internal sealed class LockstepRunEventStream : IRunEventStream private static readonly string s_namespace = typeof(LockstepRunEventStream).Namespace!; private static readonly ActivitySource s_activitySource = new(s_namespace); + private readonly CancellationTokenSource _stopCancellation = new(); + private readonly InputWaiter _inputWaiter = new(); + private int _isDisposed; + + private readonly ISuperStepRunner _stepRunner; + public ValueTask GetStatusAsync(CancellationToken cancellation = default) => new(this.RunStatus); public LockstepRunEventStream(ISuperStepRunner stepRunner) { - this.StepRunner = stepRunner; + this._stepRunner = stepRunner; } private RunStatus RunStatus { get; set; } = RunStatus.NotStarted; - private ISuperStepRunner StepRunner { get; } public void Start() { // No-op for lockstep execution } - public async IAsyncEnumerable TakeEventStreamAsync([EnumeratorCancellation] CancellationToken cancellation = default) + public async IAsyncEnumerable TakeEventStreamAsync(bool blockOnPendingRequest, [EnumeratorCancellation] CancellationToken cancellation = default) { +#if NET + ObjectDisposedException.ThrowIf(Volatile.Read(ref this._isDisposed) == 1, this); +#else + if (Volatile.Read(ref this._isDisposed) == 1) + { + throw new ObjectDisposedException(nameof(LockstepRunEventStream)); + } +#endif + + CancellationTokenSource linkedSource = CancellationTokenSource.CreateLinkedTokenSource(this._stopCancellation.Token, cancellation); + ConcurrentQueue eventSink = []; - this.StepRunner.OutgoingEvents.EventRaised += OnWorkflowEventAsync; + this._stepRunner.OutgoingEvents.EventRaised += OnWorkflowEventAsync; using Activity? activity = s_activitySource.StartActivity(ActivityNames.WorkflowRun); - activity?.SetTag(Tags.WorkflowId, this.StepRunner.StartExecutorId).SetTag(Tags.RunId, this.StepRunner.RunId); + activity?.SetTag(Tags.WorkflowId, this._stepRunner.StartExecutorId).SetTag(Tags.RunId, this._stepRunner.RunId); try { @@ -47,63 +63,81 @@ internal sealed class LockstepRunEventStream : IRunEventStream do { - // Because we may be yielding out of this function, we need to ensure that the Activity.Current - // is set to our activity for the duration of this loop iteration. - Activity.Current = activity; + while (this._stepRunner.HasUnprocessedMessages && + !linkedSource.Token.IsCancellationRequested) + { + // Because we may be yielding out of this function, we need to ensure that the Activity.Current + // is set to our activity for the duration of this loop iteration. + Activity.Current = activity; - // Drain SuperSteps while there are steps to run - try - { - await this.StepRunner.RunSuperStepAsync(cancellation).ConfigureAwait(false); - } - catch (Exception ex) when (activity is not null) - { - activity.AddEvent(new ActivityEvent(EventNames.WorkflowError, tags: new() { - { Tags.ErrorType, ex.GetType().FullName }, - { Tags.BuildErrorMessage, ex.Message }, - })); - activity.CaptureException(ex); - throw; - } + // Drain SuperSteps while there are steps to run + try + { + await this._stepRunner.RunSuperStepAsync(linkedSource.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + } + catch (Exception ex) when (activity is not null) + { + activity.AddEvent(new ActivityEvent(EventNames.WorkflowError, tags: new() { + { Tags.ErrorType, ex.GetType().FullName }, + { Tags.BuildErrorMessage, ex.Message }, + })); + activity.CaptureException(ex); + throw; + } - if (cancellation.IsCancellationRequested) - { - yield break; // Exit if cancellation is requested - } - - bool hadRequestHaltEvent = false; - foreach (WorkflowEvent raisedEvent in Interlocked.Exchange(ref eventSink, [])) - { - if (cancellation.IsCancellationRequested) + if (linkedSource.Token.IsCancellationRequested) { yield break; // Exit if cancellation is requested } - // TODO: Do we actually want to interpret this as a termination request? - if (raisedEvent is RequestHaltEvent) + bool hadRequestHaltEvent = false; + foreach (WorkflowEvent raisedEvent in Interlocked.Exchange(ref eventSink, [])) { - hadRequestHaltEvent = true; + if (linkedSource.Token.IsCancellationRequested) + { + yield break; // Exit if cancellation is requested + } + + // TODO: Do we actually want to interpret this as a termination request? + if (raisedEvent is RequestHaltEvent) + { + hadRequestHaltEvent = true; + } + else + { + yield return raisedEvent; + } } - else + + if (hadRequestHaltEvent || linkedSource.Token.IsCancellationRequested) { - yield return raisedEvent; + // If we had a completion event, we are done. + yield break; } + + this.RunStatus = this._stepRunner.HasUnservicedRequests ? RunStatus.PendingRequests : RunStatus.Idle; } - if (hadRequestHaltEvent) + if (blockOnPendingRequest && this.RunStatus == RunStatus.PendingRequests) { - // If we had a completion event, we are done. - yield break; + try + { + await this._inputWaiter.WaitForInputAsync(TimeSpan.FromSeconds(1), linkedSource.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { } } - } while (this.StepRunner.HasUnprocessedMessages && - !cancellation.IsCancellationRequested); + } while (!ShouldBreak()); activity?.AddEvent(new ActivityEvent(EventNames.WorkflowCompleted)); } finally { - this.RunStatus = this.StepRunner.HasUnservicedRequests ? RunStatus.PendingRequests : RunStatus.Idle; - this.StepRunner.OutgoingEvents.EventRaised -= OnWorkflowEventAsync; + this.RunStatus = this._stepRunner.HasUnservicedRequests ? RunStatus.PendingRequests : RunStatus.Idle; + this._stepRunner.OutgoingEvents.EventRaised -= OnWorkflowEventAsync; } ValueTask OnWorkflowEventAsync(object? sender, WorkflowEvent e) @@ -111,7 +145,40 @@ internal sealed class LockstepRunEventStream : IRunEventStream eventSink.Enqueue(e); return default; } + + // If we are Idle or Ended, we should break out of the loop + // If we are PendingRequests and not blocking on pending requests, we should break out of the loop + // If cancellation is requested, we should break out of the loop + bool ShouldBreak() => this.RunStatus is RunStatus.Idle or RunStatus.Ended || + (this.RunStatus == RunStatus.PendingRequests && !blockOnPendingRequest) || + linkedSource.Token.IsCancellationRequested; } - public ValueTask DisposeAsync() => default; + /// + /// Signals that new input has been provided and the run loop should continue processing. + /// Called by AsyncRunHandle when the user enqueues a message or response. + /// + public void SignalInput() + { + this._inputWaiter?.SignalInput(); + } + + public ValueTask StopAsync() + { + this._stopCancellation.Cancel(); + return default; + } + + public ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref this._isDisposed, 1) == 0) + { + this._stopCancellation.Cancel(); + + this._stopCancellation.Dispose(); + this._inputWaiter.Dispose(); + } + + return default; + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StepContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StepContext.cs index 30a92c6f9c..db03e5b7c7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StepContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StepContext.cs @@ -11,7 +11,7 @@ internal sealed class StepContext { public ConcurrentDictionary> QueuedMessages { get; } = []; - public bool HasMessages => this.QueuedMessages.Values.Any(messageQueue => !messageQueue.IsEmpty); + public bool HasMessages => !this.QueuedMessages.IsEmpty && this.QueuedMessages.Values.Any(messageQueue => !messageQueue.IsEmpty); public ConcurrentQueue MessagesFor(string target) { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs new file mode 100644 index 0000000000..0f941ea733 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs @@ -0,0 +1,262 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows.Execution; + +/// +/// A modern implementation of IRunEventStream that streams events as they are created, +/// using System.Threading.Channels for thread-safe coordination. +/// +internal sealed class StreamingRunEventStream : IRunEventStream +{ + private readonly Channel _eventChannel; + private readonly ISuperStepRunner _stepRunner; + private readonly InputWaiter _inputWaiter; + private readonly CancellationTokenSource _runLoopCancellation; + private readonly bool _disableRunLoop; + private Task? _runLoopTask; + private RunStatus _runStatus = RunStatus.NotStarted; + private int _completionEpoch; // Tracks which completion signal belongs to which consumer iteration + + public StreamingRunEventStream(ISuperStepRunner stepRunner, bool disableRunLoop = false) + { + this._stepRunner = stepRunner; + this._runLoopCancellation = new CancellationTokenSource(); + this._inputWaiter = new(); + this._disableRunLoop = disableRunLoop; + + // Unbounded channel - events never block the producer + // This allows events to flow freely during superstep execution + this._eventChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = true, // Only one consumer at a time (enforced by AsyncRunHandle) + SingleWriter = false, // Events can come from multiple threads during superstep execution + AllowSynchronousContinuations = false // Prevent potential deadlocks + }); + } + + public void Start() + { + // Start the background run loop that drives superstep execution + if (!this._disableRunLoop) + { + this._runLoopTask = Task.Run(() => this.RunLoopAsync(this._runLoopCancellation.Token)); + } + } + + private async Task RunLoopAsync(CancellationToken cancellation) + { + using CancellationTokenSource errorSource = new(); + CancellationTokenSource linkedSource = CancellationTokenSource.CreateLinkedTokenSource(errorSource.Token, cancellation); + + // Subscribe to events - they will flow directly to the channel as they're raised + this._stepRunner.OutgoingEvents.EventRaised += OnEventRaisedAsync; + + try + { + // Wait for the first input before starting + // The consumer will call EnqueueMessageAsync which signals the run loop + await this._inputWaiter.WaitForInputAsync(cancellation: linkedSource.Token).ConfigureAwait(false); + + this._runStatus = RunStatus.Running; + + while (!linkedSource.Token.IsCancellationRequested) + { + // Run all available supersteps continuously + // Events are streamed out in real-time as they happen via the event handler + while (this._stepRunner.HasUnprocessedMessages && !linkedSource.Token.IsCancellationRequested) + { + await this._stepRunner.RunSuperStepAsync(linkedSource.Token).ConfigureAwait(false); + } + + // Update status based on what's waiting + this._runStatus = this._stepRunner.HasUnservicedRequests + ? RunStatus.PendingRequests + : RunStatus.Idle; + + // Signal completion to consumer so they can check status and decide whether to continue + // Increment epoch so next consumer iteration gets a new completion signal + // Capture the status at this moment to avoid race conditions with event reading + int currentEpoch = Interlocked.Increment(ref this._completionEpoch); + RunStatus capturedStatus = this._runStatus; + await this._eventChannel.Writer.WriteAsync(new InternalHaltSignal(currentEpoch, capturedStatus), linkedSource.Token).ConfigureAwait(false); + + // Wait for next input from the consumer + // Works for both Idle (no work) and PendingRequests (waiting for responses) + await this._inputWaiter.WaitForInputAsync(TimeSpan.FromSeconds(1), linkedSource.Token).ConfigureAwait(false); + + // When signaled, resume running + this._runStatus = RunStatus.Running; + } + } + catch (OperationCanceledException) + { + // Expected during shutdown + } + catch (Exception e) + { + await this._eventChannel.Writer.WriteAsync(new WorkflowErrorEvent(e), linkedSource.Token).ConfigureAwait(false); + } + finally + { + this._stepRunner.OutgoingEvents.EventRaised -= OnEventRaisedAsync; + this._eventChannel.Writer.Complete(); + + // Mark as ended when run loop exits + this._runStatus = RunStatus.Ended; + } + + async ValueTask OnEventRaisedAsync(object? sender, WorkflowEvent e) + { + // Write event directly to channel - it's thread-safe and non-blocking + // The channel handles all synchronization internally using lock-free algorithms + // Events flow immediately to consumers rather than being batched + await this._eventChannel.Writer.WriteAsync(e, linkedSource.Token).ConfigureAwait(false); + + if (e is WorkflowErrorEvent error) + { + errorSource.Cancel(); + } + } + } + + /// + /// Signals that new input has been provided and the run loop should continue processing. + /// Called by AsyncRunHandle when the user enqueues a message or response. + /// + public void SignalInput() => this._inputWaiter.SignalInput(); + + public async IAsyncEnumerable TakeEventStreamAsync( + bool blockOnPendingRequest, + [EnumeratorCancellation] CancellationToken cancellation = default) + { + // Get the current epoch - we'll only respond to completion signals from this epoch or later + int myEpoch = Volatile.Read(ref this._completionEpoch) + 1; + + // Simply read from channel - all coordination is handled by Channel infrastructure + // Note: When cancellation is requested, ReadAllAsync may throw OperationCanceledException + // or may complete the enumeration. We check IsCancellationRequested explicitly at superstep + // boundaries to ensure clean cancellation. + await foreach (WorkflowEvent evt in this._eventChannel.Reader.ReadAllAsync(cancellation).ConfigureAwait(false)) + { + // Filter out internal signals used for run loop coordination + if (evt is InternalHaltSignal completionSignal) + { + // Ignore completion signals from previous iterations + if (completionSignal.Epoch < myEpoch) + { + continue; + } + + // Check for cancellation at superstep boundaries (before processing completion signal) + // This allows consumers to stop reading events cleanly between supersteps + if (cancellation.IsCancellationRequested) + { + yield break; + } + + // Check if we should stop streaming based on the status captured at completion time + // This avoids race conditions where _runStatus changes while events are being read + // - Idle: Workflow completed, no pending requests + // - Ended: Run loop disposed/cancelled + // Note: PendingRequests is handled by WatchStreamAsync's do-while loop + if (completionSignal.Status is RunStatus.Idle or RunStatus.Ended) + { + yield break; + } + + if (!blockOnPendingRequest && completionSignal.Status is RunStatus.PendingRequests) + { + yield break; + } + + // Otherwise continue reading (more events coming after input provided) + continue; + } + + // RequestHaltEvent signals the end of the event stream + if (evt is RequestHaltEvent) + { + yield break; + } + + if (cancellation.IsCancellationRequested) + { + yield break; + } + + yield return evt; + } + } + + public ValueTask GetStatusAsync(CancellationToken cancellation = default) + { + // Thread-safe read of status (enum is read atomically on most platforms) + return new ValueTask(this._runStatus); + } + + /// + /// Clears all buffered events from the channel. + /// This should be called when restoring a checkpoint to discard stale events from superseded supersteps. + /// + public void ClearBufferedEvents() + { + // Drain all events currently in the channel buffer + // We discard all events since they're from a timeline that's been superseded by the checkpoint restore + while (this._eventChannel.Reader.TryRead(out _)) + { + // Discard each event (including InternalCompletionSignals) + } + + // After clearing, signal the run loop to continue if needed + // The run loop will send a new completion signal when it finishes processing from the restored state + this.SignalInput(); + } + + public async ValueTask StopAsync() + { + // Cancel the run loop + this._runLoopCancellation.Cancel(); + + // Release the event waiter, if any + this._inputWaiter.SignalInput(); + + // Wait for clean shutdown + if (this._runLoopTask != null) + { + try + { + await this._runLoopTask.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Expected during cancellation + } + } + } + + public async ValueTask DisposeAsync() + { + await this.StopAsync().ConfigureAwait(false); + + // Dispose resources + this._runLoopCancellation.Dispose(); + this._inputWaiter.Dispose(); + } + + /// + /// Internal signal used to mark completion of a work batch and allow status checking. + /// This is never exposed to consumers. + /// + private sealed class InternalHaltSignal(int epoch, RunStatus status) : WorkflowEvent + { + public int Epoch => epoch; + public RunStatus Status => status; + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutionMode.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutionMode.cs index 0cfecc6773..15c1fca7cb 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutionMode.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutionMode.cs @@ -4,6 +4,21 @@ namespace Microsoft.Agents.AI.Workflows; internal enum ExecutionMode { + /// + /// Normal streaming mode using the new channel-based implementation. + /// Events stream out immediately as they are created. + /// OffThread, - Lockstep + + /// + /// Lockstep mode where events are batched per superstep. + /// Events are accumulated and emitted after each superstep completes. + /// + Lockstep, + + /// + /// A special execution mode for subworkflows - it functions like OffThread, but without the internal task + /// running super steps, as they are implemented by being driven directly by the hosting workflow + /// + Subworkflow, } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs new file mode 100644 index 0000000000..cab1196e84 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs @@ -0,0 +1,188 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Defines an execution environment for running, streaming, and resuming workflows asynchronously, with optional +/// checkpointing and run management capabilities. +/// +public interface IWorkflowExecutionEnvironment +{ + /// + /// Initiates an asynchronous streaming execution using the specified input. + /// + /// The returned provides methods to observe and control + /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or + /// cancelled. + /// A 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// Initiates an asynchronous streaming execution using the specified input. + /// + /// The returned provides methods to observe and control + /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or + /// cancelled. + /// A 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. + /// + /// 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. + /// The to monitor for cancellationToken requests. The default is . + /// A that provides access to the results of the streaming run. + ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default); + + /// + /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. + /// + /// 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. + /// The to monitor for cancellationToken requests. The default is . + /// A that provides access to the results of the streaming run. + ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// Initiates a non-streaming execution of the workflow with the specified input. + /// + /// 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// Initiates a non-streaming execution of the workflow with the specified input. + /// + /// 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// 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. + /// An optional unique identifier for the run. If not provided, a new identifier will be generated. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + + /// + /// 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 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. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default); + + /// + /// 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 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. + /// The to monitor for cancellationToken requests. The default is . + /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. + ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs new file mode 100644 index 0000000000..67c35099bc --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs @@ -0,0 +1,207 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Checkpointing; +using Microsoft.Agents.AI.Workflows.Execution; + +namespace Microsoft.Agents.AI.Workflows.InProc; + +/// +/// Provides an in-process implementation of the workflow execution environment for running, streaming, and +/// checkpointing workflows within the current application domain. +/// +public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironment +{ + private readonly ExecutionMode _executionMode; + internal InProcessExecutionEnvironment(ExecutionMode mode) + { + this._executionMode = mode; + } + + internal ValueTask BeginRunAsync(Workflow workflow, ICheckpointManager? checkpointManager, string? runId, IEnumerable knownValidInputTypes, CancellationToken cancellationToken) + { + InProcessRunner runner = new(workflow, checkpointManager, runId, knownValidInputTypes: knownValidInputTypes); + return runner.BeginStreamAsync(this._executionMode, cancellationToken); + } + + internal ValueTask ResumeRunAsync(Workflow workflow, ICheckpointManager? checkpointManager, string? runId, CheckpointInfo fromCheckpoint, IEnumerable knownValidInputTypes, CancellationToken cancellationToken) + { + InProcessRunner runner = new(workflow, checkpointManager, runId, knownValidInputTypes: knownValidInputTypes); + return runner.ResumeStreamAsync(this._executionMode, fromCheckpoint, cancellationToken); + } + + /// + public async ValueTask StreamAsync( + Workflow workflow, + TInput input, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); + } + + /// + public async ValueTask StreamAsync( + Workflow workflow, + TInput input, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); + } + + /// + public async ValueTask> StreamAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndStreamAsync(input, cancellationToken)) + .ConfigureAwait(false); + } + + /// + public async ValueTask> StreamAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndStreamAsync(input, cancellationToken)) + .ConfigureAwait(false); + } + + /// + public async ValueTask> ResumeStreamAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) + { + AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => new(new StreamingRun(runHandle))) + .ConfigureAwait(false); + } + + /// + public async ValueTask> ResumeStreamAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => new(new StreamingRun(runHandle))) + .ConfigureAwait(false); + } + + /// + public async ValueTask RunAsync( + Workflow workflow, + TInput input, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.EnqueueAndRunAsync(input, cancellationToken).ConfigureAwait(false); + } + + /// + public async ValueTask RunAsync( + Workflow workflow, + TInput input, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.EnqueueAndRunAsync(input, cancellationToken).ConfigureAwait(false); + } + + /// + public async ValueTask> RunAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndRunAsync(input, cancellationToken)) + .ConfigureAwait(false); + } + + /// + public async ValueTask> RunAsync( + Workflow workflow, + TInput input, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndRunAsync(input, cancellationToken)) + .ConfigureAwait(false); + } + + /// + public async ValueTask> ResumeAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) + { + AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) + .ConfigureAwait(false); + } + + /// + public async ValueTask> ResumeAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + CheckpointManager checkpointManager, + string? runId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken) + .ConfigureAwait(false); + + return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) + .ConfigureAwait(false); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs index 2a0c74327b..6f4cef55c5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs @@ -148,7 +148,17 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle this.RunContext.HasQueuedExternalDeliveries || this.RunContext.JoinedRunnersHaveActions) { - await this.RunSuperstepAsync(currentStep).ConfigureAwait(false); + try + { + await this.RunSuperstepAsync(currentStep).ConfigureAwait(false); + } + catch (OperationCanceledException) + { } + catch (Exception e) + { + await this.RaiseWorkflowEventAsync(new WorkflowErrorEvent(e)).ConfigureAwait(false); + } + return true; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index f5808dd793..cd3f763ae8 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -367,6 +367,19 @@ internal sealed class InProcessRunnerContext : IRunnerContext { if (Interlocked.Exchange(ref this._runEnded, 1) == 0) { + foreach (string executorId in this._executors.Keys) + { + Task executor = this._executors[executorId]; + if (executor is IAsyncDisposable asyncDisposable) + { + await asyncDisposable.DisposeAsync().ConfigureAwait(false); + } + else if (executor is IDisposable disposable) + { + disposable.Dispose(); + } + } + await this._workflow.ReleaseOwnershipAsync(this).ConfigureAwait(false); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs index 99ab74b64b..831d4f5b29 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs @@ -1,11 +1,7 @@ // Copyright (c) Microsoft. All rights reserved. -using System; -using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; -using Microsoft.Agents.AI.Workflows.Checkpointing; -using Microsoft.Agents.AI.Workflows.Execution; using Microsoft.Agents.AI.Workflows.InProc; namespace Microsoft.Agents.AI.Workflows; @@ -14,338 +10,77 @@ namespace Microsoft.Agents.AI.Workflows; /// Provides methods to initiate and manage in-process workflow executions, supporting both streaming and /// non-streaming modes with asynchronous operations. /// -public sealed class InProcessExecution +public static class InProcessExecution { - private static InProcessExecution DefaultInstance { get; } = new(ExecutionMode.Lockstep); - - private readonly ExecutionMode _executionMode; - private InProcessExecution(ExecutionMode mode) - { - this._executionMode = mode; - } - - internal static ExecutionMode DefaultMode => DefaultInstance._executionMode; - - internal ValueTask BeginRunAsync(Workflow workflow, ICheckpointManager? checkpointManager, string? runId, IEnumerable knownValidInputTypes, CancellationToken cancellationToken) - { - InProcessRunner runner = new(workflow, checkpointManager, runId, knownValidInputTypes: knownValidInputTypes); - return runner.BeginStreamAsync(this._executionMode, cancellationToken); - } - - internal ValueTask ResumeRunAsync(Workflow workflow, ICheckpointManager? checkpointManager, string? runId, CheckpointInfo fromCheckpoint, IEnumerable knownValidInputTypes, CancellationToken cancellationToken) - { - InProcessRunner runner = new(workflow, checkpointManager, runId, knownValidInputTypes: knownValidInputTypes); - return runner.ResumeStreamAsync(this._executionMode, fromCheckpoint, cancellationToken); - } + /// + /// The default InProcess execution environment. + /// + public static InProcessExecutionEnvironment Default => OffThread; /// - /// Initiates an asynchronous streaming execution using the specified input. + /// An InProcessExecution environment which will run SuperSteps in a background thread, streaming + /// events out as they are raised. /// - /// The returned provides methods to observe and control - /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or - /// cancelled. - /// A 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); - } + public static InProcessExecutionEnvironment OffThread { get; } = new(ExecutionMode.OffThread); /// - /// Initiates an asynchronous streaming execution using the specified input. + /// An InProcesExecution environment which will run SuperSteps in the event watching thread, + /// accumulating events during each SuperStep and streaming them out after each SuperStep is + /// completed. /// - /// The returned provides methods to observe and control - /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or - /// cancelled. - /// A 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); - } + public static InProcessExecutionEnvironment Lockstep { get; } = new(ExecutionMode.Lockstep); /// - /// Initiates an asynchronous streaming execution using the specified input, with checkpointing. + /// An InProcessExecution environment which will not run SuperSteps directly, relying instead + /// on the hosting workflow to run them directly, while streaming events out as they are raised. /// - /// 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken) - .ConfigureAwait(false); + internal static InProcessExecutionEnvironment Subworkflow { get; } = new(ExecutionMode.Subworkflow); - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndStreamAsync(input, cancellationToken)) - .ConfigureAwait(false); - } + /// + public static ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.StreamAsync(workflow, input, runId, cancellationToken); - /// - /// 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); + /// + public static ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.StreamAsync(workflow, input, runId, cancellationToken); - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndStreamAsync(input, cancellationToken)) - .ConfigureAwait(false); - } + /// + public static ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.StreamAsync(workflow, input, checkpointManager, runId, 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 - /// 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. - /// The to monitor for cancellationToken requests. The default is . - /// A that provides access to the results of the streaming run. - public static async ValueTask> ResumeStreamAsync( - Workflow workflow, - CheckpointInfo fromCheckpoint, - CheckpointManager checkpointManager, - string? runId = null, - CancellationToken cancellationToken = default) - { - AsyncRunHandle runHandle = await DefaultInstance.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [], cancellationToken) - .ConfigureAwait(false); + /// + public static ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.StreamAsync(workflow, input, checkpointManager, runId, cancellationToken); - return await runHandle.WithCheckpointingAsync(() => new(new StreamingRun(runHandle))) - .ConfigureAwait(false); - } + /// + public static ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) + => Default.ResumeStreamAsync(workflow, fromCheckpoint, checkpointManager, runId, 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 - /// 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. - /// The to monitor for cancellationToken requests. The default is . - /// A that provides access to the results of the streaming run. - public static async ValueTask> ResumeStreamAsync( - Workflow workflow, - CheckpointInfo fromCheckpoint, - CheckpointManager checkpointManager, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); + /// + public static ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.ResumeStreamAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); - return await runHandle.WithCheckpointingAsync(() => new(new StreamingRun(runHandle))) - .ConfigureAwait(false); - } + /// + public static ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.RunAsync(workflow, input, runId, cancellationToken); - /// - /// Initiates a non-streaming execution of the workflow with the specified input. - /// - /// 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken) - .ConfigureAwait(false); + /// + public static ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.RunAsync(workflow, input, runId, cancellationToken); - return await runHandle.EnqueueAndRunAsync(input, cancellationToken).ConfigureAwait(false); - } + /// + public static ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.RunAsync(workflow, input, checkpointManager, runId, cancellationToken); - /// - /// Initiates a non-streaming execution of the workflow with the specified input. - /// - /// 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); + /// + public static ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.RunAsync(workflow, input, checkpointManager, runId, cancellationToken); - return await runHandle.EnqueueAndRunAsync(input, cancellationToken).ConfigureAwait(false); - } + /// + public static ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) + => Default.ResumeAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); - /// - /// 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndRunAsync(input, cancellationToken)) - .ConfigureAwait(false); - } - - /// - /// 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. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndRunAsync(input, cancellationToken)) - .ConfigureAwait(false); - } - - /// - /// 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 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. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) - { - AsyncRunHandle runHandle = await DefaultInstance.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) - .ConfigureAwait(false); - } - - /// - /// 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 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. - /// The to monitor for cancellationToken requests. The default is . - /// 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, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await DefaultInstance.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) - .ConfigureAwait(false); - } + /// + public static ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull + => Default.ResumeAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs index 0c83b4a970..dec02fbd46 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Run.cs @@ -1,6 +1,8 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -8,44 +10,11 @@ using Microsoft.Agents.AI.Workflows.Execution; namespace Microsoft.Agents.AI.Workflows; -/// -/// Specifies the current operational state of a workflow run. -/// -public enum RunStatus -{ - /// - /// The run has not yet started. This only occurs when running in "lockstep" mode. - /// - NotStarted, - - /// - /// The run has halted, has no outstanding requets, but has not received a . - /// - Idle, - - /// - /// The run has halted, and has at least one outstanding . - /// - PendingRequests, - - /// - /// The user has ended the run. No further events will be emitted, and no messages can be sent to it. - /// - /// - /// - Ended, - - /// - /// The workflow is currently running, and may receive events or requests. - /// - Running -} - /// /// Represents a workflow run that tracks execution status and emitted workflow events, supporting resumption /// with responses to . /// -public sealed class Run +public sealed class Run : IAsyncDisposable { private readonly List _eventSink = []; private readonly AsyncRunHandle _runHandle; @@ -57,7 +26,7 @@ public sealed class Run internal async ValueTask RunToNextHaltAsync(CancellationToken cancellationToken = default) { bool hadEvents = false; - await foreach (WorkflowEvent evt in this._runHandle.TakeEventStreamAsync(breakOnHalt: true, cancellationToken).ConfigureAwait(false)) + await foreach (WorkflowEvent evt in this._runHandle.TakeEventStreamAsync(blockOnPendingRequest: false, cancellationToken).ConfigureAwait(false)) { hadEvents = true; this._eventSink.Add(evt); @@ -84,9 +53,15 @@ public sealed class Run private int _lastBookmark; + /// + /// The number of events emitted by the workflow since the last access to + /// + public int NewEventCount => this._eventSink.Count - this._lastBookmark; + /// /// Gets all events emitted by the workflow since the last access to . /// + [DebuggerDisplay("NewEvents[{NewEventCount}]")] public IEnumerable NewEvents { get @@ -152,6 +127,9 @@ public sealed class Run return await this.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); } - /// - public ValueTask EndRunAsync() => this._runHandle.RequestEndRunAsync(); + /// + public ValueTask DisposeAsync() + { + return this._runHandle.DisposeAsync(); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/RunStatus.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/RunStatus.cs new file mode 100644 index 0000000000..ee2dabb5de --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/RunStatus.cs @@ -0,0 +1,34 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Specifies the current operational state of a workflow run. +/// +public enum RunStatus +{ + /// + /// The run has not yet started. This only occurs when running in "lockstep" mode. + /// + NotStarted, + + /// + /// The run has halted, has no outstanding requets, but has not received a . + /// + Idle, + + /// + /// The run has halted, and has at least one outstanding . + /// + PendingRequests, + + /// + /// The user has ended the run. No further events will be emitted, and no messages can be sent to it. + /// + Ended, + + /// + /// The workflow is currently running, and may receive events or requests. + /// + Running +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs index 74a4f87b32..78f79148e7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs @@ -13,7 +13,7 @@ using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.AI.Workflows.Specialized; -internal class WorkflowHostExecutor : Executor, IResettableExecutor +internal class WorkflowHostExecutor : Executor, IAsyncDisposable { private readonly string _runId; private readonly Workflow _workflow; @@ -114,8 +114,9 @@ internal class WorkflowHostExecutor : Executor, IResettableExecutor throw new InvalidOperationException("No checkpoints available to resume from."); } - runHandle = await activeRunner.ResumeStreamAsync(InProcessExecution.DefaultMode, lastCheckpoint!, cancellation) - .ConfigureAwait(false); + runHandle = await activeRunner.ResumeStreamAsync(ExecutionMode.Subworkflow, lastCheckpoint!, cancellation) + .ConfigureAwait(false); + if (incomingMessage != null) { await runHandle.EnqueueUntypedAndRunAsync(incomingMessage, cancellation).ConfigureAwait(false); @@ -123,8 +124,8 @@ internal class WorkflowHostExecutor : Executor, IResettableExecutor } else if (incomingMessage != null) { - runHandle = await activeRunner.BeginStreamAsync(InProcessExecution.DefaultMode, cancellation) - .ConfigureAwait(false); + runHandle = await activeRunner.BeginStreamAsync(ExecutionMode.Subworkflow, cancellation) + .ConfigureAwait(false); await runHandle.EnqueueUntypedAndRunAsync(incomingMessage, cancellation).ConfigureAwait(false); } @@ -135,7 +136,7 @@ internal class WorkflowHostExecutor : Executor, IResettableExecutor } else { - runHandle = await activeRunner.BeginStreamAsync(InProcessExecution.DefaultMode, cancellation).ConfigureAwait(false); + runHandle = await activeRunner.BeginStreamAsync(ExecutionMode.Subworkflow, cancellation).ConfigureAwait(false); await runHandle.EnqueueMessageUntypedAsync(Throw.IfNull(incomingMessage), cancellation: cancellation).ConfigureAwait(false); } @@ -251,9 +252,13 @@ internal class WorkflowHostExecutor : Executor, IResettableExecutor StreamingRun run = await this.EnsureRunSendMessageAsync(cancellation: cancellationToken).ConfigureAwait(false); } - public async ValueTask ResetAsync() + private async ValueTask ResetAsync() { - this._run = null; + if (this._run != null) + { + await this._run.DisposeAsync().ConfigureAwait(false); + this._run = null; + } if (this._activeRunner != null) { @@ -263,4 +268,6 @@ internal class WorkflowHostExecutor : Executor, IResettableExecutor this._activeRunner = new(this._workflow, this._checkpointManager, this._runId); } } + + public ValueTask DisposeAsync() => this.ResetAsync(); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs index 0d8e9f370c..aa70307c1e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs @@ -2,7 +2,6 @@ using System; using System.Collections.Generic; -using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; @@ -15,7 +14,7 @@ namespace Microsoft.Agents.AI.Workflows; /// A run instance supporting a streaming form of receiving workflow events, and providing /// a mechanism to send responses back to the workflow. /// -public sealed class StreamingRun +public sealed class StreamingRun : IAsyncDisposable { private readonly AsyncRunHandle _runHandle; @@ -24,9 +23,6 @@ public sealed class StreamingRun this._runHandle = Throw.IfNull(runHandle); } - private ValueTask WaitOnInputAsync(CancellationToken cancellation = default) - => this._runHandle.WaitForNextInputAsync(cancellation); - /// /// A unique identifier for the run. Can be provided at the start of the run, or auto-generated. /// @@ -78,50 +74,15 @@ public sealed class StreamingRun CancellationToken cancellationToken = default) => this.WatchStreamAsync(blockOnPendingRequest: true, cancellationToken); - internal async IAsyncEnumerable WatchStreamAsync( + internal IAsyncEnumerable WatchStreamAsync( bool blockOnPendingRequest, - [EnumeratorCancellation] CancellationToken cancellationToken = default) + CancellationToken cancellationToken = default) + => this._runHandle.TakeEventStreamAsync(blockOnPendingRequest, cancellationToken); + + /// + public ValueTask DisposeAsync() { - RunStatus runStatus; - - do - { - await foreach (WorkflowEvent @event in this._runHandle.TakeEventStreamAsync(breakOnHalt: true, cancellationToken) - .WithCancellation(cancellationToken) - .ConfigureAwait(false)) - { - yield return @event; - } - - if (cancellationToken.IsCancellationRequested) - { - yield break; // We are done. - } - - runStatus = await this._runHandle.GetStatusAsync(cancellationToken).ConfigureAwait(false); - if (runStatus == RunStatus.Idle) - { - yield break; // We are done. - } - - if (blockOnPendingRequest && runStatus == RunStatus.PendingRequests) - { - // Although we are only doing this while there are pending requests, any input allows us to continue - // running, so we should not wait until the input is specifically an ExternalResponse. - await this.WaitOnInputAsync(cancellationToken).ConfigureAwait(false); - } - } while (runStatus == RunStatus.Running); - } - - /// - /// Signals the end of the current run and initiates any necessary cleanup operations asynchronously. - /// Enables the underlying Workflow instance to be reused in subsequent runs. - /// - /// A ValueTask that represents the asynchronous operation. The task is complete when the run has - /// ended and cleanup is finished. - public async ValueTask EndRunAsync() - { - await this._runHandle.DisposeAsync().ConfigureAwait(false); + return this._runHandle.DisposeAsync(); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs index a656e33d1c..4964332148 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs @@ -193,7 +193,6 @@ public class Workflow await this.TryResetExecutorRegistrationsAsync().ConfigureAwait(false); Interlocked.CompareExchange(ref this._ownerToken, null, ownerToken); - this._ownerToken = null; } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs index 6e739ad92f..965a23a22f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs @@ -65,7 +65,7 @@ internal sealed class WorkflowHostAgent : AIAgent if (!this._runningWorkflows.TryGetValue(runId, out StreamingRun? run)) { run = await InProcessExecution.StreamAsync(this._workflow, messages, cancellationToken: cancellationToken) - .ConfigureAwait(false); + .ConfigureAwait(false); this._runningWorkflows[runId] = run; } else diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.IntegrationTests/Framework/WorkflowHarness.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.IntegrationTests/Framework/WorkflowHarness.cs index 679e1299c8..797e254800 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.IntegrationTests/Framework/WorkflowHarness.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.IntegrationTests/Framework/WorkflowHarness.cs @@ -40,7 +40,7 @@ internal sealed class WorkflowHarness(Workflow workflow, string runId) { Console.WriteLine("RUNNING WORKFLOW..."); Checkpointed run = await InProcessExecution.StreamAsync(workflow, input, this._checkpointManager, runId); - IReadOnlyList workflowEvents = await MonitorWorkflowRunAsync(run).ToArrayAsync(); + IReadOnlyList workflowEvents = await MonitorAndDisposeWorkflowRunAsync(run).ToArrayAsync(); this.LastCheckpoint = workflowEvents.OfType().LastOrDefault()?.CompletionInfo?.Checkpoint; return new WorkflowEvents(workflowEvents); } @@ -50,7 +50,7 @@ internal sealed class WorkflowHarness(Workflow workflow, string runId) Console.WriteLine("RESUMING WORKFLOW..."); Assert.NotNull(this.LastCheckpoint); Checkpointed run = await InProcessExecution.ResumeStreamAsync(workflow, this.LastCheckpoint, this._checkpointManager, runId); - IReadOnlyList workflowEvents = await MonitorWorkflowRunAsync(run, response).ToArrayAsync(); + IReadOnlyList workflowEvents = await MonitorAndDisposeWorkflowRunAsync(run, response).ToArrayAsync(); return new WorkflowEvents(workflowEvents); } @@ -75,8 +75,10 @@ internal sealed class WorkflowHarness(Workflow workflow, string runId) return new WorkflowHarness(workflow, runId); } - private static async IAsyncEnumerable MonitorWorkflowRunAsync(Checkpointed run, InputResponse? response = null) + private static async IAsyncEnumerable MonitorAndDisposeWorkflowRunAsync(Checkpointed run, InputResponse? response = null) { + await using IAsyncDisposable disposeRun = run; + await foreach (WorkflowEvent workflowEvent in run.Run.WatchStreamAsync().ConfigureAwait(false)) { bool exitLoop = false; @@ -93,7 +95,6 @@ internal sealed class WorkflowHarness(Workflow workflow, string runId) } else { - await run.Run.EndRunAsync().ConfigureAwait(false); exitLoop = true; } break; diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs index fb7c77f59d..ba77761906 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs @@ -259,7 +259,7 @@ public sealed class DeclarativeWorkflowTest(ITestOutputHelper output) : Workflow Workflow workflow = DeclarativeWorkflowBuilder.Build(yamlReader, workflowContext); - StreamingRun run = await InProcessExecution.StreamAsync(workflow, workflowInput); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, workflowInput); this.WorkflowEvents = run.WatchStreamAsync().ToEnumerable().ToList(); foreach (WorkflowEvent workflowEvent in this.WorkflowEvents) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/WorkflowActionExecutorTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/WorkflowActionExecutorTest.cs index 10997e397c..62f9231e68 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/WorkflowActionExecutorTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/WorkflowActionExecutorTest.cs @@ -29,7 +29,7 @@ public abstract class WorkflowActionExecutorTest(ITestOutputHelper output) : Wor TestWorkflowExecutor workflowExecutor = new(); WorkflowBuilder workflowBuilder = new(workflowExecutor); workflowBuilder.AddEdge(workflowExecutor, executor); - StreamingRun run = await InProcessExecution.StreamAsync(workflowBuilder.Build(), this.State); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflowBuilder.Build(), this.State); WorkflowEvent[] events = await run.WatchStreamAsync().ToArrayAsync(); Assert.Contains(events, e => e is DeclarativeActionInvokedEvent); Assert.Contains(events, e => e is DeclarativeActionCompletedEvent); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs index d9c3863dd6..8a534269f0 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs @@ -385,31 +385,24 @@ public class AgentWorkflowBuilderTests { StringBuilder sb = new(); - StreamingRun run = await InProcessExecution.StreamAsync(workflow, input); - try - { - await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input); + await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); - WorkflowOutputEvent? output = null; - await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) + WorkflowOutputEvent? output = null; + await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) + { + if (evt is AgentRunUpdateEvent executorComplete) { - if (evt is AgentRunUpdateEvent executorComplete) - { - sb.Append(executorComplete.Data); - } - else if (evt is WorkflowOutputEvent e) - { - output = e; - break; - } + sb.Append(executorComplete.Data); } + else if (evt is WorkflowOutputEvent e) + { + output = e; + break; + } + } - return (sb.ToString(), output?.As>()); - } - finally - { - await run.EndRunAsync(); - } + return (sb.ToString(), output?.As>()); } private sealed class DoubleEchoAgentWithBarrier(string name, StrongBox> barrier, StrongBox remaining) : DoubleEchoAgent(name) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ExecutionExtensions.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ExecutionExtensions.cs new file mode 100644 index 0000000000..747461323d --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ExecutionExtensions.cs @@ -0,0 +1,20 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using Microsoft.Agents.AI.Workflows.InProc; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +internal static class ExecutionExtensions +{ + public static InProcessExecutionEnvironment GetEnvironment(this ExecutionMode executionMode) + { + return executionMode switch + { + ExecutionMode.OffThread => InProcessExecution.OffThread, + ExecutionMode.Lockstep => InProcessExecution.Lockstep, + ExecutionMode.Subworkflow => throw new NotSupportedException(), + _ => throw new InvalidOperationException($"Unknown execution mode {executionMode}") + }; + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessStateTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessStateTests.cs index fff53a8f55..debde5fc95 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessStateTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessStateTests.cs @@ -163,9 +163,25 @@ public class InProcessStateTests .AddFanOutEdge(forward, targets: [testExecutor, testExecutor2]) .Build(); - var act = async () => await InProcessExecution.RunAsync(workflow, new TurnToken()); + Run runWithFailure = await InProcessExecution.RunAsync(workflow, new TurnToken()); - var result = await act.Should() - .ThrowAsync("multiple writers to the same shared scope key"); + bool hadFailure = false; + foreach (WorkflowEvent evt in runWithFailure.NewEvents) + { + if (evt is WorkflowErrorEvent errorEvent) + { + hadFailure.Should().BeFalse("There can be only one!"); + hadFailure = true; + + errorEvent.Data.Should().BeOfType() + .Subject.Message.Should().Contain("TestKey"); + } + } + + hadFailure.Should().BeTrue(); + + //var act = async () => await InProcessExecution.RunAsync(workflow, new TurnToken()); + //var result = await act.Should() + // .ThrowAsync("multiple writers to the same shared scope key"); } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs index fff05ae607..2e89dabd9e 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs @@ -1,9 +1,12 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.IO; using System.Linq; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Agents.AI.Workflows.Reflection; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -23,9 +26,11 @@ internal static class Step1EntryPoint } } - public static async ValueTask RunAsync(TextWriter writer) + public static async ValueTask RunAsync(TextWriter writer, ExecutionMode executionMode) { - StreamingRun run = await InProcessExecution.StreamAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + + StreamingRun run = await env.StreamAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs index 692001351c..04757b9350 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01a_Simple_Workflow_Sequential.cs @@ -2,16 +2,18 @@ using System.IO; using System.Threading.Tasks; - +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.UnitTests; using static Microsoft.Agents.AI.Workflows.Sample.Step1EntryPoint; namespace Microsoft.Agents.AI.Workflows.Sample; internal static class Step1aEntryPoint { - public static async ValueTask RunAsync(TextWriter writer) + public static async ValueTask RunAsync(TextWriter writer, ExecutionMode executionMode) { - Run run = await InProcessExecution.RunAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + Run run = await env.RunAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false); Assert.Equal(RunStatus.Idle, await run.GetStatusAsync()); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs index ed0ccbd8f3..dd02f93594 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs @@ -4,7 +4,9 @@ using System; using System.IO; using System.Linq; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Agents.AI.Workflows.Reflection; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -28,9 +30,10 @@ internal static class Step2EntryPoint } } - public static async ValueTask RunAsync(TextWriter writer, string input = "This is a spam message.") + public static async ValueTask RunAsync(TextWriter writer, ExecutionMode executionMode, string input = "This is a spam message.") { - StreamingRun handle = await InProcessExecution.StreamAsync(WorkflowInstance, input).ConfigureAwait(false); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + StreamingRun handle = await env.StreamAsync(WorkflowInstance, input).ConfigureAwait(false); await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false)) { switch (evt) 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 4da5e82407..be1d1e1623 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 @@ -4,7 +4,9 @@ using System; using System.IO; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Agents.AI.Workflows.Reflection; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -25,9 +27,10 @@ internal static class Step3EntryPoint } } - public static async ValueTask RunAsync(TextWriter writer) + public static async ValueTask RunAsync(TextWriter writer, ExecutionMode executionMode) { - StreamingRun run = await InProcessExecution.StreamAsync(WorkflowInstance, NumberSignal.Init).ConfigureAwait(false); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + StreamingRun run = await env.StreamAsync(WorkflowInstance, NumberSignal.Init).ConfigureAwait(false); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs index 7e96584ed4..e297e9da91 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs @@ -4,6 +4,8 @@ using System; using System.Collections.Generic; using System.IO; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -37,13 +39,14 @@ internal static class Step4EntryPoint } } - public static async ValueTask RunAsync(TextWriter writer, Func userGuessCallback) + public static async ValueTask RunAsync(TextWriter writer, Func userGuessCallback, ExecutionMode executionMode) { NumberSignal signal = NumberSignal.Init; string? prompt = UpdatePrompt(null, signal); Workflow workflow = WorkflowInstance; - StreamingRun handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + StreamingRun handle = await env.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false); List requests = []; await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false)) 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 4fd983f0e3..4e6a137cdc 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 @@ -6,12 +6,14 @@ using System.IO; using System.Threading; using System.Threading.Tasks; using FluentAssertions; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; internal static class Step5EntryPoint { - public static async ValueTask RunAsync(TextWriter writer, Func userGuessCallback, bool rehydrateToRestore = false, CheckpointManager? checkpointManager = null) + public static async ValueTask RunAsync(TextWriter writer, Func userGuessCallback, ExecutionMode executionMode, bool rehydrateToRestore = false, CheckpointManager? checkpointManager = null) { Dictionary checkpointedOutputs = []; @@ -21,9 +23,11 @@ internal static class Step5EntryPoint checkpointManager ??= CheckpointManager.Default; Workflow workflow = Step4EntryPoint.CreateWorkflowInstance(out JudgeExecutor judge); + + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); Checkpointed checkpointed = - await InProcessExecution.StreamAsync(workflow, NumberSignal.Init, checkpointManager) - .ConfigureAwait(false); + await env.StreamAsync(workflow, NumberSignal.Init, checkpointManager) + .ConfigureAwait(false); List checkpoints = []; CancellationTokenSource cancellationSource = new(); @@ -40,10 +44,10 @@ internal static class Step5EntryPoint Console.WriteLine($"Restoring to checkpoint {targetCheckpoint} from run {targetCheckpoint.RunId}"); if (rehydrateToRestore) { - await handle.EndRunAsync().ConfigureAwait(false); + await handle.DisposeAsync().ConfigureAwait(false); - checkpointed = await InProcessExecution.ResumeStreamAsync(workflow, targetCheckpoint, checkpointManager, runId: handle.RunId, cancellationToken: CancellationToken.None) - .ConfigureAwait(false); + checkpointed = await env.ResumeStreamAsync(workflow, targetCheckpoint, checkpointManager, runId: handle.RunId, cancellationToken: CancellationToken.None) + .ConfigureAwait(false); handle = checkpointed.Run; } else @@ -112,20 +116,21 @@ internal static class Step5EntryPoint { Console.WriteLine($"*** Max step {maxStep} reached, cancelling."); cancellationSource.Cancel(); + return null; } - else - { - Console.WriteLine($"*** Processing {requests.Count} queued requests."); - foreach (ExternalRequest request in requests) - { - ExternalResponse response = ExecuteExternalRequest(request, userGuessCallback, prompt); - Console.WriteLine($"!!! Sending response: {response}"); - await handle.SendResponseAsync(response).ConfigureAwait(false); - } - requests.Clear(); - Console.WriteLine("*** Completed processing requests."); + Console.WriteLine($"*** Processing {requests.Count} queued requests."); + foreach (ExternalRequest request in requests) + { + ExternalResponse response = ExecuteExternalRequest(request, userGuessCallback, prompt); + Console.WriteLine($"!!! Sending response: {response}"); + await handle.SendResponseAsync(response).ConfigureAwait(false); } + + requests.Clear(); + + Console.WriteLine("*** Completed processing requests."); + break; case ExecutorCompletedEvent executorCompleteEvt: diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs index 2f9eaf0d32..3e6049f4ce 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs @@ -10,6 +10,8 @@ using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.UnitTests; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -22,12 +24,13 @@ internal static class Step6EntryPoint .AddParticipants(new HelloAgent(), new EchoAgent()) .Build(); - public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2) + public static async ValueTask RunAsync(TextWriter writer, ExecutionMode executionMode, int maxSteps = 2) { Workflow workflow = CreateWorkflow(maxSteps); - StreamingRun run = await InProcessExecution.StreamAsync(workflow, Array.Empty()) - .ConfigureAwait(false); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + StreamingRun run = await env.StreamAsync(workflow, Array.Empty()) + .ConfigureAwait(false); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs index 39bf2cb626..9c74bc47e8 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs @@ -1,6 +1,5 @@ // Copyright (c) Microsoft. All rights reserved. -using System; using System.Collections.Generic; using System.IO; using System.Threading.Tasks; @@ -26,7 +25,6 @@ internal static class Step7EntryPoint ?? update.AgentId ?? update.Role.ToString() ?? ChatRole.Assistant.ToString()}: {update.Text}"; - Console.WriteLine(updateText); writer.WriteLine(updateText); } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/08_Subworkflow_Simple.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/08_Subworkflow_Simple.cs index 3b47cafa39..d865a6275c 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/08_Subworkflow_Simple.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/08_Subworkflow_Simple.cs @@ -7,6 +7,8 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; using FluentAssertions; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -26,7 +28,7 @@ internal static class Step8EntryPoint " Spaces around text ", ]; - public static async ValueTask> RunAsync(TextWriter writer, List textsToProcess) + public static async ValueTask> RunAsync(TextWriter writer, ExecutionMode executionMode, List textsToProcess) { Func processTextAsyncFunc = ProcessTextAsync; ExecutorIsh processText = processTextAsyncFunc.AsExecutor("TextProcessor"); @@ -41,7 +43,8 @@ internal static class Step8EntryPoint .AddEdge(textProcessor, orchestrator) .Build(); - Run workflowRun = await InProcessExecution.RunAsync(workflow, textsToProcess); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + Run workflowRun = await env.RunAsync(workflow, textsToProcess); RunStatus status = await workflowRun.GetStatusAsync(); status.Should().Be(RunStatus.Idle); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs index a238292df4..083221ab2a 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs @@ -7,6 +7,8 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; using FluentAssertions; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.UnitTests; namespace Microsoft.Agents.AI.Workflows.Sample; @@ -170,12 +172,13 @@ internal static class Step9EntryPoint .Select(request => Part2FinishedResponses[request.Id]) .OrderBy(request => request.Id)]; - public static async ValueTask> RunAsync(TextWriter writer) + public static async ValueTask> RunAsync(TextWriter writer, ExecutionMode executionMode) { RunStatus runStatus; List results = []; - Run workflowRun = await InProcessExecution.RunAsync(WorkflowInstance, RequestsToProcess.ToList()); + InProcessExecutionEnvironment env = executionMode.GetEnvironment(); + Run workflowRun = await env.RunAsync(WorkflowInstance, RequestsToProcess.ToList()); RunStatus part1Status = ExpectedResponsesPart2.Length > 0 ? RunStatus.PendingRequests : RunStatus.Idle; runStatus = await workflowRun.GetStatusAsync(); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SampleSmokeTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SampleSmokeTest.cs index 44bc1222da..378475a856 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SampleSmokeTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SampleSmokeTest.cs @@ -13,12 +13,14 @@ namespace Microsoft.Agents.AI.Workflows.UnitTests; public class SampleSmokeTest { - [Fact] - public async Task Test_RunSample_Step1Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step1Async(ExecutionMode executionMode) { using StringWriter writer = new(); - await Step1EntryPoint.RunAsync(writer); + await Step1EntryPoint.RunAsync(writer, executionMode); string result = writer.ToString(); string[] lines = result.Split([Environment.NewLine], StringSplitOptions.RemoveEmptyEntries); @@ -31,12 +33,14 @@ public class SampleSmokeTest ); } - [Fact] - public async Task Test_RunSample_Step1aAsync() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step1aAsync(ExecutionMode executionMode) { using StringWriter writer = new(); - await Step1aEntryPoint.RunAsync(writer); + await Step1aEntryPoint.RunAsync(writer, executionMode); string result = writer.ToString(); string[] lines = result.Split([Environment.NewLine], StringSplitOptions.RemoveEmptyEntries); @@ -49,32 +53,38 @@ public class SampleSmokeTest ); } - [Fact] - public async Task Test_RunSample_Step2Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step2Async(ExecutionMode executionMode) { using StringWriter writer = new(); - string spamResult = await Step2EntryPoint.RunAsync(writer); + string spamResult = await Step2EntryPoint.RunAsync(writer, executionMode); Assert.Equal(RemoveSpamExecutor.ActionResult, spamResult); - string nonSpamResult = await Step2EntryPoint.RunAsync(writer, "This is a valid message."); + string nonSpamResult = await Step2EntryPoint.RunAsync(writer, executionMode, "This is a valid message."); Assert.Equal(RespondToMessageExecutor.ActionResult, nonSpamResult); } - [Fact] - public async Task Test_RunSample_Step3Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step3Async(ExecutionMode executionMode) { using StringWriter writer = new(); - string guessResult = await Step3EntryPoint.RunAsync(writer); + string guessResult = await Step3EntryPoint.RunAsync(writer, executionMode); Assert.Equal("Guessed the number: 42", guessResult); } - [Fact] - public async Task Test_RunSample_Step4Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step4Async(ExecutionMode executionMode) { using StringWriter writer = new(); @@ -83,12 +93,14 @@ 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); + string guessResult = await Step4EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext, executionMode); Assert.Equal("You guessed correctly! You Win!", guessResult); } - [Fact] - public async Task Test_RunSample_Step5Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step5Async(ExecutionMode executionMode) { using StringWriter writer = new(); @@ -102,12 +114,14 @@ public class SampleSmokeTest ("Your guess was too low. Try again.", 42) ); - string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext); + string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext, executionMode); Assert.Equal("You guessed correctly! You Win!", guessResult); } - [Fact] - public async Task Test_RunSample_Step5aAsync() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step5aAsync(ExecutionMode executionMode) { using StringWriter writer = new(); @@ -121,12 +135,14 @@ public class SampleSmokeTest ("Your guess was too low. Try again.", 42) ); - string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext, rehydrateToRestore: true); + string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext, executionMode, rehydrateToRestore: true); Assert.Equal("You guessed correctly! You Win!", guessResult); } - [Fact] - public async Task Test_RunSample_Step5bAsync() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step5bAsync(ExecutionMode executionMode) { using StringWriter writer = new(); @@ -144,16 +160,18 @@ public class SampleSmokeTest options.MakeReadOnly(); CheckpointManager memoryJsonManager = CheckpointManager.CreateJson(new InMemoryJsonStore(), options); - string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext, rehydrateToRestore: true, checkpointManager: memoryJsonManager); + string guessResult = await Step5EntryPoint.RunAsync(writer, userGuessCallback: responder.InvokeNext, executionMode, rehydrateToRestore: true, checkpointManager: memoryJsonManager); Assert.Equal("You guessed correctly! You Win!", guessResult); } - [Fact] - public async Task Test_RunSample_Step6Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step6Async(ExecutionMode executionMode) { using StringWriter writer = new(); - await Step6EntryPoint.RunAsync(writer); + await Step6EntryPoint.RunAsync(writer, executionMode); string result = writer.ToString(); string[] lines = result.Split([Environment.NewLine], StringSplitOptions.RemoveEmptyEntries); @@ -180,8 +198,10 @@ public class SampleSmokeTest ); } - [Fact] - public async Task Test_RunSample_Step8Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step8Async(ExecutionMode executionMode) { List textsToProcess = [ "Hello world! This is a simple test.", @@ -194,7 +214,7 @@ public class SampleSmokeTest using StringWriter writer = new(); - List results = await Step8EntryPoint.RunAsync(writer, textsToProcess); + List results = await Step8EntryPoint.RunAsync(writer, executionMode, textsToProcess); Assert.Equal(textsToProcess.Count, results.Count); Assert.Collection(results, @@ -216,11 +236,13 @@ public class SampleSmokeTest } } - [Fact] - public async Task Test_RunSample_Step9Async() + [Theory] + [InlineData(ExecutionMode.Lockstep)] + [InlineData(ExecutionMode.OffThread)] + internal async Task Test_RunSample_Step9Async(ExecutionMode executionMode) { using StringWriter writer = new(); - _ = await Step9EntryPoint.RunAsync(writer); + _ = await Step9EntryPoint.RunAsync(writer, executionMode); } }