Handle external input request and response conversion for workflow as agent scenario

2026-06-16 21:04:09 +08:00 · 2026-02-27 14:38:36 -08:00
parent b295a16c0e
commit 345f6165ca
2 changed files with 328 additions and 3 deletions
@@ -25,6 +25,10 @@ internal sealed class WorkflowSession : AgentSession

    private InMemoryCheckpointManager? _inMemoryCheckpointManager;

+    // Key prefix used in StateBag to track pending external requests by their content ID
+    // This enables converting incoming response content back to ExternalResponse when resuming.
+    private const string PendingRequestKeyPrefix = "workflow.pendingRequest:";
+
    internal static bool VerifyCheckpointingConfiguration(IWorkflowExecutionEnvironment executionEnvironment, [NotNullWhen(true)] out InProcessExecutionEnvironment? inProcEnv)
    {
        inProcEnv = null;
@@ -154,7 +158,8 @@ internal sealed class WorkflowSession : AgentSession
                                               cancellationToken)
                            .ConfigureAwait(false);

-            await run.TrySendMessageAsync(messages).ConfigureAwait(false);
+            // Process messages: convert response content to ExternalResponse, send regular messages as-is
+            await this.SendMessagesWithResponseConversionAsync(run, messages, cancellationToken).ConfigureAwait(false);
            return run;
        }

@@ -166,6 +171,116 @@ internal sealed class WorkflowSession : AgentSession
                            .ConfigureAwait(false);
    }

+    /// <summary>
+    /// Sends messages to the run, converting FunctionResultContent and UserInputResponseContent
+    /// to ExternalResponse when there's a matching pending request.
+    /// </summary>
+    private async ValueTask SendMessagesWithResponseConversionAsync(StreamingRun run, List<ChatMessage> messages, CancellationToken cancellationToken)
+    {
+        List<ChatMessage> regularMessages = [];
+
+        foreach (ChatMessage message in messages)
+        {
+            List<AIContent> regularContents = [];
+
+            foreach (AIContent content in message.Contents)
+            {
+                if (this.TryCreateExternalResponse(content) is ExternalResponse response)
+                {
+                    await run.SendResponseAsync(response).ConfigureAwait(false);
+
+                    if (GetResponseContentId(content) is string contentId)
+                    {
+                        this.RemovePendingRequest(contentId);
+                    }
+                }
+                else
+                {
+                    regularContents.Add(content);
+                }
+            }
+
+            if (regularContents.Count > 0)
+            {
+                regularMessages.Add(new ChatMessage(message.Role, regularContents)
+                {
+                    AuthorName = message.AuthorName,
+                    MessageId = message.MessageId,
+                    CreatedAt = message.CreatedAt
+                });
+            }
+        }
+
+        // Send any remaining regular messages
+        if (regularMessages.Count > 0)
+        {
+            await run.TrySendMessageAsync(regularMessages).ConfigureAwait(false);
+        }
+    }
+
+    /// <summary>
+    /// Attempts to create an ExternalResponse from response content (FunctionResultContent or UserInputResponseContent)
+    /// by matching it to a pending request.
+    /// </summary>
+    private ExternalResponse? TryCreateExternalResponse(AIContent content)
+    {
+        string? contentId = GetResponseContentId(content);
+        if (contentId == null)
+        {
+            return null;
+        }
+
+        ExternalRequest? pendingRequest = this.TryGetPendingRequest(contentId);
+        if (pendingRequest == null)
+        {
+            return null;
+        }
+
+        // Create the response data based on content type
+        object? responseData = content switch
+        {
+            FunctionResultContent functionResultContent => functionResultContent,
+            UserInputResponseContent userInputResponseContent => userInputResponseContent,
+            _ => null
+        };
+
+        if (responseData == null)
+        {
+            return null;
+        }
+
+        // Create ExternalResponse using the pending request's port info
+        return new ExternalResponse(pendingRequest.PortInfo, pendingRequest.RequestId, new PortableValue(responseData));
+    }
+
+    /// <summary>
+    /// Gets the content ID from response content types.
+    /// </summary>
+    private static string? GetResponseContentId(AIContent content) => content switch
+    {
+        FunctionResultContent functionResultContent => functionResultContent.CallId,
+        UserInputResponseContent userInputResponseContent => userInputResponseContent.Id,
+        _ => null
+    };
+
+    /// <summary>
+    /// Tries to get a pending request from the state bag by content ID.
+    /// </summary>
+    private ExternalRequest? TryGetPendingRequest(string contentId) =>
+        this.StateBag.GetValue<ExternalRequest>(PendingRequestKeyPrefix + contentId);
+
+    /// <summary>
+    /// Adds a pending request to the state bag.
+    /// </summary>
+    private void AddPendingRequest(string contentId, ExternalRequest request) =>
+        this.StateBag.SetValue(PendingRequestKeyPrefix + contentId, request);
+
+    /// <summary>
+    /// Removes a pending request from the state bag.
+    /// </summary>
+    private void RemovePendingRequest(string contentId) =>
+        this.StateBag.TryRemoveValue(PendingRequestKeyPrefix + contentId);
+
    internal async
    IAsyncEnumerable<AgentResponseUpdate> InvokeStageAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
@@ -192,8 +307,20 @@ internal sealed class WorkflowSession : AgentSession
                        break;

                    case RequestInfoEvent requestInfo:
-                        FunctionCallContent fcContent = requestInfo.Request.ToFunctionCall();
-                        AgentResponseUpdate update = this.CreateUpdate(this.LastResponseId, evt, fcContent);
+                        (AIContent requestContent, string? contentId) = requestInfo.Request switch
+                        {
+                            ExternalRequest externalRequest when externalRequest.TryGetDataAs(out FunctionCallContent? fcc) => (fcc, fcc.CallId),
+                            ExternalRequest externalRequest when externalRequest.TryGetDataAs(out UserInputRequestContent? uic) => (uic, uic.Id),
+                            ExternalRequest externalRequest => ((AIContent)externalRequest.ToFunctionCall(), externalRequest.RequestId)
+                        };
+
+                        // Track the pending request so we can convert incoming responses back to ExternalResponse
+                        if (contentId != null)
+                        {
+                            this.AddPendingRequest(contentId, requestInfo.Request);
+                        }
+
+                        AgentResponseUpdate update = this.CreateUpdate(this.LastResponseId, evt, requestContent);
                        yield return update;
                        break;

@@ -28,6 +28,43 @@ public sealed class ExpectedException : Exception
    }
 }

+/// <summary>
+/// A simple agent that emits a FunctionCallContent or UserInputRequestContent request.
+/// Used to test that RequestInfoEvent handling preserves the original content type.
+/// </summary>
+internal sealed class RequestEmittingAgent : AIAgent
+{
+    private readonly AIContent _requestContent;
+
+    public RequestEmittingAgent(AIContent requestContent)
+    {
+        this._requestContent = requestContent;
+    }
+
+    private sealed class Session : AgentSession
+    {
+        public Session() { }
+    }
+
+    protected override ValueTask<AgentSession> DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
+        => new(new Session());
+
+    protected override ValueTask<AgentSession> CreateSessionCoreAsync(CancellationToken cancellationToken = default)
+        => new(new Session());
+
+    protected override ValueTask<JsonElement> SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
+        => default;
+
+    protected override Task<AgentResponse> RunCoreAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+        => this.RunStreamingAsync(messages, session, options, cancellationToken).ToAgentResponseAsync(cancellationToken);
+
+    protected override async IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        // Emit the request content
+        yield return new AgentResponseUpdate(ChatRole.Assistant, [this._requestContent]);
+    }
+}
+
 public class WorkflowHostSmokeTests
 {
    private sealed class AlwaysFailsAIAgent(bool failByThrowing) : AIAgent
@@ -112,4 +149,165 @@ public class WorkflowHostSmokeTests

        hadErrorContent.Should().BeTrue();
    }
+
+    /// <summary>
+    /// Tests that when a workflow emits a RequestInfoEvent with FunctionCallContent data,
+    /// the AgentResponseUpdate preserves the original FunctionCallContent type.
+    /// Regression test for GitHub issue #3029.
+    /// </summary>
+    [Fact]
+    public async Task Test_AsAgent_FunctionCallContentPreservedInRequestInfoAsync()
+    {
+        // Arrange
+        const string CallId = "test-call-id";
+        const string FunctionName = "testFunction";
+        FunctionCallContent originalContent = new(CallId, FunctionName);
+        RequestEmittingAgent requestAgent = new(originalContent);
+        ExecutorBinding agentBinding = requestAgent.BindAsExecutor(
+            new AIAgentHostOptions { InterceptUnterminatedFunctionCalls = false, EmitAgentUpdateEvents = true });
+        Workflow workflow = new WorkflowBuilder(agentBinding).Build();
+
+        // Act
+        List<AgentResponseUpdate> updates = await workflow.AsAIAgent("WorkflowAgent")
+                                                           .RunStreamingAsync(new ChatMessage(ChatRole.User, "Hello"))
+                                                           .ToListAsync();
+
+        // Assert
+        AgentResponseUpdate? updateWithFunctionCall = updates.FirstOrDefault(u =>
+            u.Contents.Any(c => c is FunctionCallContent));
+
+        updateWithFunctionCall.Should().NotBeNull("a FunctionCallContent should be present in the response updates");
+        FunctionCallContent retrievedContent = updateWithFunctionCall!.Contents
+            .OfType<FunctionCallContent>()
+            .Should().ContainSingle()
+            .Which;
+
+        retrievedContent.CallId.Should().Be(CallId);
+        retrievedContent.Name.Should().Be(FunctionName);
+    }
+
+    /// <summary>
+    /// Tests that when a workflow emits a RequestInfoEvent with UserInputRequestContent data,
+    /// the AgentResponseUpdate preserves the original UserInputRequestContent type.
+    /// Regression test for GitHub issue #3029.
+    /// </summary>
+    [Fact]
+    public async Task Test_AsAgent_UserInputRequestContentPreservedInRequestInfoAsync()
+    {
+        // Arrange
+        const string RequestId = "test-request-id";
+        McpServerToolCallContent mcpCalll = new("call-id", "testToolName", "http://localhost");
+        UserInputRequestContent originalContent = new McpServerToolApprovalRequestContent(RequestId, mcpCalll);
+        RequestEmittingAgent requestAgent = new(originalContent);
+        ExecutorBinding agentBinding = requestAgent.BindAsExecutor(
+            new AIAgentHostOptions { InterceptUserInputRequests = false, EmitAgentUpdateEvents = true });
+        Workflow workflow = new WorkflowBuilder(agentBinding).Build();
+
+        // Act
+        List<AgentResponseUpdate> updates = await workflow.AsAIAgent("WorkflowAgent")
+                                                           .RunStreamingAsync(new ChatMessage(ChatRole.User, "Hello"))
+                                                           .ToListAsync();
+
+        // Assert
+        AgentResponseUpdate? updateWithUserInput = updates.FirstOrDefault(u =>
+            u.Contents.Any(c => c is UserInputRequestContent));
+
+        updateWithUserInput.Should().NotBeNull("a UserInputRequestContent should be present in the response updates");
+        UserInputRequestContent retrievedContent = updateWithUserInput!.Contents
+            .OfType<UserInputRequestContent>()
+            .Should().ContainSingle()
+            .Which;
+
+        retrievedContent.Should().BeOfType<McpServerToolApprovalRequestContent>();
+        retrievedContent.Id.Should().Be(RequestId);
+    }
+
+    /// <summary>
+    /// Tests the full roundtrip: workflow emits a request, external caller responds, workflow processes response.
+    /// </summary>
+    [Fact]
+    public async Task Test_AsAgent_FunctionCallRoundtrip_ResponseIsProcessedAsync()
+    {
+        // Arrange: Create an agent that emits a FunctionCallContent request
+        const string CallId = "roundtrip-call-id";
+        const string FunctionName = "testFunction";
+        FunctionCallContent requestContent = new(CallId, FunctionName);
+        RequestEmittingAgent requestAgent = new(requestContent);
+        ExecutorBinding agentBinding = requestAgent.BindAsExecutor(
+            new AIAgentHostOptions { InterceptUnterminatedFunctionCalls = false, EmitAgentUpdateEvents = true });
+        Workflow workflow = new WorkflowBuilder(agentBinding).Build();
+        AIAgent agent = workflow.AsAIAgent("WorkflowAgent");
+
+        // Act 1: First call - should receive the FunctionCallContent request
+        AgentSession session = await agent.CreateSessionAsync();
+        List<AgentResponseUpdate> firstCallUpdates = await agent.RunStreamingAsync(
+            new ChatMessage(ChatRole.User, "Start"),
+            session).ToListAsync();
+
+        // Assert 1: We should have received a FunctionCallContent
+        AgentResponseUpdate? updateWithRequest = firstCallUpdates.FirstOrDefault(u =>
+            u.Contents.Any(c => c is FunctionCallContent));
+        updateWithRequest.Should().NotBeNull("a FunctionCallContent should be present in the response updates");
+
+        FunctionCallContent receivedRequest = updateWithRequest!.Contents
+            .OfType<FunctionCallContent>()
+            .First();
+        receivedRequest.CallId.Should().Be(CallId);
+
+        // Act 2: Send the response back
+        FunctionResultContent responseContent = new(CallId, "test result");
+        ChatMessage responseMessage = new(ChatRole.Tool, [responseContent]);
+
+        // This should work without throwing - the response should be converted to ExternalResponse
+        // and processed by the workflow
+        Func<Task> sendResponse = () => agent.RunStreamingAsync(responseMessage, session).ToListAsync().AsTask();
+
+        // Assert 2: The response should be accepted without error
+        await sendResponse.Should().NotThrowAsync("the response should be converted to ExternalResponse and processed");
+    }
+
+    /// <summary>
+    /// Tests the full roundtrip for UserInputRequestContent: workflow emits request, external caller responds.
+    /// Verifying inbound UserInputResponseContent conversion.
+    /// </summary>
+    [Fact]
+    public async Task Test_AsAgent_UserInputRoundtrip_ResponseIsProcessedAsync()
+    {
+        // Arrange: Create an agent that emits a UserInputRequestContent request
+        const string RequestId = "roundtrip-request-id";
+        McpServerToolCallContent mcpCall = new("mcp-call-id", "testMcpTool", "http://localhost");
+        McpServerToolApprovalRequestContent requestContent = new(RequestId, mcpCall);
+        RequestEmittingAgent requestAgent = new(requestContent);
+        ExecutorBinding agentBinding = requestAgent.BindAsExecutor(
+            new AIAgentHostOptions { InterceptUserInputRequests = false, EmitAgentUpdateEvents = true });
+        Workflow workflow = new WorkflowBuilder(agentBinding).Build();
+        AIAgent agent = workflow.AsAIAgent("WorkflowAgent");
+
+        // Act 1: First call - should receive the UserInputRequestContent request
+        AgentSession session = await agent.CreateSessionAsync();
+        List<AgentResponseUpdate> firstCallUpdates = await agent.RunStreamingAsync(
+            new ChatMessage(ChatRole.User, "Start"),
+            session).ToListAsync();
+
+        // Assert 1: We should have received a UserInputRequestContent
+        AgentResponseUpdate? updateWithRequest = firstCallUpdates.FirstOrDefault(u =>
+            u.Contents.Any(c => c is UserInputRequestContent));
+        updateWithRequest.Should().NotBeNull("a UserInputRequestContent should be present in the response updates");
+
+        UserInputRequestContent receivedRequest = updateWithRequest!.Contents
+            .OfType<UserInputRequestContent>()
+            .First();
+        receivedRequest.Id.Should().Be(RequestId);
+
+        // Act 2: Send the response back - use CreateResponse to get the right response type
+        UserInputResponseContent responseContent = requestContent.CreateResponse(approved: true);
+        ChatMessage responseMessage = new(ChatRole.User, [responseContent]);
+
+        // This should work without throwing - the response should be converted to ExternalResponse
+        // and processed by the workflow
+        Func<Task> sendResponse = () => agent.RunStreamingAsync(responseMessage, session).ToListAsync().AsTask();
+
+        // Assert 2: The response should be accepted without error
+        await sendResponse.Should().NotThrowAsync("the response should be converted to ExternalResponse and processed");
+    }
 }