diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs index f5d1d40370..bda7e61a38 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs @@ -53,6 +53,9 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable public ValueTask GetStatusAsync(CancellationToken cancellationToken = default) => this._eventStream.GetStatusAsync(cancellationToken); + internal bool TryGetResponsePortExecutorId(string portId, out string? executorId) + => this._stepRunner.TryGetResponsePortExecutorId(portId, out executorId); + public async IAsyncEnumerable TakeEventStreamAsync(bool blockOnPendingRequest, [EnumeratorCancellation] CancellationToken cancellationToken = default) { //Debug.Assert(breakOnHalt); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/EdgeMap.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/EdgeMap.cs index 8c2162508d..6e3f2af5e6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/EdgeMap.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/EdgeMap.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -95,6 +96,18 @@ internal sealed class EdgeMap return portRunner.ChaseEdgeAsync(new MessageEnvelope(response, ExecutorIdentity.None), this._stepTracer, cancellationToken); } + internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId) + { + if (this._portEdgeRunners.TryGetValue(portId, out ResponseEdgeRunner? portRunner)) + { + executorId = portRunner.ExecutorId; + return true; + } + + executorId = null; + return false; + } + internal async ValueTask> ExportStateAsync() { Dictionary exportedStates = []; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs index 9b8c3c460c..8de0dbd5e2 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs @@ -19,6 +19,7 @@ internal interface ISuperStepRunner bool HasUnprocessedMessages { get; } ValueTask EnqueueResponseAsync(ExternalResponse response, CancellationToken cancellationToken = default); + bool TryGetResponsePortExecutorId(string portId, out string? executorId); ValueTask IsValidInputTypeAsync(CancellationToken cancellationToken = default); ValueTask EnqueueMessageAsync(T message, CancellationToken cancellationToken = default); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs index 2a61f80ced..f93b09ddf3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs @@ -160,6 +160,8 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests; bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions; + bool ISuperStepRunner.TryGetResponsePortExecutorId(string portId, out string? executorId) + => this.RunContext.TryGetResponsePortExecutorId(portId, out executorId); public bool IsCheckpointingEnabled => this.RunContext.IsCheckpointingEnabled; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index eda7b90a80..f0bb8cac26 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -296,6 +296,9 @@ internal sealed class InProcessRunnerContext : IRunnerContext return this._externalRequests.TryRemove(requestId, out _); } + internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId) + => this._edgeMap.TryGetResponsePortExecutorId(portId, out executorId); + private IEventSink OutgoingEvents { get; } internal StateManager StateManager { get; } = new(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIContentExternalHandler.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIContentExternalHandler.cs index c19fab62bb..15203fd5dc 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIContentExternalHandler.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIContentExternalHandler.cs @@ -16,10 +16,12 @@ internal sealed class AIContentExternalHandler _pendingRequests = new(); public AIContentExternalHandler(ref ProtocolBuilder protocolBuilder, string portId, bool intercepted, Func handler) { + this._portId = portId; PortBinding? portBinding = null; protocolBuilder = protocolBuilder.ConfigureRoutes(routeBuilder => ConfigureRoutes(routeBuilder, out portBinding)); this._portBinding = portBinding; @@ -65,7 +67,7 @@ internal sealed class AIContentExternalHandler this._portBinding == null; + private string CreateExternalRequestId(string requestId) => $"{this._portId.Length}:{this._portId}:{requestId}"; + private static string MakeKey(string id) => $"{id}_PendingRequests"; public async ValueTask OnCheckpointingAsync(string id, IWorkflowContext context, CancellationToken cancellationToken = default) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs index b479cae75e..c7a19380b2 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs @@ -60,6 +60,9 @@ public sealed class StreamingRun : CheckpointableRunBase, IAsyncDisposable internal ValueTask TrySendMessageUntypedAsync(object message, Type? declaredType = null) => this._runHandle.EnqueueMessageUntypedAsync(message, declaredType); + internal bool TryGetResponsePortExecutorId(string portId, out string? executorId) + => this._runHandle.TryGetResponsePortExecutorId(portId, out executorId); + /// /// Asynchronously streams workflow events as they occur during workflow execution. /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs index c26c9b1985..76c66df6ec 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs @@ -26,9 +26,9 @@ internal sealed class WorkflowSession : AgentSession private InMemoryCheckpointManager? _inMemoryCheckpointManager; /// - /// Tracks pending external requests by their content ID (e.g., - /// or ). This mapping enables converting incoming response - /// content back to when resuming a workflow from a checkpoint. + /// Tracks pending external requests by their workflow-facing request ID. + /// This mapping enables converting incoming response content back to + /// when resuming a workflow from a checkpoint. /// /// /// @@ -176,8 +176,8 @@ internal sealed class WorkflowSession : AgentSession .ConfigureAwait(false); // Process messages: convert response content to ExternalResponse, send regular messages as-is - bool hasMatchedExternalResponses = await this.SendMessagesWithResponseConversionAsync(run, messages).ConfigureAwait(false); - return new ResumeRunResult(run, hasMatchedExternalResponses: hasMatchedExternalResponses); + ResumeDispatchInfo dispatchInfo = await this.SendMessagesWithResponseConversionAsync(run, messages).ConfigureAwait(false); + return new ResumeRunResult(run, dispatchInfo); } StreamingRun newRun = await this._executionEnvironment @@ -194,15 +194,15 @@ internal sealed class WorkflowSession : AgentSession /// to ExternalResponse when there's a matching pending request. /// /// - /// if any external responses were sent; otherwise, . + /// Structured information about how resume content was dispatched. /// - private async ValueTask SendMessagesWithResponseConversionAsync(StreamingRun run, List messages) + private async ValueTask SendMessagesWithResponseConversionAsync(StreamingRun run, List messages) { List regularMessages = []; // Responses are deferred until after regular messages are queued so response handlers // can merge buffered regular content in the same continuation turn. - List<(ExternalResponse Response, string? ContentId)> externalResponses = []; - bool hasMatchedExternalResponses = false; + List<(ExternalResponse Response, string RequestId)> externalResponses = []; + bool hasMatchedResponseForStartExecutor = false; // Tracks content IDs already matched to pending requests within this invocation, // preventing duplicate responses for the same ID from being sent to the workflow engine. @@ -225,8 +225,16 @@ internal sealed class WorkflowSession : AgentSession if (contentId != null && this.TryGetPendingRequest(contentId) is ExternalRequest pendingRequest) { - externalResponses.Add((pendingRequest.CreateResponse(content), contentId)); - (matchedContentIds ??= new(StringComparer.OrdinalIgnoreCase)).Add(contentId); + if (!run.TryGetResponsePortExecutorId(pendingRequest.PortInfo.PortId, out string? responseExecutorId)) + { + throw new InvalidOperationException( + $"Matched pending request '{pendingRequest.RequestId}' refers to unknown response port '{pendingRequest.PortInfo.PortId}'."); + } + + AIContent normalizedResponseContent = NormalizeResponseContentForDelivery(content, pendingRequest); + externalResponses.Add((pendingRequest.CreateResponse(normalizedResponseContent), pendingRequest.RequestId)); + (matchedContentIds ??= new(StringComparer.Ordinal)).Add(contentId); + hasMatchedResponseForStartExecutor |= string.Equals(responseExecutorId, this._workflow.StartExecutorId, StringComparison.Ordinal); } else { @@ -243,28 +251,54 @@ internal sealed class WorkflowSession : AgentSession } // Send regular messages first so response handlers can merge them with responses. - if (regularMessages.Count > 0) + bool hasRegularMessages = regularMessages.Count > 0; + if (hasRegularMessages) { await run.TrySendMessageAsync(regularMessages).ConfigureAwait(false); } // Send external responses after regular messages. - foreach ((ExternalResponse response, string? contentId) in externalResponses) + bool hasMatchedExternalResponses = false; + foreach ((ExternalResponse response, string requestId) in externalResponses) { await run.SendResponseAsync(response).ConfigureAwait(false); hasMatchedExternalResponses = true; - - if (contentId is string id) - { - this.RemovePendingRequest(id); - } + this.RemovePendingRequest(requestId); } - return hasMatchedExternalResponses; + return new ResumeDispatchInfo( + hasRegularMessages, + hasMatchedExternalResponses, + hasMatchedResponseForStartExecutor); } /// - /// Gets the content ID from response content types. + /// Creates the workflow-facing request content surfaced in response updates. + /// + private static AIContent CreateRequestContentForDelivery(ExternalRequest request) => request switch + { + ExternalRequest externalRequest when externalRequest.TryGetDataAs(out FunctionCallContent? functionCallContent) + => CloneFunctionCallContent(functionCallContent, externalRequest.RequestId), + ExternalRequest externalRequest when externalRequest.TryGetDataAs(out UserInputRequestContent? userInputRequestContent) + => CloneUserInputRequestContent(userInputRequestContent, externalRequest.RequestId), + ExternalRequest externalRequest + => externalRequest.ToFunctionCall(), + }; + + /// + /// Rewrites workflow-facing response content back to the original agent-owned content ID. + /// + private static AIContent NormalizeResponseContentForDelivery(AIContent content, ExternalRequest request) => content switch + { + FunctionResultContent functionResultContent when request.TryGetDataAs(out FunctionCallContent? functionCallContent) + => CloneFunctionResultContent(functionResultContent, functionCallContent.CallId), + UserInputResponseContent userInputResponseContent when request.TryGetDataAs(out UserInputRequestContent? userInputRequestContent) + => CloneUserInputResponseContent(userInputResponseContent, userInputRequestContent.Id), + _ => content, + }; + + /// + /// Gets the workflow-facing request ID from response content types. /// private static string? GetResponseContentId(AIContent content) => content switch { @@ -274,22 +308,21 @@ internal sealed class WorkflowSession : AgentSession }; /// - /// Tries to get a pending request by content ID. + /// Tries to get a pending request by workflow-facing request ID. /// - private ExternalRequest? TryGetPendingRequest(string contentId) => - this._pendingRequests.TryGetValue(contentId, out ExternalRequest? request) ? request : null; + private ExternalRequest? TryGetPendingRequest(string requestId) => + this._pendingRequests.TryGetValue(requestId, out ExternalRequest? request) ? request : null; /// - /// Adds a pending request indexed by content ID. + /// Adds a pending request indexed by workflow-facing request ID. /// - private void AddPendingRequest(string contentId, ExternalRequest request) => - this._pendingRequests[contentId] = request; + private void AddPendingRequest(string requestId, ExternalRequest request) => this._pendingRequests[requestId] = request; /// - /// Removes a pending request by content ID. + /// Removes a pending request by workflow-facing request ID. /// - private void RemovePendingRequest(string contentId) => - this._pendingRequests.Remove(contentId); + private void RemovePendingRequest(string requestId) => + this._pendingRequests.Remove(requestId); internal async IAsyncEnumerable InvokeStageAsync( @@ -306,10 +339,11 @@ internal sealed class WorkflowSession : AgentSession await using StreamingRun run = resumeResult.Run; #pragma warning restore CA2007 - // Send a TurnToken only when no external responses were delivered. - // External response handlers already drive continuation turns and can merge - // buffered regular messages, so an extra TurnToken would cause a redundant turn. - if (!resumeResult.HasMatchedExternalResponses) + ResumeDispatchInfo dispatchInfo = resumeResult.DispatchInfo; + bool shouldSendTurnToken = + !dispatchInfo.HasMatchedExternalResponses + || (dispatchInfo.HasRegularMessages && !dispatchInfo.HasMatchedResponseForStartExecutor); + if (shouldSendTurnToken) { await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); } @@ -324,18 +358,11 @@ internal sealed class WorkflowSession : AgentSession break; case RequestInfoEvent requestInfo: - (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) - }; + AIContent requestContent = CreateRequestContentForDelivery(requestInfo.Request); - // Track the pending request so we can convert incoming responses back to ExternalResponse - if (contentId != null) - { - this.AddPendingRequest(contentId, requestInfo.Request); - } + // Track the pending request so we can convert incoming responses back to ExternalResponse. + // External callers respond using the workflow-facing request ID, which is always RequestId. + this.AddPendingRequest(requestInfo.Request.RequestId, requestInfo.Request); AgentResponseUpdate update = this.CreateUpdate(this.LastResponseId, evt, requestContent); yield return update; @@ -420,16 +447,113 @@ internal sealed class WorkflowSession : AgentSession /// The streaming run that was created or resumed. public StreamingRun Run { get; } - /// Whether any external responses (e.g., ) were delivered. - public bool HasMatchedExternalResponses { get; } + /// How resume-time content was dispatched into the workflow runtime. + public ResumeDispatchInfo DispatchInfo { get; } - public ResumeRunResult(StreamingRun run, bool hasMatchedExternalResponses = false) + public ResumeRunResult(StreamingRun run, ResumeDispatchInfo dispatchInfo = default) { this.Run = Throw.IfNull(run); - this.HasMatchedExternalResponses = hasMatchedExternalResponses; + this.DispatchInfo = dispatchInfo; } } + /// + /// Captures how resumed input was split across regular-message and external-response delivery paths. + /// + private readonly struct ResumeDispatchInfo + { + public ResumeDispatchInfo(bool hasRegularMessages, bool hasMatchedExternalResponses, bool hasMatchedResponseForStartExecutor) + { + this.HasRegularMessages = hasRegularMessages; + this.HasMatchedExternalResponses = hasMatchedExternalResponses; + this.HasMatchedResponseForStartExecutor = hasMatchedResponseForStartExecutor; + } + + public bool HasRegularMessages { get; } + + public bool HasMatchedExternalResponses { get; } + + public bool HasMatchedResponseForStartExecutor { get; } + } + + /// + /// Clones a with a workflow-facing call ID. + /// + private static FunctionCallContent CloneFunctionCallContent(FunctionCallContent content, string callId) + { + FunctionCallContent clone = new(callId, content.Name, content.Arguments) + { + Exception = content.Exception, + InformationalOnly = content.InformationalOnly, + }; + + return CopyContentMetadata(content, clone); + } + + /// + /// Clones a with an agent-owned call ID. + /// + private static FunctionResultContent CloneFunctionResultContent(FunctionResultContent content, string callId) + { + FunctionResultContent clone = new(callId, content.Result) + { + Exception = content.Exception, + }; + + return CopyContentMetadata(content, clone); + } + + /// + /// Clones a with a workflow-facing request ID. + /// + private static UserInputRequestContent CloneUserInputRequestContent(UserInputRequestContent content, string id) + { + UserInputRequestContent clone = content switch + { + FunctionApprovalRequestContent functionApprovalRequestContent => + new FunctionApprovalRequestContent(id, functionApprovalRequestContent.FunctionCall), + McpServerToolApprovalRequestContent mcpApprovalRequestContent => + new McpServerToolApprovalRequestContent(id, mcpApprovalRequestContent.ToolCall), + _ => throw new NotSupportedException( + $"Unsupported user input request content type '{content.GetType().Name}' for workflow request ID rewriting."), + }; + + return CopyContentMetadata(content, clone); + } + + /// + /// Clones a with an agent-owned request ID. + /// + private static UserInputResponseContent CloneUserInputResponseContent(UserInputResponseContent content, string id) + { + UserInputResponseContent clone = content switch + { + FunctionApprovalResponseContent functionApprovalResponseContent => + new FunctionApprovalResponseContent(id, functionApprovalResponseContent.Approved, functionApprovalResponseContent.FunctionCall) + { + Reason = functionApprovalResponseContent.Reason, + }, + McpServerToolApprovalResponseContent mcpApprovalResponseContent => + new McpServerToolApprovalResponseContent(id, mcpApprovalResponseContent.Approved), + _ => throw new NotSupportedException( + $"Unsupported user input response content type '{content.GetType().Name}' for workflow response ID rewriting."), + }; + + return CopyContentMetadata(content, clone); + } + + /// + /// Copies shared metadata to a cloned content instance. + /// + private static TContent CopyContentMetadata(AIContent source, TContent target) + where TContent : AIContent + { + target.AdditionalProperties = source.AdditionalProperties; + target.Annotations = source.Annotations; + target.RawRepresentation = source.RawRepresentation; + return target; + } + internal sealed class SessionState( string sessionId, CheckpointInfo? lastCheckpoint, diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs index 1ee9f748a8..013c0eb2ea 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs @@ -85,6 +85,71 @@ internal sealed class RequestEmittingAgent : AIAgent } } +internal sealed class KickoffOnStartExecutor : ChatProtocolExecutor +{ + private static readonly ChatProtocolExecutorOptions s_options = new() + { + AutoSendTurnToken = false, + }; + + private readonly string _downstreamExecutorId; + private readonly string _kickoffInputText; + private readonly string _kickoffMessageText; + private readonly string _regularResumeText; + private readonly string _regularProcessedText; + + public KickoffOnStartExecutor( + string id, + string downstreamExecutorId, + string kickoffInputText, + string kickoffMessageText, + string regularResumeText, + string regularProcessedText) + : base(id, s_options) + { + this._downstreamExecutorId = downstreamExecutorId; + this._kickoffInputText = kickoffInputText; + this._kickoffMessageText = kickoffMessageText; + this._regularResumeText = regularResumeText; + this._regularProcessedText = regularProcessedText; + } + + protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) + { + List textContents = + [ + .. messages + .SelectMany(message => message.Contents.OfType()) + .Select(content => content.Text) + ]; + + if (textContents.Contains(this._kickoffInputText, StringComparer.Ordinal)) + { + await context.SendMessageAsync( + new List { new(ChatRole.User, this._kickoffMessageText) }, + this._downstreamExecutorId, + cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync( + new TurnToken(emitEvents), + this._downstreamExecutorId, + cancellationToken).ConfigureAwait(false); + } + + if (textContents.Contains(this._regularResumeText, StringComparer.Ordinal)) + { + AgentResponseUpdate update = new(ChatRole.Assistant, [new TextContent(this._regularProcessedText)]) + { + CreatedAt = DateTimeOffset.UtcNow, + MessageId = Guid.NewGuid().ToString("N"), + ResponseId = Guid.NewGuid().ToString("N"), + Role = ChatRole.Assistant, + }; + + await context.AddEventAsync(new AgentResponseUpdateEvent(this.Id, update), cancellationToken).ConfigureAwait(false); + } + } +} + public class WorkflowHostSmokeTests { private sealed class AlwaysFailsAIAgent(bool failByThrowing) : AIAgent @@ -193,7 +258,7 @@ public class WorkflowHostSmokeTests // Assert AgentResponseUpdate? updateWithFunctionCall = updates.FirstOrDefault(u => - u.Contents.Any(c => c is FunctionCallContent)); + u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is FunctionCallContent)); updateWithFunctionCall.Should().NotBeNull("a FunctionCallContent should be present in the response updates"); FunctionCallContent retrievedContent = updateWithFunctionCall!.Contents @@ -201,7 +266,8 @@ public class WorkflowHostSmokeTests .Should().ContainSingle() .Which; - retrievedContent.CallId.Should().Be(CallId); + retrievedContent.CallId.Should().NotBe(CallId); + retrievedContent.CallId.Should().EndWith($":{CallId}"); retrievedContent.Name.Should().Be(FunctionName); } @@ -228,7 +294,7 @@ public class WorkflowHostSmokeTests // Assert AgentResponseUpdate? updateWithUserInput = updates.FirstOrDefault(u => - u.Contents.Any(c => c is UserInputRequestContent)); + u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is UserInputRequestContent)); updateWithUserInput.Should().NotBeNull("a UserInputRequestContent should be present in the response updates"); UserInputRequestContent retrievedContent = updateWithUserInput!.Contents @@ -237,7 +303,8 @@ public class WorkflowHostSmokeTests .Which; retrievedContent.Should().BeOfType(); - retrievedContent.Id.Should().Be(RequestId); + retrievedContent.Id.Should().NotBe(RequestId); + retrievedContent.Id.Should().EndWith($":{RequestId}"); } /// @@ -264,16 +331,16 @@ public class WorkflowHostSmokeTests // Assert 1: We should have received a FunctionCallContent AgentResponseUpdate? updateWithRequest = firstCallUpdates.FirstOrDefault(u => - u.Contents.Any(c => c is FunctionCallContent)); + u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is FunctionCallContent)); updateWithRequest.Should().NotBeNull("a FunctionCallContent should be present in the response updates"); FunctionCallContent receivedRequest = updateWithRequest!.Contents .OfType() .First(); - receivedRequest.CallId.Should().Be(CallId); + receivedRequest.CallId.Should().EndWith($":{CallId}"); // Act 2: Send the response back - FunctionResultContent responseContent = new(CallId, "test result"); + FunctionResultContent responseContent = new(receivedRequest.CallId, "test result"); ChatMessage responseMessage = new(ChatRole.Tool, [responseContent]); // Act 2: Run the workflow with the response and capture the resulting updates @@ -284,9 +351,10 @@ public class WorkflowHostSmokeTests secondCallUpdates.Should().NotBeNull("processing the response should produce updates"); secondCallUpdates.Should().NotBeEmpty("processing the response should progress the workflow"); secondCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) .SelectMany(u => u.Contents.OfType()) .Should() - .NotContain(c => c.CallId == CallId, "the original FunctionCallContent request should be cleared after processing the response"); + .NotContain(c => c.CallId == receivedRequest.CallId, "the external FunctionCallContent request should be cleared after processing the response"); } /// @@ -314,16 +382,17 @@ public class WorkflowHostSmokeTests // Assert 1: We should have received a UserInputRequestContent AgentResponseUpdate? updateWithRequest = firstCallUpdates.FirstOrDefault(u => - u.Contents.Any(c => c is UserInputRequestContent)); + u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is UserInputRequestContent)); updateWithRequest.Should().NotBeNull("a UserInputRequestContent should be present in the response updates"); UserInputRequestContent receivedRequest = updateWithRequest!.Contents .OfType() .First(); - receivedRequest.Id.Should().Be(RequestId); + receivedRequest.Id.Should().EndWith($":{RequestId}"); + receivedRequest.Should().BeOfType(); // Act 2: Send the response back - use CreateResponse to get the right response type - UserInputResponseContent responseContent = requestContent.CreateResponse(approved: true); + UserInputResponseContent responseContent = ((McpServerToolApprovalRequestContent)receivedRequest).CreateResponse(approved: true); ChatMessage responseMessage = new(ChatRole.User, [responseContent]); // Act 2: Run the workflow again with the response and capture the updates @@ -331,7 +400,9 @@ public class WorkflowHostSmokeTests // Assert 2: The response should be applied so that the original request is no longer pending secondCallUpdates.Should().NotBeEmpty("handling the user input response should produce follow-up updates"); - bool requestStillPresent = secondCallUpdates.Any(u => u.Contents.OfType().Any(r => r.Id == RequestId)); + bool requestStillPresent = secondCallUpdates.Any(u => + u.RawRepresentation is RequestInfoEvent + && u.Contents.OfType().Any(r => r.Id == receivedRequest.Id)); requestStillPresent.Should().BeFalse("the original UserInputRequestContent should not be re-emitted after its response is processed"); } @@ -363,11 +434,15 @@ public class WorkflowHostSmokeTests session).ToListAsync(); // Assert 1: We should have received a FunctionCallContent + AgentResponseUpdate requestUpdate = firstCallUpdates.First(u => + u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is FunctionCallContent)); + FunctionCallContent emittedRequest = requestUpdate.Contents.OfType().Single(); + firstCallUpdates.Should().Contain(u => u.Contents.Any(c => c is FunctionCallContent), "the first call should emit a FunctionCallContent request"); // Act 2: Send a mixed message containing both the function result AND regular non-response content - FunctionResultContent responseContent = new(CallId, "tool output"); + FunctionResultContent responseContent = new(emittedRequest.CallId, "tool output"); ChatMessage mixedMessage = new(ChatRole.Tool, [responseContent, new TextContent("additional context")]); List secondCallUpdates = await agent.RunStreamingAsync(mixedMessage, session).ToListAsync(); @@ -375,9 +450,10 @@ public class WorkflowHostSmokeTests // Assert 2: The workflow should have processed both parts without errors secondCallUpdates.Should().NotBeEmpty("the mixed message should produce follow-up updates"); secondCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) .SelectMany(u => u.Contents.OfType()) .Should() - .NotContain(c => c.CallId == CallId, "the original FunctionCallContent should be cleared after the response is processed"); + .NotContain(c => c.CallId == emittedRequest.CallId, "the external FunctionCallContent should be cleared after the response is processed"); secondCallUpdates .SelectMany(u => u.Contents.OfType()) .Should() @@ -398,11 +474,14 @@ public class WorkflowHostSmokeTests AgentSession session = await agent.CreateSessionAsync(); List firstCallUpdates = await agent.RunStreamingAsync(new ChatMessage(ChatRole.User, "Start"), session).ToListAsync(); - firstCallUpdates.Should().Contain(u => u.Contents.Any(c => c is FunctionCallContent)); + FunctionCallContent emittedRequest = firstCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) + .SelectMany(u => u.Contents.OfType()) + .Single(); ChatMessage[] resumeMessages = [ - new(ChatRole.Tool, [new FunctionResultContent(CallId, "tool output")]), + new(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]), new(ChatRole.Tool, [new TextContent("extra context in separate message")]) ]; @@ -410,9 +489,10 @@ public class WorkflowHostSmokeTests secondCallUpdates.Should().NotBeEmpty(); secondCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) .SelectMany(u => u.Contents.OfType()) .Should() - .NotContain(c => c.CallId == CallId, "response+regular content split across messages should not re-emit the handled request"); + .NotContain(c => c.CallId == emittedRequest.CallId, "response+regular content split across messages should not re-emit the handled external request"); secondCallUpdates .SelectMany(u => u.Contents.OfType()) .Should() @@ -433,20 +513,82 @@ public class WorkflowHostSmokeTests AgentSession session = await agent.CreateSessionAsync(); List firstCallUpdates = await agent.RunStreamingAsync(new ChatMessage(ChatRole.User, "Start"), session).ToListAsync(); - firstCallUpdates.Should().Contain(u => u.Contents.Any(c => c is FunctionCallContent)); + FunctionCallContent emittedRequest = firstCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) + .SelectMany(u => u.Contents.OfType()) + .Single(); List secondCallUpdates = await agent.RunStreamingAsync( - new ChatMessage(ChatRole.Tool, [new FunctionResultContent(CallId, "tool output")]), + new ChatMessage(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]), session).ToListAsync(); int functionCallCount = secondCallUpdates .Where(u => u.RawRepresentation is RequestInfoEvent) .SelectMany(u => u.Contents.OfType()) - .Count(c => c.CallId == CallId); + .Count(c => c.CallId == emittedRequest.CallId); functionCallCount.Should().Be(1, "a matching external response should not trigger an extra TurnToken-driven turn"); } + [Fact] + public async Task Test_AsAgent_MixedResponseAndRegularMessage_CrossExecutorStartExecutorIsReawakenedAsync() + { + const string StartExecutorId = "start-executor"; + const string KickoffInputText = "Start"; + const string KickoffMessageText = "kickoff downstream"; + const string ResumeRegularText = "resume regular"; + const string ResumeProcessedText = "regular message processed"; + const string CallId = "cross-executor-call-id"; + const string FunctionName = "crossExecutorFunction"; + + RequestEmittingAgent requestAgent = new(new FunctionCallContent(CallId, FunctionName), completeOnResponse: true); + ExecutorBinding requestBinding = requestAgent.BindAsExecutor( + new AIAgentHostOptions { InterceptUnterminatedFunctionCalls = false, EmitAgentUpdateEvents = true }); + + KickoffOnStartExecutor startExecutor = new( + StartExecutorId, + requestBinding.Id, + KickoffInputText, + KickoffMessageText, + ResumeRegularText, + ResumeProcessedText); + ExecutorBinding startBinding = startExecutor.BindExecutor(); + + Workflow workflow = new WorkflowBuilder(startBinding) + .AddEdge>(startBinding, requestBinding, messages => + messages?.Any(message => message.Contents.OfType().Any(content => content.Text == KickoffMessageText)) == true) + .AddEdge(startBinding, requestBinding, _ => true) + .Build(); + AIAgent agent = workflow.AsAIAgent("WorkflowAgent"); + + AgentSession session = await agent.CreateSessionAsync(); + List firstCallUpdates = await agent.RunStreamingAsync( + new ChatMessage(ChatRole.User, KickoffInputText), + session).ToListAsync(); + FunctionCallContent emittedRequest = firstCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) + .SelectMany(u => u.Contents.OfType()) + .Single(); + + ChatMessage[] resumeMessages = + [ + new(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]), + new(ChatRole.User, ResumeRegularText) + ]; + + List secondCallUpdates = await agent.RunStreamingAsync(resumeMessages, session).ToListAsync(); + List textContents = [.. secondCallUpdates.SelectMany(update => update.Contents.OfType()).Select(content => content.Text)]; + + textContents.Should().Contain(ResumeProcessedText, "the start executor should receive an explicit TurnToken when the matched response wakes a different executor"); + textContents.Should().Contain("Request processed", "the matched external response should still be delivered to the downstream request owner"); + secondCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) + .SelectMany(u => u.Contents.OfType()) + .Should() + .NotContain(c => c.CallId == emittedRequest.CallId, "the handled external request should not be re-emitted while waking the start executor"); + secondCallUpdates.SelectMany(u => u.Contents.OfType()).Should().BeEmpty(); + } + [Fact] public async Task Test_AsAgent_UnmatchedResponse_TriggersTurnAndKeepsProgressingAsync() {