diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs index 76c66df6ec..fcdefcfd22 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs @@ -225,16 +225,16 @@ internal sealed class WorkflowSession : AgentSession if (contentId != null && this.TryGetPendingRequest(contentId) is ExternalRequest pendingRequest) { - if (!run.TryGetResponsePortExecutorId(pendingRequest.PortInfo.PortId, out string? responseExecutorId)) + // For intercepted/complex topologies the port may not be registered in the EdgeMap. + // Treat unknown port as non-start-executor (conservative): TurnToken will still be sent. + 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}'."); + hasMatchedResponseForStartExecutor |= string.Equals(responseExecutorId, this._workflow.StartExecutorId, StringComparison.Ordinal); } 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 { @@ -340,9 +340,14 @@ internal sealed class WorkflowSession : AgentSession #pragma warning restore CA2007 ResumeDispatchInfo dispatchInfo = resumeResult.DispatchInfo; + + // Send a TurnToken to the start executor unless the only activity is an external + // response directed at the start executor itself (which self-emits a TurnToken via + // ContinueTurnAsync). Non-start executors (e.g., RequestInfoExecutor) do not emit + // TurnTokens after processing responses, so the session must always provide one. bool shouldSendTurnToken = !dispatchInfo.HasMatchedExternalResponses - || (dispatchInfo.HasRegularMessages && !dispatchInfo.HasMatchedResponseForStartExecutor); + || !dispatchInfo.HasMatchedResponseForStartExecutor; if (shouldSendTurnToken) { await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowsJsonUtilities.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowsJsonUtilities.cs index 5e9a86fbec..08d1dcbcbb 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowsJsonUtilities.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowsJsonUtilities.cs @@ -90,7 +90,6 @@ internal static partial class WorkflowsJsonUtilities [JsonSerializable(typeof(ChatMessage))] [JsonSerializable(typeof(ExternalRequest))] [JsonSerializable(typeof(ExternalResponse))] - [JsonSerializable(typeof(Dictionary))] [JsonSerializable(typeof(TurnToken))] // Built-in Executor State Types diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs index 013c0eb2ea..0d543acec3 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs @@ -150,6 +150,62 @@ internal sealed class KickoffOnStartExecutor : ChatProtocolExecutor } } +/// +/// A start executor that always emits a response update on every turn, +/// useful for verifying that a TurnToken was delivered by the session. +/// On the first turn (user messages present), it kicks off a downstream executor. +/// +internal sealed class TurnTrackingStartExecutor : ChatProtocolExecutor +{ + private static readonly ChatProtocolExecutorOptions s_options = new() + { + AutoSendTurnToken = false, + }; + + private readonly string _downstreamExecutorId; + private readonly string _activatedMarker; + private int _activationCount; + + /// Gets the number of times this executor has been activated (i.e., called). + public int ActivationCount => this._activationCount; + + public TurnTrackingStartExecutor(string id, string downstreamExecutorId, string activatedMarker) + : base(id, s_options) + { + this._downstreamExecutorId = downstreamExecutorId; + this._activatedMarker = activatedMarker; + } + + protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref this._activationCount); + + // On the first turn, forward user messages and a TurnToken to the downstream executor. + if (messages.Any(m => m.Role == ChatRole.User)) + { + await context.SendMessageAsync( + messages, + this._downstreamExecutorId, + cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync( + new TurnToken(emitEvents), + this._downstreamExecutorId, + cancellationToken).ConfigureAwait(false); + } + + // Always emit a marker to prove this executor was activated. + AgentResponseUpdate update = new(ChatRole.Assistant, [new TextContent(this._activatedMarker)]) + { + 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 @@ -616,4 +672,64 @@ public class WorkflowHostSmokeTests functionCallCount.Should().Be(1, "an unmatched response should be treated as regular input and still drive a TurnToken continuation without workflow errors"); secondCallUpdates.SelectMany(u => u.Contents.OfType()).Should().BeEmpty(); } + + /// + /// Tests that when a resume contains only an external response directed at a non-start executor + /// (no regular messages), the start executor still receives a TurnToken and is activated. + /// This is a regression test for the case where the TurnToken was previously skipped because + /// HasRegularMessages was , leaving the start executor dormant. + /// + [Fact] + public async Task Test_AsAgent_ResponseOnlyToNonStartExecutor_StartExecutorIsStillActivatedAsync() + { + // Arrange + const string StartExecutorId = "start-executor"; + const string ActivatedMarker = "start-executor-activated"; + const string CallId = "response-only-call-id"; + const string FunctionName = "responseOnlyFunction"; + + RequestEmittingAgent requestAgent = new(new FunctionCallContent(CallId, FunctionName), completeOnResponse: true); + ExecutorBinding requestBinding = requestAgent.BindAsExecutor( + new AIAgentHostOptions { InterceptUnterminatedFunctionCalls = false, EmitAgentUpdateEvents = true }); + + TurnTrackingStartExecutor startExecutor = new(StartExecutorId, requestBinding.Id, ActivatedMarker); + ExecutorBinding startBinding = startExecutor.BindExecutor(); + + Workflow workflow = new WorkflowBuilder(startBinding) + .AddEdge>(startBinding, requestBinding, messages => + messages?.Any(m => m.Contents.OfType().Any()) == true) + .AddEdge(startBinding, requestBinding, _ => true) + .Build(); + AIAgent agent = workflow.AsAIAgent("WorkflowAgent"); + + // Act 1: First call triggers the downstream FunctionCallContent request + AgentSession session = await agent.CreateSessionAsync(); + List firstCallUpdates = await agent.RunStreamingAsync( + new ChatMessage(ChatRole.User, "Start"), + session).ToListAsync(); + + FunctionCallContent emittedRequest = firstCallUpdates + .Where(u => u.RawRepresentation is RequestInfoEvent) + .SelectMany(u => u.Contents.OfType()) + .Single(); + + // Act 2: Resume with ONLY the external response (no regular messages) + List secondCallUpdates = await agent.RunStreamingAsync( + new ChatMessage(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]), + session).ToListAsync(); + + // Assert: Both the downstream and start executor should have been activated + List textContents = [.. secondCallUpdates + .SelectMany(u => u.Contents.OfType()) + .Select(c => c.Text)]; + + textContents.Should().Contain("Request processed", + "the downstream executor should process the external response"); + textContents.Should().Contain(ActivatedMarker, + "the start executor should receive a TurnToken and be activated even when resume contains only an external response"); + secondCallUpdates + .SelectMany(u => u.Contents.OfType()) + .Should() + .BeEmpty(); + } }