diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs index 67c35099bc..525632862f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs @@ -126,10 +126,11 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull { - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken) - .ConfigureAwait(false); + var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: null, runId: runId, cancellationToken).ConfigureAwait(false); - return await runHandle.EnqueueAndRunAsync(input, cancellationToken).ConfigureAwait(false); + Run run = new(runHandle); + await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); + return run; } /// @@ -139,10 +140,11 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull { - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); + var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: null, runId: runId, cancellationToken).ConfigureAwait(false); - return await runHandle.EnqueueAndRunAsync(input, cancellationToken).ConfigureAwait(false); + Run run = new(runHandle); + await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); + return run; } /// @@ -153,10 +155,11 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull { - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken) - .ConfigureAwait(false); + var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: checkpointManager, runId: runId, cancellationToken).ConfigureAwait(false); - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndRunAsync(input, cancellationToken)) + Run run = new(runHandle); + await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); + return await runHandle.WithCheckpointingAsync(() => new ValueTask(run)) .ConfigureAwait(false); } @@ -168,10 +171,11 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull { - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); + var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: checkpointManager, runId: runId, cancellationToken).ConfigureAwait(false); - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndRunAsync(input, cancellationToken)) + Run run = new(runHandle); + await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); + return await runHandle.WithCheckpointingAsync(() => new ValueTask(run)) .ConfigureAwait(false); } @@ -204,4 +208,48 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) .ConfigureAwait(false); } + + // Helper to construct a RunHandle with the provided input enqueued. If the starting executor supports it, a TurnToken will be enqueued also. + private async ValueTask GetRunHandleWithTurnTokenAsync( + Workflow workflow, + TInput input, + CheckpointManager? checkpointManager, + string? runId, + CancellationToken cancellationToken) + { + var knownTypes = new List() { typeof(TInput) }; + var needsTurnToken = await StartingExecutorHandlesTurnTokenAsync(workflow).ConfigureAwait(false); + if (needsTurnToken) + { + knownTypes.Add(typeof(TurnToken)); + } + + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: checkpointManager, runId: runId, knownTypes, cancellationToken) + .ConfigureAwait(false); + + await runHandle.EnqueueMessageAsync(input, cancellationToken).ConfigureAwait(false); + + if (needsTurnToken) + { + await runHandle.EnqueueMessageAsync(new TurnToken(emitEvents: true), cancellationToken).ConfigureAwait(false); + } + + return runHandle; + } + + /// + /// Helper method to detect if the starting executor of a given workflow accepts the provided input type as well as a TurnToken. + /// + private static async ValueTask StartingExecutorHandlesTurnTokenAsync(Workflow workflow) + { + if (workflow.Registrations.TryGetValue(workflow.StartExecutorId, out var registration)) + { + // Create instance to check type + Executor startExecutor = await registration.CreateInstanceAsync(string.Empty) + .ConfigureAwait(false); + return startExecutor.CanHandle(typeof(TInput)) && startExecutor.CanHandle(typeof(TurnToken)); + } + + return false; + } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessExecutionTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessExecutionTests.cs new file mode 100644 index 0000000000..ac4f8a5d95 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessExecutionTests.cs @@ -0,0 +1,200 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +/// +/// Tests for InProcessExecution to verify streaming and non-streaming execution behavior. +/// +public class InProcessExecutionTests +{ + /// + /// The non-streaming version (RunAsync) should execute the workflow and produce events, + /// similar to the streaming version (StreamAsync + TrySendMessageAsync). + /// + [Fact] + public async Task RunAsyncShouldExecuteWorkflowAsync() + { + // Arrange: Create a simple agent that responds to messages + var agent = new SimpleTestAgent("test-agent"); + var workflow = AgentWorkflowBuilder.BuildSequential(agent); + var inputMessage = new ChatMessage(ChatRole.User, "Hello"); + + // Act: Execute using non-streaming RunAsync + Run run = await InProcessExecution.RunAsync(workflow, new List { inputMessage }); + + // Assert: The workflow should have executed and produced events + RunStatus status = await run.GetStatusAsync(); + status.Should().Be(RunStatus.Idle, "workflow should complete execution"); + + // The run should have events (at minimum, a WorkflowOutputEvent) + run.OutgoingEvents.Should().NotBeEmpty("workflow should produce events during execution"); + + // Check that we have an agent execution event + var agentEvents = run.OutgoingEvents.OfType().ToList(); + agentEvents.Should().NotBeEmpty("agent should have executed and produced update events"); + + // Check that we have output events + var outputEvents = run.OutgoingEvents.OfType().ToList(); + outputEvents.Should().NotBeEmpty("workflow should produce output events"); + } + + /// + /// This test shows that the streaming version works correctly when TurnToken is sent following a message. + /// + [Fact] + public async Task StreamAsyncWithTurnTokenShouldExecuteWorkflowAsync() + { + // Arrange: Create a simple agent that responds to messages + var agent = new SimpleTestAgent("test-agent"); + var workflow = AgentWorkflowBuilder.BuildSequential(agent); + var inputMessage = new ChatMessage(ChatRole.User, "Hello"); + + // Act: Execute using streaming version with TurnToken + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new List { inputMessage }); + + // Send TurnToken to actually trigger execution (this is the key step) + bool messageSent = await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); + messageSent.Should().BeTrue("TurnToken should be accepted"); + + // Collect events + List events = new(); + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + events.Add(evt); + } + + // Assert: The workflow should have executed and produced events + RunStatus status = await run.GetStatusAsync(); + status.Should().Be(RunStatus.Idle, "workflow should complete execution"); + + events.Should().NotBeEmpty("workflow should produce events during execution"); + + // Check that we have agent execution events + var agentEvents = events.OfType().ToList(); + agentEvents.Should().NotBeEmpty("agent should have executed and produced update events"); + + // Check that we have output events + var outputEvents = events.OfType().ToList(); + outputEvents.Should().NotBeEmpty("workflow should produce output events"); + } + + /// + /// This test compares the behavior of RunAsync vs StreamAsync to highlight the difference. + /// Both should produce similar results, but as of issue #1315, RunAsync fails to execute. + /// + [Fact] + public async Task RunAsyncAndStreamAsyncShouldProduceSimilarResultsAsync() + { + // Arrange: Create the same workflow for both tests + var agent1 = new SimpleTestAgent("test-agent-1"); + var workflow1 = AgentWorkflowBuilder.BuildSequential(agent1); + + var agent2 = new SimpleTestAgent("test-agent-2"); + var workflow2 = AgentWorkflowBuilder.BuildSequential(agent2); + + var inputMessage = new ChatMessage(ChatRole.User, "Test message"); + + // Act 1: Execute using RunAsync (non-streaming) + Run nonStreamingRun = await InProcessExecution.RunAsync(workflow1, new List { inputMessage }); + var nonStreamingEvents = nonStreamingRun.OutgoingEvents.ToList(); + + // Act 2: Execute using StreamAsync (streaming) with TurnToken + await using StreamingRun streamingRun = await InProcessExecution.StreamAsync(workflow2, new List { inputMessage }); + await streamingRun.TrySendMessageAsync(new TurnToken(emitEvents: true)); + + List streamingEvents = new(); + await foreach (WorkflowEvent evt in streamingRun.WatchStreamAsync()) + { + streamingEvents.Add(evt); + } + + // Assert: Both should have produced events + // The streaming version works (we know this from the issue report) + streamingEvents.Should().NotBeEmpty("streaming version should produce events"); + + // The non-streaming version should also produce events (this is the bug being tested) + nonStreamingEvents.Should().NotBeEmpty("non-streaming version should also produce events"); + + // Both should have similar types of events + var streamingAgentEvents = streamingEvents.OfType().Count(); + var nonStreamingAgentEvents = nonStreamingEvents.OfType().Count(); + + nonStreamingAgentEvents.Should().Be(streamingAgentEvents, + "both versions should produce the same number of agent events"); + } + + /// + /// Simple test agent that echoes back the input message. + /// + private sealed class SimpleTestAgent : AIAgent + { + private readonly string _name; + + public SimpleTestAgent(string name) + { + this._name = name; + } + + public override string Name => this._name; + + public override AgentThread GetNewThread() => new SimpleTestAgentThread(); + + public override AgentThread DeserializeThread(System.Text.Json.JsonElement serializedThread, + System.Text.Json.JsonSerializerOptions? jsonSerializerOptions = null) => new SimpleTestAgentThread(); + + public override Task RunAsync( + IEnumerable messages, + AgentThread? thread = null, + AgentRunOptions? options = null, + CancellationToken cancellationToken = default) + { + var lastMessage = messages.LastOrDefault(); + var responseMessage = new ChatMessage(ChatRole.Assistant, $"Echo: {lastMessage?.Text ?? "no message"}"); + return Task.FromResult(new AgentRunResponse(responseMessage)); + } + + public override async IAsyncEnumerable RunStreamingAsync( + IEnumerable messages, + AgentThread? thread = null, + AgentRunOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.Yield(); + + var lastMessage = messages.LastOrDefault(); + var responseText = $"Echo: {lastMessage?.Text ?? "no message"}"; + + string messageId = Guid.NewGuid().ToString("N"); + + // Yield role first + yield return new AgentRunResponseUpdate(ChatRole.Assistant, this._name) + { + AuthorName = this._name, + MessageId = messageId + }; + + // Then yield content + yield return new AgentRunResponseUpdate(ChatRole.Assistant, responseText) + { + AuthorName = this._name, + MessageId = messageId + }; + } + } + + /// + /// Simple thread implementation for SimpleTestAgent. + /// + private sealed class SimpleTestAgentThread : InMemoryAgentThread + { + } +}