From 7ebe00ec3d9459a539ed73e2a8bfb9f33663ce0b Mon Sep 17 00:00:00 2001 From: Jacob Alber Date: Mon, 6 Oct 2025 21:07:38 -0400 Subject: [PATCH] [BREAKING] .NET: Workflow Off-Thread Execution Mode (#1233) * Updates to async run loop. * fix: Workflow Onwership can be release by nonowner * fix: Incorrect handling of blockOnPending in StreamingRun Depending on whether we are running in streaming on non-streaming mode, we may be using the StreamingRun in different ways. Unfortunately, the only place we can really know what is the actual state of execution is in the RunEventStream implementations. This resulted in blocking where blocking was unneeded and occasionally not-blocking when blocking was needed. The fix is to move the logic of handling this blocking into RunEventStream implementations. * fix: Fix cleanup on error and end run This ensures we clean up the background resources correctly. * fix: Ensure we let the run loop proceed when shutting down * fix: Add timeout for Input Waiting * fix: Make the samples properly clean up `Run`s and `StreamingRun`s * fix: Simplify Declarative Workflow Run disposal pattern * Also fixes missing .Disposal() in Integration tests --------- Co-authored-by: Ben Thomas --- .../Agents/CustomAgentExecutors/Program.cs | 2 +- .../Workflows/Agents/FoundryAgent/Program.cs | 2 +- .../CheckpointAndRehydrate/Program.cs | 4 +- .../Checkpoint/CheckpointAndResume/Program.cs | 2 +- .../CheckpointWithHumanInTheLoop/Program.cs | 2 +- .../Concurrent/Concurrent/Program.cs | 2 +- .../Workflows/Concurrent/MapReduce/Program.cs | 2 +- .../01_EdgeCondition/Program.cs | 2 +- .../ConditionalEdges/02_SwitchCase/Program.cs | 2 +- .../03_MultiSelection/Program.cs | 2 +- .../Workflows/Declarative/Program.cs | 9 +- .../Workflows/DeclarativeCode/Program.cs | 6 +- .../HumanInTheLoopBasic/Program.cs | 2 +- .../GettingStarted/Workflows/Loop/Program.cs | 2 +- .../Observability/AspireDashboard/Program.cs | 2 +- .../Workflows/SharedStates/Program.cs | 2 +- .../01_ExecutorsAndEdges/Program.cs | 2 +- .../_Foundational/02_Streaming/Program.cs | 2 +- .../03_AgentsInWorkflows/Program.cs | 2 +- .../04_AgentWorkflowPatterns/Program.cs | 2 +- .../Checkpointed.cs | 16 +- .../Execution/AsyncRunHandle.cs | 96 +++-- .../Execution/IRunEventStream.cs | 8 +- .../Execution/InputWaiter.cs | 42 ++ .../Execution/LockstepRunEventStream.cs | 155 +++++--- .../Execution/StepContext.cs | 2 +- .../Execution/StreamingRunEventStream.cs | 262 +++++++++++++ .../ExecutionMode.cs | 17 +- .../IWorkflowExecutionEnvironment.cs | 188 +++++++++ .../InProc/InProcessExecutionEnvironment.cs | 207 ++++++++++ .../InProc/InProcessRunner.cs | 12 +- .../InProc/InProcessRunnerContext.cs | 13 + .../InProcessExecution.cs | 367 +++--------------- .../src/Microsoft.Agents.AI.Workflows/Run.cs | 52 +-- .../RunStatus.cs | 34 ++ .../Specialized/WorkflowHostExecutor.cs | 23 +- .../StreamingRun.cs | 55 +-- .../Microsoft.Agents.AI.Workflows/Workflow.cs | 1 - .../WorkflowHostAgent.cs | 2 +- .../Framework/WorkflowHarness.cs | 9 +- .../DeclarativeWorkflowTest.cs | 2 +- .../ObjectModel/WorkflowActionExecutorTest.cs | 2 +- .../AgentWorkflowBuilderTests.cs | 35 +- .../ExecutionExtensions.cs | 20 + .../InProcessStateTests.cs | 22 +- .../Sample/01_Simple_Workflow_Sequential.cs | 9 +- .../Sample/01a_Simple_Workflow_Sequential.cs | 8 +- .../Sample/02_Simple_Workflow_Condition.cs | 7 +- .../Sample/03_Simple_Workflow_Loop.cs | 7 +- .../04_Simple_Workflow_ExternalRequest.cs | 7 +- .../05_Simple_Workflow_Checkpointing.cs | 39 +- .../Sample/06_GroupChat_Workflow.cs | 9 +- .../07_GroupChat_Workflow_HostAsAgent.cs | 2 - .../Sample/08_Subworkflow_Simple.cs | 7 +- .../Sample/09_Subworkflow_ExternalRequest.cs | 7 +- .../SampleSmokeTest.cs | 90 +++-- 56 files changed, 1275 insertions(+), 612 deletions(-) create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/Execution/InputWaiter.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/RunStatus.cs create mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ExecutionExtensions.cs 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); } }