diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepJoinContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepJoinContext.cs index f4af19bcfd..f0d6d5c151 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepJoinContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepJoinContext.cs @@ -13,6 +13,7 @@ internal interface ISuperStepJoinContext ValueTask ForwardWorkflowEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default); ValueTask SendMessageAsync(string senderId, [DisallowNull] TMessage message, CancellationToken cancellationToken = default); + ValueTask YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken = default); ValueTask AttachSuperstepAsync(ISuperStepRunner superStepRunner, CancellationToken cancellationToken = default); ValueTask DetachSuperstepAsync(string id); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index 1750f779f2..47f04f8da0 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -429,4 +429,21 @@ internal sealed class InProcessRunnerContext : IRunnerContext ValueTask ISuperStepJoinContext.SendMessageAsync(string senderId, [DisallowNull] TMessage message, CancellationToken cancellationToken) => this.SendMessageAsync(senderId, Throw.IfNull(message), cancellationToken: cancellationToken); + + async ValueTask ISuperStepJoinContext.YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken) + { + this.CheckEnded(); + Throw.IfNull(output); + + Executor sourceExecutor = await this.EnsureExecutorAsync(sourceId, tracer: null, cancellationToken).ConfigureAwait(false); + if (!sourceExecutor.CanOutput(output.GetType())) + { + throw new InvalidOperationException($"Cannot output object of type {output.GetType().Name}. Expecting one of [{string.Join(", ", sourceExecutor.OutputTypes)}]."); + } + + if (this._outputFilter.CanOutput(sourceId, output)) + { + await this.AddEventAsync(new WorkflowOutputEvent(output, sourceId), cancellationToken).ConfigureAwait(false); + } + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs index 409f751107..cd3f392718 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs @@ -192,11 +192,16 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable resultTask = this._joinContext?.ForwardWorkflowEventAsync(new SubworkflowErrorEvent(this.Id, errorEvent.Data as Exception)).AsTask() ?? Task.CompletedTask; break; case WorkflowOutputEvent outputEvent: - if (this._joinContext != null && - this._options.AutoSendMessageHandlerResultObject - && outputEvent.Data != null) + if (this._joinContext != null && outputEvent.Data != null) { - resultTask = this._joinContext.SendMessageAsync(this.Id, outputEvent.Data).AsTask(); + if (this._options.AutoSendMessageHandlerResultObject) + { + resultTask = this._joinContext.SendMessageAsync(this.Id, outputEvent.Data).AsTask(); + } + else if (this._options.AutoYieldOutputHandlerResultObject) + { + resultTask = this._joinContext.YieldOutputAsync(this.Id, outputEvent.Data).AsTask(); + } } break; case RequestHaltEvent requestHaltEvent: diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestRunContext.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestRunContext.cs index 57375b8341..9e7f391f62 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestRunContext.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestRunContext.cs @@ -107,6 +107,9 @@ public class TestRunContext : IRunnerContext public ValueTask SendMessageAsync(string senderId, [System.Diagnostics.CodeAnalysis.DisallowNull] TMessage message, CancellationToken cancellationToken = default) => this.SendMessageAsync(senderId, message, cancellationToken); + public ValueTask YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken = default) + => this.AddEventAsync(new WorkflowOutputEvent(output, sourceId), cancellationToken); + ValueTask ISuperStepJoinContext.AttachSuperstepAsync(ISuperStepRunner superStepRunner, CancellationToken cancellationToken) => new(string.Empty); ValueTask ISuperStepJoinContext.DetachSuperstepAsync(string joinId) => new(false); } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostExecutorTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostExecutorTests.cs new file mode 100644 index 0000000000..27bbb21d76 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostExecutorTests.cs @@ -0,0 +1,171 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.Specialized; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +public sealed class WorkflowHostExecutorTests +{ + [Fact] + public async Task ForwardWorkflowEventAsync_WithAutoYieldOutputHandlerResultObjectTrue_YieldsOutput() + { + // Arrange + const string testData = "test output data"; + WorkflowOutputEvent outputEvent = new(testData, "SubworkflowExecutor"); + + TestRunContext testContext = new(); + + ExecutorOptions options = new() + { + AutoSendMessageHandlerResultObject = false, + AutoYieldOutputHandlerResultObject = true + }; + + Workflow emptyWorkflow = new WorkflowBuilder(new SimpleTestExecutor("start")).Build(); + TestableWorkflowHostExecutor hostExecutor = new("TestHost", emptyWorkflow, "run1", new object(), options); + + await hostExecutor.AttachSuperStepContextAsync(testContext); + + // Act + await hostExecutor.SimulateForwardWorkflowEventAsync(outputEvent); + + // Assert + testContext.Events.OfType().Should().HaveCount(1, "YieldOutputAsync should create one WorkflowOutputEvent"); + WorkflowOutputEvent? yieldedEvent = testContext.Events.OfType().FirstOrDefault(); + yieldedEvent.Should().NotBeNull(); + yieldedEvent!.SourceId.Should().Be("TestHost"); + yieldedEvent.As().Should().Be(testData); + + testContext.QueuedMessages.Should().BeEmpty("SendMessageAsync should not be called"); + } + + [Fact] + public async Task ForwardWorkflowEventAsync_WithAutoSendMessageHandlerResultObjectTrue_SendsMessage() + { + // Arrange + const string testData = "test output data"; + WorkflowOutputEvent outputEvent = new(testData, "SubworkflowExecutor"); + + TestRunContext testContext = new(); + + ExecutorOptions options = new() + { + AutoSendMessageHandlerResultObject = true, + AutoYieldOutputHandlerResultObject = false + }; + + Workflow emptyWorkflow = new WorkflowBuilder(new SimpleTestExecutor("start")).Build(); + TestableWorkflowHostExecutor hostExecutor = new("TestHost", emptyWorkflow, "run1", new object(), options); + + await hostExecutor.AttachSuperStepContextAsync(testContext); + + // Act + await hostExecutor.SimulateForwardWorkflowEventAsync(outputEvent); + + // Assert + testContext.QueuedMessages.Should().ContainKey("TestHost"); + testContext.QueuedMessages["TestHost"].Should().HaveCount(1); + testContext.QueuedMessages["TestHost"][0].Message.Should().Be(testData); + + testContext.Events.OfType().Should().BeEmpty("YieldOutputAsync should not be called"); + } + + [Fact] + public async Task ForwardWorkflowEventAsync_WithBothOptionsFalse_DoesNothing() + { + // Arrange + const string testData = "test output data"; + WorkflowOutputEvent outputEvent = new(testData, "SubworkflowExecutor"); + + TestRunContext testContext = new(); + + ExecutorOptions options = new() + { + AutoSendMessageHandlerResultObject = false, + AutoYieldOutputHandlerResultObject = false + }; + + Workflow emptyWorkflow = new WorkflowBuilder(new SimpleTestExecutor("start")).Build(); + TestableWorkflowHostExecutor hostExecutor = new("TestHost", emptyWorkflow, "run1", new object(), options); + + await hostExecutor.AttachSuperStepContextAsync(testContext); + + // Act + await hostExecutor.SimulateForwardWorkflowEventAsync(outputEvent); + + // Assert + testContext.QueuedMessages.Should().BeEmpty("SendMessageAsync should not be called"); + testContext.Events.OfType().Should().BeEmpty("YieldOutputAsync should not be called"); + } + + [Fact] + public async Task ForwardWorkflowEventAsync_WithNullOutputData_DoesNothing() + { + // Arrange + WorkflowOutputEvent outputEvent = new(null!, "SubworkflowExecutor"); + + TestRunContext testContext = new(); + + ExecutorOptions options = new() + { + AutoSendMessageHandlerResultObject = false, + AutoYieldOutputHandlerResultObject = true + }; + + Workflow emptyWorkflow = new WorkflowBuilder(new SimpleTestExecutor("start")).Build(); + TestableWorkflowHostExecutor hostExecutor = new("TestHost", emptyWorkflow, "run1", new object(), options); + + await hostExecutor.AttachSuperStepContextAsync(testContext); + + // Act + await hostExecutor.SimulateForwardWorkflowEventAsync(outputEvent); + + // Assert + testContext.Events.OfType().Should().BeEmpty("YieldOutputAsync should not be called when data is null"); + testContext.QueuedMessages.Should().BeEmpty("SendMessageAsync should not be called when data is null"); + } + + /// + /// Simple executor for testing that doesn't require type parameters. + /// + private sealed class SimpleTestExecutor : Executor + { + public SimpleTestExecutor(string id) : base(id) + { + } + + protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder; + } + + /// + /// Testable wrapper for WorkflowHostExecutor that exposes internal methods for testing. + /// + private sealed class TestableWorkflowHostExecutor : WorkflowHostExecutor + { + public TestableWorkflowHostExecutor(string id, Workflow workflow, string runId, object ownershipToken, ExecutorOptions? options = null) + : base(id, workflow, runId, ownershipToken, options) + { + } + + public async ValueTask SimulateForwardWorkflowEventAsync(WorkflowEvent evt) + { + // Use reflection to invoke the private ForwardWorkflowEventAsync method + System.Reflection.MethodInfo? method = typeof(WorkflowHostExecutor) + .GetMethod("ForwardWorkflowEventAsync", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance); + + if (method != null) + { + object? result = method.Invoke(this, [null, evt]); + if (result is ValueTask valueTask) + { + await valueTask; + } + } + } + } +}