Merge branch 'main' into copilot/process-workflow-output-event

- Updated tests to register workflow host executor as output source
- Simplified test implementation to use full workflow execution
- All tests now passing

Co-authored-by: lokitoth <6936551+lokitoth@users.noreply.github.com>

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<TMessage>(string senderId, [DisallowNull] TMessage message, CancellationToken cancellationToken = default);
+    ValueTask YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken = default);
 
     ValueTask<string> AttachSuperstepAsync(ISuperStepRunner superStepRunner, CancellationToken cancellationToken = default);
     ValueTask<bool> DetachSuperstepAsync(string id);

dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs

@@ -429,4 +429,21 @@ internal sealed class InProcessRunnerContext : IRunnerContext
 
     ValueTask ISuperStepJoinContext.SendMessageAsync<TMessage>(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);
+        }
+    }
 }

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:

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/TestRunContext.cs

@@ -107,6 +107,9 @@ public class TestRunContext : IRunnerContext
     public ValueTask SendMessageAsync<TMessage>(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<string> ISuperStepJoinContext.AttachSuperstepAsync(ISuperStepRunner superStepRunner, CancellationToken cancellationToken) => new(string.Empty);
     ValueTask<bool> ISuperStepJoinContext.DetachSuperstepAsync(string joinId) => new(false);
 }

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostExecutorTests.cs

@@ -0,0 +1,190 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+
+namespace Microsoft.Agents.AI.Workflows.UnitTests;
+
+public sealed class WorkflowHostExecutorTests
+{
+    [Fact]
+    public async Task WorkflowHostExecutor_WithAutoYieldOutputHandlerResultObjectTrue_YieldsSubworkflowOutput()
+    {
+        // Arrange
+        const string outputData = "test output from subworkflow";
+
+        Func<string, IWorkflowContext, CancellationToken, ValueTask> processFunc = (input, context, cancellationToken) => context.YieldOutputAsync(input, cancellationToken);
+
+        ExecutorBinding subworkflowExecutor = processFunc.BindAsExecutor("SubworkflowExecutor", threadsafe: true);
+        Workflow subworkflow = new WorkflowBuilder(subworkflowExecutor)
+            .WithOutputFrom(subworkflowExecutor)
+            .Build();
+
+        ExecutorBinding workflowHostExecutor = subworkflow.BindAsExecutor(
+            "HostExecutor",
+            new ExecutorOptions
+            {
+                AutoSendMessageHandlerResultObject = false,
+                AutoYieldOutputHandlerResultObject = true
+            });
+
+        Func<string, string, ValueTask<Executor>> createOrchestrator = (id, _) => new(new OrchestratorExecutor(id));
+        ExecutorBinding orchestrator = createOrchestrator.BindExecutor();
+
+        Workflow workflow = new WorkflowBuilder(orchestrator)
+            .AddEdge(orchestrator, workflowHostExecutor)
+            .AddEdge(workflowHostExecutor, orchestrator)
+            .WithOutputFrom(workflowHostExecutor)
+            .Build();
+
+        // Act
+        Run workflowRun = await InProcessExecution.RunAsync(workflow, outputData);
+
+        // Assert
+        RunStatus status = await workflowRun.GetStatusAsync();
+        status.Should().Be(RunStatus.Idle);
+
+        List<WorkflowOutputEvent> outputEvents = workflowRun.OutgoingEvents
+            .OfType<WorkflowOutputEvent>()
+            .ToList();
+
+        outputEvents.Should().HaveCount(1, "the workflow should produce exactly one output event");
+        outputEvents[0].As<string>().Should().Be(outputData, "the output should match the input data");
+    }
+
+    [Fact]
+    public async Task WorkflowHostExecutor_WithAutoSendMessageHandlerResultObjectTrue_SendsMessageNotYield()
+    {
+        // Arrange
+        const string outputData = "test output from subworkflow";
+
+        Func<string, IWorkflowContext, CancellationToken, ValueTask> processFunc = (input, context, cancellationToken) => context.YieldOutputAsync(input, cancellationToken);
+
+        ExecutorBinding subworkflowExecutor = processFunc.BindAsExecutor("SubworkflowExecutor", threadsafe: true);
+        Workflow subworkflow = new WorkflowBuilder(subworkflowExecutor)
+            .WithOutputFrom(subworkflowExecutor)
+            .Build();
+
+        ExecutorBinding workflowHostExecutor = subworkflow.BindAsExecutor(
+            "HostExecutor",
+            new ExecutorOptions
+            {
+                AutoSendMessageHandlerResultObject = true,
+                AutoYieldOutputHandlerResultObject = false
+            });
+
+        Func<string, string, ValueTask<Executor>> createOrchestrator = (id, _) => new(new OrchestratorExecutor(id));
+        ExecutorBinding orchestrator = createOrchestrator.BindExecutor();
+
+        Workflow workflow = new WorkflowBuilder(orchestrator)
+            .AddEdge(orchestrator, workflowHostExecutor)
+            .AddEdge(workflowHostExecutor, orchestrator)
+            .WithOutputFrom(orchestrator)
+            .Build();
+
+        // Act
+        Run workflowRun = await InProcessExecution.RunAsync(workflow, outputData);
+
+        // Assert
+        RunStatus status = await workflowRun.GetStatusAsync();
+        status.Should().Be(RunStatus.Idle);
+
+        List<WorkflowOutputEvent> outputEvents = workflowRun.OutgoingEvents
+            .OfType<WorkflowOutputEvent>()
+            .ToList();
+
+        // With AutoSendMessageHandlerResultObject, the output is sent as a message back to orchestrator, which yields it
+        outputEvents.Should().HaveCount(1, "the workflow should produce exactly one output event");
+        outputEvents[0].As<string>().Should().Be(outputData, "the output should match the input data");
+    }
+
+    [Fact]
+    public async Task WorkflowHostExecutor_WithBothOptionsFalse_DoesNotPropagate()
+    {
+        // Arrange
+        const string outputData = "test output from subworkflow";
+
+        Func<string, IWorkflowContext, CancellationToken, ValueTask> processFunc = (input, context, cancellationToken) => context.YieldOutputAsync(input, cancellationToken);
+
+        ExecutorBinding subworkflowExecutor = processFunc.BindAsExecutor("SubworkflowExecutor", threadsafe: true);
+        Workflow subworkflow = new WorkflowBuilder(subworkflowExecutor)
+            .WithOutputFrom(subworkflowExecutor)
+            .Build();
+
+        ExecutorBinding workflowHostExecutor = subworkflow.BindAsExecutor(
+            "HostExecutor",
+            new ExecutorOptions
+            {
+                AutoSendMessageHandlerResultObject = false,
+                AutoYieldOutputHandlerResultObject = false
+            });
+
+        Func<string, string, ValueTask<Executor>> createOrchestrator = (id, _) => new(new OrchestratorExecutor(id));
+        ExecutorBinding orchestrator = createOrchestrator.BindExecutor();
+
+        Workflow workflow = new WorkflowBuilder(orchestrator)
+            .AddEdge(orchestrator, workflowHostExecutor)
+            .AddEdge(workflowHostExecutor, orchestrator)
+            .WithOutputFrom(orchestrator)
+            .Build();
+
+        // Act
+        Run workflowRun = await InProcessExecution.RunAsync(workflow, outputData);
+
+        // Assert
+        RunStatus status = await workflowRun.GetStatusAsync();
+        status.Should().Be(RunStatus.Idle);
+
+        List<WorkflowOutputEvent> outputEvents = workflowRun.OutgoingEvents
+            .OfType<WorkflowOutputEvent>()
+            .ToList();
+
+        // When both options are false, the subworkflow output is not propagated
+        outputEvents.Should().BeEmpty("no output should be yielded when both options are false");
+    }
+
+    private sealed class OrchestratorExecutor : StatefulExecutor<OrchestratorExecutor.State>
+    {
+        internal sealed class State
+        {
+            public bool ReceivedInput { get; set; }
+            public string? Result { get; set; }
+        }
+
+        public OrchestratorExecutor(string id)
+            : base(id, () => new State(), declareCrossRunShareable: false)
+        {
+        }
+
+        protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
+        {
+            return routeBuilder
+                .AddHandler<string>(this.HandleInputAsync);
+        }
+
+        private async ValueTask HandleInputAsync(string input, IWorkflowContext context, CancellationToken cancellationToken)
+        {
+            await this.InvokeWithStateAsync(ProcessInputAsync, context, cancellationToken: cancellationToken);
+
+            async ValueTask<State?> ProcessInputAsync(State state, IWorkflowContext context, CancellationToken cancellationToken)
+            {
+                if (!state.ReceivedInput)
+                {
+                    state.ReceivedInput = true;
+                    await context.SendMessageAsync(input, cancellationToken: cancellationToken);
+                }
+                else
+                {
+                    state.Result = input;
+                    await context.YieldOutputAsync(input, cancellationToken);
+                }
+
+                return state;
+            }
+        }
+    }
+}