diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs index 2dcbe8e87a0..d7a1d5f7157 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs @@ -1,6 +1,8 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.AI; @@ -22,19 +24,56 @@ public static async ValueTask InvokeAgentAsync( { IAsyncEnumerable agentUpdates = agentProvider.InvokeAgentAsync(agentName, null, conversationId, inputMessages, inputArguments, cancellationToken); - // Determine whether the target conversation is the workflow conversation - // (used below to decide whether to mirror messages into the workflow conversation - // when an agent runs against a different conversation). The caller's autoSend - // value is honored as-is — when the workflow.yaml specifies autoSend: false the - // raw agent output must not be streamed to the caller, even when the agent is - // running on the workflow conversation. + // Foundry managed workflows treat responses produced on the workflow conversation + // as workflow output even when autoSend is explicitly false. Preserve that direct-run + // contract here. Workflow.AsAIAgent separately removes matching streamed/completed + // message duplicates at its hosting boundary. bool isWorkflowConversation = context.IsWorkflowConversation(conversationId, out string? workflowConversationId); + autoSend |= isWorkflowConversation; - // Process the agent response updates. + // Assign stable IDs to content-bearing chat updates before emitting and aggregating them. + // Contentless updates may carry only metadata and must not become empty messages. List updates = []; + string? generatedMessageId = null; + string? generatedMessageResponseId = null; + ChatRole? generatedMessageRole = null; await foreach (AgentResponseUpdate update in agentUpdates.ConfigureAwait(false)) { - await AssignConversationIdAsync(((ChatResponseUpdate?)update.RawRepresentation)?.ConversationId).ConfigureAwait(false); + await AssignConversationIdAsync((update.RawRepresentation as ChatResponseUpdate)?.ConversationId).ConfigureAwait(false); + + if (string.IsNullOrWhiteSpace(update.MessageId)) + { + bool hasContent = + update.Contents.Any( + content => content is not TextContent textContent || !string.IsNullOrEmpty(textContent.Text)); + if (hasContent) + { + if (generatedMessageId is null + || (generatedMessageResponseId is not null + && update.ResponseId is not null + && !string.Equals(generatedMessageResponseId, update.ResponseId, StringComparison.Ordinal)) + || (generatedMessageRole is not null + && update.Role is not null + && generatedMessageRole != update.Role)) + { + generatedMessageId = Guid.NewGuid().ToString("N"); + } + + generatedMessageResponseId = update.ResponseId ?? generatedMessageResponseId; + generatedMessageRole = update.Role ?? generatedMessageRole; + update.MessageId = generatedMessageId; + if (update.RawRepresentation is ChatResponseUpdate rawUpdate) + { + rawUpdate.MessageId = generatedMessageId; + } + } + } + else + { + generatedMessageId = null; + generatedMessageResponseId = null; + generatedMessageRole = null; + } updates.Add(update); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/SendActivityExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/SendActivityExecutor.cs index 66af8d52e3b..3b9794b197f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/SendActivityExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/SendActivityExecutor.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Declarative.Extensions; @@ -21,21 +22,29 @@ internal sealed class SendActivityExecutor(SendActivity model, WorkflowFormulaSt await context.AddEventAsync(new MessageActivityEvent(activityText.Trim()), cancellationToken).ConfigureAwait(false); - ChatMessage message = new(ChatRole.Assistant, activityText); + string responseId = Guid.NewGuid().ToString("N"); + string messageId = Guid.NewGuid().ToString("N"); + ChatMessage message = new(ChatRole.Assistant, activityText) { MessageId = messageId }; // Emit an AgentResponseUpdateEvent so chat protocols (e.g. AsAIAgent) receive the // activity text as streaming chat content. This event is yielded by WorkflowSession // unconditionally, mirroring how AgentProviderExtensions surfaces autoSend agent // updates — without it, SendActivity output is dropped whenever the host runs with // includeWorkflowOutputsInResponse = false (the default). - AgentResponseUpdate update = new(ChatRole.Assistant, activityText) { AuthorName = this.Id }; + AgentResponseUpdate update = + new(ChatRole.Assistant, activityText) + { + AuthorName = this.Id, + MessageId = messageId, + ResponseId = responseId, + }; await context.AddEventAsync(new AgentResponseUpdateEvent(this.Id, update), cancellationToken).ConfigureAwait(false); // Route through YieldOutputAsync so the activity participates in the workflow's // output-filter pipeline. The runner currently special-cases AgentResponse to // produce an AgentResponseEvent identical to the one we'd build by hand, which // is the gated summary surfaced only when includeWorkflowOutputsInResponse = true. - AgentResponse response = new([message]); + AgentResponse response = new([message]) { ResponseId = responseId }; await context.YieldOutputAsync(response, cancellationToken).ConfigureAwait(false); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs index cd20fc4336e..85ff6e40528 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs @@ -229,8 +229,46 @@ await this.EnsureSessionAsync(context, cancellationToken).ConfigureAwait(false), cancellationToken: cancellationToken); List updates = []; + string? generatedMessageId = null; + string? generatedMessageResponseId = null; + ChatRole? generatedMessageRole = null; await foreach (AgentResponseUpdate update in agentStream.ConfigureAwait(false)) { + // Contentless updates may carry only metadata and must not become empty messages. + if (string.IsNullOrWhiteSpace(update.MessageId)) + { + bool hasContent = + update.Contents.Any( + content => content is not TextContent textContent || !string.IsNullOrEmpty(textContent.Text)); + if (hasContent) + { + if (generatedMessageId is null + || (generatedMessageResponseId is not null + && update.ResponseId is not null + && !string.Equals(generatedMessageResponseId, update.ResponseId, StringComparison.Ordinal)) + || (generatedMessageRole is not null + && update.Role is not null + && generatedMessageRole != update.Role)) + { + generatedMessageId = Guid.NewGuid().ToString("N"); + } + + generatedMessageResponseId = update.ResponseId ?? generatedMessageResponseId; + generatedMessageRole = update.Role ?? generatedMessageRole; + update.MessageId = generatedMessageId; + if (update.RawRepresentation is ChatResponseUpdate rawUpdate) + { + rawUpdate.MessageId = generatedMessageId; + } + } + } + else + { + generatedMessageId = null; + generatedMessageResponseId = null; + generatedMessageRole = null; + } + await context.YieldOutputAsync(update, cancellationToken).ConfigureAwait(false); collector.ProcessAgentResponseUpdate(update); updates.Add(update); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs index 3705cdfe123..4e262a6f319 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs @@ -455,6 +455,7 @@ IAsyncEnumerable InvokeStageAsync( #pragma warning restore CA2007 ResumeDispatchInfo dispatchInfo = resumeResult.DispatchInfo; + HashSet<(string ExecutorId, string MessageId)> streamedMessageIds = []; // 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 @@ -485,6 +486,11 @@ AgentResponseUpdate CreateObservabilityUpdate(WorkflowEvent evt) switch (evt) { case AgentResponseUpdateEvent agentUpdate: + string? messageId = agentUpdate.Update.MessageId; + if (!string.IsNullOrWhiteSpace(messageId)) + { + streamedMessageIds.Add((agentUpdate.ExecutorId, messageId)); + } yield return agentUpdate.Update; break; @@ -559,10 +565,29 @@ AgentResponseUpdate CreateObservabilityUpdate(WorkflowEvent evt) // _includeWorkflowOutputInResponse flag is set. Reason being: The user specifies // exclusion of an event by enabling filtering and then _not_ marking an Executor // as an output executor. + bool emittedMessage = false; + bool suppressedStreamedMessage = false; foreach (ChatMessage message in agentResponse.Response.Messages) { + string? completedMessageId = message.MessageId; + bool messageWasStreamed = + !string.IsNullOrWhiteSpace(completedMessageId) + && streamedMessageIds.Contains((agentResponse.ExecutorId, completedMessageId)); + if (messageWasStreamed) + { + suppressedStreamedMessage = true; + continue; + } + + emittedMessage = true; yield return this.CreateUpdate(this.LastResponseId, evt, message); } + if (!emittedMessage && suppressedStreamedMessage) + { + // Preserve the completion event for observability after its correlated + // streamed content has already been forwarded. + yield return CreateObservabilityUpdate(evt); + } break; case WorkflowOutputEvent output: diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Extensions/AgentProviderExtensionsTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Extensions/AgentProviderExtensionsTest.cs index 0393401b481..238fd412245 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Extensions/AgentProviderExtensionsTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Extensions/AgentProviderExtensionsTest.cs @@ -24,8 +24,8 @@ public sealed class AgentProviderExtensionsTest(ITestOutputHelper output) : Work private const string AgentName = "test-agent"; [Fact] - public Task AutoSendFalseOnWorkflowConversationSuppressesResponseEventsAsync() => - this.RunAsync(autoSend: false, conversationId: WorkflowConversationId, expectResponseEvents: false); + public Task AutoSendFalseOnWorkflowConversationStillEmitsResponseEventsAsync() => + this.RunAsync(autoSend: false, conversationId: WorkflowConversationId, expectResponseEvents: true); [Fact] public Task AutoSendTrueOnWorkflowConversationEmitsResponseEventsAsync() => @@ -58,8 +58,13 @@ private async Task RunAsync( MockAgentProvider mockProvider = new(); AgentResponseUpdate[] updates = [ - new(ChatRole.Assistant, "hello "), - new(ChatRole.Assistant, "world"), + new(new ChatResponseUpdate { MessageId = "" }), + new(new ChatResponseUpdate(ChatRole.Assistant, "hello ") { MessageId = " " }), + new(new ChatResponseUpdate(ChatRole.Assistant, "world") { MessageId = "", Role = null }), + new(ChatRole.Assistant, "!") + { + RawRepresentation = new object(), + }, ]; mockProvider .Setup(p => p.InvokeAgentAsync( @@ -109,6 +114,18 @@ await mockProvider.Object.InvokeAgentAsync( { Assert.Equal(updates.Length, updateEventCount); Assert.Equal(1, responseEventCount); + + AgentResponseUpdateEvent[] updateEvents = events.OfType().ToArray(); + Assert.Equal("", updateEvents[0].Update.MessageId); + + string messageId = Assert.IsType(updateEvents[1].Update.MessageId); + Assert.NotEmpty(messageId); + Assert.All(updateEvents.Skip(1), updateEvent => Assert.Equal(messageId, updateEvent.Update.MessageId)); + + AgentResponseEvent responseEvent = Assert.Single(events.OfType()); + ChatMessage responseMessage = Assert.Single(responseEvent.Response.Messages); + Assert.Equal(messageId, responseMessage.MessageId); + Assert.Equal("hello world!", responseMessage.Text); } else { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/InvokeAzureAgentExecutorTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/InvokeAzureAgentExecutorTest.cs index 94f5a190e64..b4f80f97c90 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/InvokeAzureAgentExecutorTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/InvokeAzureAgentExecutorTest.cs @@ -23,6 +23,28 @@ public void InvokeAzureAgentThrowsWhenModelInvalid() => // Arrange, Act & Assert Assert.Throws(() => new InvokeAzureAgentExecutor(new InvokeAzureAgent(), new CapturingAgentProvider("text"), this.State)); + [Theory] + [InlineData(null, true)] + [InlineData(true, true)] + [InlineData(false, false)] + public async Task AutoSendDefaultsToTrueAndHonorsExplicitValueAsync(bool? autoSend, bool expectResponseEvents) + { + // Arrange + this.State.InitializeSystem(); + CapturingAgentProvider provider = new("response"); + InvokeAzureAgent model = this.CreateAutoSendModel( + nameof(AutoSendDefaultsToTrueAndHonorsExplicitValueAsync), + autoSend); + + // Act + WorkflowEvent[] events = + await this.ExecuteAsync(new InvokeAzureAgentExecutor(model, provider, this.State), isDiscrete: false); + + // Assert + Assert.Equal(expectResponseEvents ? 1 : 0, events.OfType().Count()); + Assert.Equal(expectResponseEvents ? 1 : 0, events.OfType().Count()); + } + #region Input argument binding [Fact] @@ -306,6 +328,30 @@ private InvokeAzureAgent CreateModel( return AssignParent(builder); } + private InvokeAzureAgent CreateAutoSendModel(string displayName, bool? autoSend) + { + AzureAgentOutput.Builder outputBuilder = new(); + if (autoSend.HasValue) + { + outputBuilder.AutoSend = new BoolExpression.Builder(BoolExpression.Literal(autoSend.Value)); + } + + InvokeAzureAgent.Builder builder = + new() + { + Id = this.CreateActionId(), + DisplayName = this.FormatDisplayName(displayName), + Agent = + new AzureAgentUsage.Builder + { + Name = new StringExpression.Builder(StringExpression.Literal("BrainAutoSend")), + }, + Output = outputBuilder, + }; + + return AssignParent(builder); + } + /// /// Minimal that returns a single configured text response and /// captures the input arguments supplied to . diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/SendActivityExecutorTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/SendActivityExecutorTest.cs index 1500c5f4388..bfddd1b8f0d 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/SendActivityExecutorTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/ObjectModel/SendActivityExecutorTest.cs @@ -37,6 +37,10 @@ public async Task CaptureActivityAsync() ChatMessage message = Assert.Single(agentEvent.Response.Messages); Assert.Equal(ChatRole.Assistant, message.Role); Assert.Equal("Test activity message", message.Text); + + AgentResponseUpdateEvent updateEvent = Assert.Single(events.OfType()); + Assert.Equal(agentEvent.Response.ResponseId, updateEvent.Update.ResponseId); + Assert.Equal(message.MessageId, updateEvent.Update.MessageId); } private SendActivity CreateModel(string displayName, string activityMessage, string? summary = null) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AIAgentHostExecutorTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AIAgentHostExecutorTests.cs index 5b70be59cb0..e5c45ebe4e0 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AIAgentHostExecutorTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AIAgentHostExecutorTests.cs @@ -64,10 +64,77 @@ public async Task Test_AgentHostExecutor_EmitsResponseIFFConfiguredAsync(bool ex CheckResponseEventsAgainstTestMessages(updates, expectingResponse: executorSetting, agent.GetDescriptiveId()); } + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData(" ")] + public async Task Test_AgentHostExecutor_AssignsStableMessageIdToContentfulStreamingUpdatesAsync(string? missingMessageId) + { + // Arrange + TestRunContext testContext = new(); + AIAgentHostExecutor executor = + new( + new MissingMessageIdAgent(missingMessageId), + new() + { + EmitAgentUpdateEvents = true, + EmitAgentResponseEvents = true, + }); + testContext.ConfigureExecutor(executor); + + // Act + await executor.TakeTurnAsync(new(), testContext.BindWorkflowContext(executor.Id)); + + // Assert + AgentResponseUpdateEvent[] updateEvents = testContext.Events.OfType().ToArray(); + updateEvents.Should().HaveCount(3); + updateEvents[0].Update.MessageId.Should().BeEmpty(); + + string? messageId = updateEvents[1].Update.MessageId; + messageId.Should().NotBeNullOrEmpty(); + updateEvents.Skip(1).Should().OnlyContain(updateEvent => updateEvent.Update.MessageId == messageId); + + AgentResponseEvent responseEvent = testContext.Events.OfType().Should().ContainSingle().Subject; + ChatMessage responseMessage = responseEvent.Response.Messages.Should().ContainSingle().Subject; + responseMessage.MessageId.Should().Be(messageId); + responseMessage.Text.Should().Be("hello world"); + } + private static ChatMessage UserMessage => new(ChatRole.User, "Hello from User!") { AuthorName = "User" }; private static ChatMessage AssistantMessage => new(ChatRole.Assistant, "Hello from Assistant!") { AuthorName = "User" }; private static ChatMessage TestAgentMessage => new(ChatRole.Assistant, $"Hello from {TestAgentName}!") { AuthorName = TestAgentName }; + private sealed class MissingMessageIdAgent(string? messageId) : TestReplayAgent + { + protected override async IAsyncEnumerable RunCoreStreamingAsync( + IEnumerable messages, + AgentSession? session = null, + AgentRunOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + yield return new AgentResponseUpdate( + new ChatResponseUpdate + { + MessageId = "", + ResponseId = "response-id", + }); + yield return new AgentResponseUpdate( + new ChatResponseUpdate(ChatRole.Assistant, "hello ") + { + MessageId = messageId, + ResponseId = "response-id", + }); + yield return new AgentResponseUpdate(ChatRole.Assistant, "world") + { + MessageId = messageId, + ResponseId = "response-id", + Role = null, + RawRepresentation = new object(), + }; + await Task.CompletedTask; + } + } + [Theory] [InlineData(true, true, false, false)] [InlineData(true, true, false, true)] diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs index 12806a98c23..92ebc5915d8 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs @@ -896,8 +896,9 @@ public async Task Test_WorkflowHostAgent_IntermediateAgentResponseForwardedInStr // the surfaced event for consumers that care to distinguish. List updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: false); - updates.Any(u => u.RawRepresentation is AgentResponseEvent are && are.IsIntermediate() && u.Text == InterText) - .Should().BeTrue("AgentResponseEvent is forwarded under Futures-on regardless of the include flag"); + updates.Count(u => u.Text == InterText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent are && are.IsIntermediate() && u.Contents.Count == 0) + .Should().BeTrue("the completion event remains observable without duplicating streamed text"); } [Fact] @@ -914,8 +915,25 @@ public async Task Test_WorkflowHostAgent_TerminalAgentResponseForwardedUnconditi // asymmetry between AgentResponse and AgentResponseUpdate is gone under Futures-on. List updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: false); - updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText) - .Should().BeTrue("terminal AgentResponseEvent is forwarded under Futures-on regardless of the include flag"); + updates.Count(u => u.Text == FinalText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Contents.Count == 0) + .Should().BeTrue("the completion event remains observable without duplicating streamed text"); + } + + [Fact] + public async Task Test_WorkflowHostAgent_EmptyAgentResponseDoesNotCreateObservabilityUpdateAsync() + { + using Futures.FuturesScope _ = new(enabled: true); + TestReplayAgent agent = new(new List()); + ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true }); + Workflow workflow = new WorkflowBuilder(binding) + .WithOutputFrom(binding) + .Build(); + + List updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: false); + + updates.Any(u => u.RawRepresentation is AgentResponseEvent) + .Should().BeFalse("an empty response did not previously produce an observable completion update"); } [Fact] @@ -937,8 +955,9 @@ static Workflow Build() .Should().BeFalse("terminal AgentResponseEvent stays gated under Futures-off"); List included = await RunStreamingAsync(Build(), includeWorkflowOutputsInResponse: true); - included.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText) - .Should().BeTrue("opting in via includeWorkflowOutputsInResponse surfaces it"); + included.Count(u => u.Text == FinalText).Should().Be(1); + included.Any(u => u.RawRepresentation is AgentResponseEvent && u.Contents.Count == 0) + .Should().BeTrue("opting in preserves the completion event without duplicating streamed text"); } [Fact] @@ -966,8 +985,9 @@ public async Task Test_WorkflowHostAgent_UndesignatedAgentResponseSurfacesWhenFu List updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true); - updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == InterText) - .Should().BeTrue("legacy bypass still emits AgentResponseEvent regardless of designation"); + updates.Count(u => u.Text == InterText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Contents.Count == 0) + .Should().BeTrue("legacy bypass preserves the completion event without duplicating streamed text"); } [Fact] @@ -982,10 +1002,166 @@ public async Task Test_WorkflowHostAgent_IntermediateTagAvailableViaRawRepresent List updates = await RunStreamingAsync(workflow); - AgentResponseUpdate progress = updates.First(u => u.RawRepresentation is AgentResponseEvent && u.Text == InterText); + AgentResponseUpdate progress = updates.First(u => u.RawRepresentation is AgentResponseEvent); AgentResponseEvent raw = (AgentResponseEvent)progress.RawRepresentation!; raw.IsIntermediate().Should().BeTrue(); raw.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate }); } + + [Fact] + public async Task Test_WorkflowHostAgent_DistinctCompletedResponseFromSameExecutorIsForwardedAsync() + { + using Futures.FuturesScope _ = new(enabled: false); + Workflow workflow = new WorkflowBuilder(new StreamThenCompleteExecutor()).Build(); + + List updates = + await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true); + + updates.Count(u => u.Text == InterText).Should().Be(1); + updates.Count(u => u.Text == FinalText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText) + .Should().BeTrue("a different response from the same executor must not be suppressed"); + } + + [Fact] + public async Task Test_WorkflowHostAgent_DistinctCompletedMessageFromSameResponseIsForwardedAsync() + { + using Futures.FuturesScope _ = new(enabled: false); + Workflow workflow = new WorkflowBuilder(new StreamThenCompleteExecutor(useSameResponseId: true)).Build(); + + List updates = + await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true); + + updates.Count(u => u.Text == InterText).Should().Be(1); + updates.Count(u => u.Text == FinalText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText) + .Should().BeTrue("a response ID alone must not suppress a distinct completed message"); + } + + [Fact] + public async Task Test_WorkflowHostAgent_WhitespaceMessageIdDoesNotSuppressCompletionAsync() + { + using Futures.FuturesScope _ = new(enabled: false); + Workflow workflow = + new WorkflowBuilder( + new StreamThenCompleteExecutor( + useSameResponseId: true, + streamedMessageId: " ", + completedMessageId: " ")) + .Build(); + + List updates = + await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true); + + updates.Count(u => u.Text == InterText).Should().Be(1); + updates.Count(u => u.Text == FinalText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText) + .Should().BeTrue("whitespace-only message IDs cannot reliably correlate streamed and completed messages"); + } + + [Fact] + public async Task Test_WorkflowHostAgent_UnstreamedMessageFromSameResponseIsForwardedAsync() + { + using Futures.FuturesScope _ = new(enabled: false); + Workflow workflow = new WorkflowBuilder(new PartiallyStreamedResponseExecutor()).Build(); + + List updates = + await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true); + + updates.Count(u => u.Text == InterText).Should().Be(1); + updates.Count(u => u.Text == FinalText).Should().Be(1); + updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText) + .Should().BeTrue("only the correlated message in a multi-message response should be suppressed"); + } + + private sealed class StreamThenCompleteExecutor( + bool useSameResponseId = false, + string streamedMessageId = "streamed-message", + string completedMessageId = "completed-message") : Executor("stream-then-complete") + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes( + routeBuilder => + routeBuilder + .AddHandler>(this.HandleMessagesAsync) + .AddHandler(this.HandleTurnAsync)); + + private ValueTask HandleMessagesAsync( + IEnumerable messages, + IWorkflowContext context, + CancellationToken cancellationToken) => default; + + private async ValueTask HandleTurnAsync( + TurnToken turnToken, + IWorkflowContext context, + CancellationToken cancellationToken) + { + AgentResponseUpdate update = + new(ChatRole.Assistant, InterText) + { + MessageId = streamedMessageId, + ResponseId = "streamed-response", + }; + await context.AddEventAsync( + new AgentResponseUpdateEvent(this.Id, update), + cancellationToken); + + ChatMessage message = + new(ChatRole.Assistant, FinalText) + { + MessageId = completedMessageId, + }; + return new AgentResponse([message]) + { + ResponseId = useSameResponseId ? update.ResponseId : "completed-response", + }; + } + } + + private sealed class PartiallyStreamedResponseExecutor() : Executor("partially-streamed-response") + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes( + routeBuilder => + routeBuilder + .AddHandler>(this.HandleMessagesAsync) + .AddHandler(this.HandleTurnAsync)); + + private ValueTask HandleMessagesAsync( + IEnumerable messages, + IWorkflowContext context, + CancellationToken cancellationToken) => default; + + private async ValueTask HandleTurnAsync( + TurnToken turnToken, + IWorkflowContext context, + CancellationToken cancellationToken) + { + const string ResponseId = "shared-response"; + ChatMessage streamedMessage = + new(ChatRole.Assistant, InterText) + { + MessageId = "streamed-message", + }; + AgentResponseUpdate streamedUpdate = + new(ChatRole.Assistant, InterText) + { + MessageId = streamedMessage.MessageId, + ResponseId = ResponseId, + }; + await context.AddEventAsync( + new AgentResponseUpdateEvent( + this.Id, + streamedUpdate), + cancellationToken); + + ChatMessage completedOnlyMessage = + new(ChatRole.Assistant, FinalText) + { + MessageId = "completed-only-message", + }; + return new AgentResponse([streamedMessage, completedOnlyMessage]) { ResponseId = ResponseId }; + } + } } }