mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Add AutoYieldOutputHandlerResultObject support to WorkflowHostExecutor
- Added YieldOutputAsync method to ISuperStepJoinContext interface - Implemented YieldOutputAsync in InProcessRunnerContext - Updated WorkflowHostExecutor to check AutoYieldOutputHandlerResultObject option - Updated TestRunContext to implement YieldOutputAsync - Added unit tests for the new functionality Co-authored-by: lokitoth <6936551+lokitoth@users.noreply.github.com>
This commit is contained in:
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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<WorkflowOutputEvent>().Should().HaveCount(1, "YieldOutputAsync should create one WorkflowOutputEvent");
|
||||
WorkflowOutputEvent? yieldedEvent = testContext.Events.OfType<WorkflowOutputEvent>().FirstOrDefault();
|
||||
yieldedEvent.Should().NotBeNull();
|
||||
yieldedEvent!.SourceId.Should().Be("TestHost");
|
||||
yieldedEvent.As<string>().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<WorkflowOutputEvent>().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<WorkflowOutputEvent>().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<WorkflowOutputEvent>().Should().BeEmpty("YieldOutputAsync should not be called when data is null");
|
||||
testContext.QueuedMessages.Should().BeEmpty("SendMessageAsync should not be called when data is null");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Simple executor for testing that doesn't require type parameters.
|
||||
/// </summary>
|
||||
private sealed class SimpleTestExecutor : Executor
|
||||
{
|
||||
public SimpleTestExecutor(string id) : base(id)
|
||||
{
|
||||
}
|
||||
|
||||
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Testable wrapper for WorkflowHostExecutor that exposes internal methods for testing.
|
||||
/// </summary>
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user