From d565413a205271d837779019685fc5dcd2446cee Mon Sep 17 00:00:00 2001 From: Jacob Alber Date: Tue, 17 Mar 2026 11:09:38 -0400 Subject: [PATCH] fix: Remove duplicate history entry creation and ad test --- .../WorkflowChatHistoryProvider.cs | 4 +- .../WorkflowSession.cs | 199 ++++++++---------- .../WorkflowHostSmokeTests.cs | 28 ++- 3 files changed, 120 insertions(+), 111 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowChatHistoryProvider.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowChatHistoryProvider.cs index 3f6656b994..69c09abb27 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowChatHistoryProvider.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowChatHistoryProvider.cs @@ -43,7 +43,7 @@ internal sealed class WorkflowChatHistoryProvider : ChatHistoryProvider => this._sessionState.GetOrInitializeState(session).Messages.AddRange(messages); protected override ValueTask> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default) - => new(this._sessionState.GetOrInitializeState(context.Session).Messages); + => new(this._sessionState.GetOrInitializeState(context.Session).Messages.AsReadOnly()); protected override ValueTask StoreChatHistoryAsync(InvokedContext context, CancellationToken cancellationToken = default) { @@ -65,7 +65,7 @@ internal sealed class WorkflowChatHistoryProvider : ChatHistoryProvider public IEnumerable GetAllMessages(AgentSession session) { var state = this._sessionState.GetOrInitializeState(session); - return state.Messages; + return state.Messages.AsReadOnly(); } public void UpdateBookmark(AgentSession session) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs index db3d299ee9..715c232e7d 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs @@ -131,7 +131,7 @@ internal sealed class WorkflowSession : AgentSession { Throw.IfNullOrEmpty(parts); - AgentResponseUpdate update = new(ChatRole.Assistant, parts) + return new(ChatRole.Assistant, parts) { CreatedAt = DateTimeOffset.UtcNow, MessageId = Guid.NewGuid().ToString("N"), @@ -139,27 +139,19 @@ internal sealed class WorkflowSession : AgentSession ResponseId = responseId, RawRepresentation = raw }; - - this.ChatHistoryProvider.AddMessages(this, update.ToChatMessage()); - - return update; } public AgentResponseUpdate CreateUpdate(string responseId, object raw, ChatMessage message) { Throw.IfNull(message); - AgentResponseUpdate update = new(message.Role, message.Contents) + return new(message.Role, message.Contents) { CreatedAt = message.CreatedAt ?? DateTimeOffset.UtcNow, MessageId = message.MessageId ?? Guid.NewGuid().ToString("N"), ResponseId = responseId, RawRepresentation = raw }; - - this.ChatHistoryProvider.AddMessages(this, update.ToChatMessage()); - - return update; } private async ValueTask CreateOrResumeRunAsync(List messages, CancellationToken cancellationToken = default) @@ -328,111 +320,104 @@ internal sealed class WorkflowSession : AgentSession IAsyncEnumerable InvokeStageAsync( [EnumeratorCancellation] CancellationToken cancellationToken = default) { - try - { - this.LastResponseId = Guid.NewGuid().ToString("N"); - List messages = this.ChatHistoryProvider.GetFromBookmark(this).ToList(); + this.LastResponseId = Guid.NewGuid().ToString("N"); + List messages = this.ChatHistoryProvider.GetFromBookmark(this).ToList(); + + ResumeRunResult resumeResult = + await this.CreateOrResumeRunAsync(messages, cancellationToken).ConfigureAwait(false); - ResumeRunResult resumeResult = - await this.CreateOrResumeRunAsync(messages, cancellationToken).ConfigureAwait(false); #pragma warning disable CA2007 // Analyzer misfiring. - await using StreamingRun run = resumeResult.Run; + await using StreamingRun run = resumeResult.Run; #pragma warning restore CA2007 - ResumeDispatchInfo dispatchInfo = resumeResult.DispatchInfo; + 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.HasMatchedResponseForStartExecutor; - if (shouldSendTurnToken) - { - await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); - } - await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken) + // 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.HasMatchedResponseForStartExecutor; + if (shouldSendTurnToken) + { + await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); + } + await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken) .ConfigureAwait(false) .WithCancellation(cancellationToken)) - { - switch (evt) - { - case AgentResponseUpdateEvent agentUpdate: - yield return agentUpdate.Update; - break; - - case RequestInfoEvent requestInfo: - AIContent requestContent = CreateRequestContentForDelivery(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; - break; - - case WorkflowErrorEvent workflowError: - Exception? exception = workflowError.Exception; - if (exception is TargetInvocationException tie && tie.InnerException != null) - { - exception = tie.InnerException; - } - - if (exception != null) - { - string message = this._includeExceptionDetails - ? exception.Message - : "An error occurred while executing the workflow."; - - ErrorContent errorContent = new(message); - yield return this.CreateUpdate(this.LastResponseId, evt, errorContent); - } - - break; - - case SuperStepCompletedEvent stepCompleted: - this.LastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint; - goto default; - - case WorkflowOutputEvent output: - IEnumerable? updateMessages = output.Data switch - { - IEnumerable chatMessages => chatMessages, - ChatMessage chatMessage => [chatMessage], - _ => null - }; - - if (!this._includeWorkflowOutputsInResponse || updateMessages == null) - { - goto default; - } - - foreach (ChatMessage message in updateMessages) - { - yield return this.CreateUpdate(this.LastResponseId, evt, message); - } - break; - - default: - // Emit all other workflow events for observability (DevUI, logging, etc.) - yield return new AgentResponseUpdate(ChatRole.Assistant, []) - { - CreatedAt = DateTimeOffset.UtcNow, - MessageId = Guid.NewGuid().ToString("N"), - Role = ChatRole.Assistant, - ResponseId = this.LastResponseId, - RawRepresentation = evt - }; - break; - } - } - } - finally { - // Do we want to try to undo the step, and not update the bookmark? - this.ChatHistoryProvider.UpdateBookmark(this); + switch (evt) + { + case AgentResponseUpdateEvent agentUpdate: + yield return agentUpdate.Update; + break; + + case RequestInfoEvent requestInfo: + AIContent requestContent = CreateRequestContentForDelivery(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; + break; + + case WorkflowErrorEvent workflowError: + Exception? exception = workflowError.Exception; + if (exception is TargetInvocationException tie && tie.InnerException != null) + { + exception = tie.InnerException; + } + + if (exception != null) + { + string message = this._includeExceptionDetails + ? exception.Message + : "An error occurred while executing the workflow."; + + ErrorContent errorContent = new(message); + yield return this.CreateUpdate(this.LastResponseId, evt, errorContent); + } + + break; + + case SuperStepCompletedEvent stepCompleted: + this.LastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint; + goto default; + + case WorkflowOutputEvent output: + IEnumerable? updateMessages = output.Data switch + { + IEnumerable chatMessages => chatMessages, + ChatMessage chatMessage => [chatMessage], + _ => null + }; + + if (!this._includeWorkflowOutputsInResponse || updateMessages == null) + { + goto default; + } + + foreach (ChatMessage message in updateMessages) + { + yield return this.CreateUpdate(this.LastResponseId, evt, message); + } + break; + + default: + // Emit all other workflow events for observability (DevUI, logging, etc.) + yield return new AgentResponseUpdate(ChatRole.Assistant, []) + { + CreatedAt = DateTimeOffset.UtcNow, + MessageId = Guid.NewGuid().ToString("N"), + Role = ChatRole.Assistant, + ResponseId = this.LastResponseId, + RawRepresentation = evt + }; + break; + } } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs index e61f4efe07..8cd063565b 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs @@ -733,7 +733,30 @@ public class WorkflowHostSmokeTests : AIAgentHostingExecutorTestsBase } [Fact] - public async Task Test_AsAgent_OutgoingMessagesInHistoryAsync() + public async Task Test_SingleAgent_AsAgent_OutgoingMessagesInHistoryAsync() + { + // Arrange + TestReplayAgent agent = new(TestMessages, TestAgentId, TestAgentName); + Workflow singleAgentWorkflow = new WorkflowBuilder(agent).Build(); + AIAgent workflowAgent = singleAgentWorkflow.AsAIAgent(); + + // Act + AgentSession session = await workflowAgent.CreateSessionAsync(); + AgentResponse response = await workflowAgent.RunAsync(session); + + // Assert + WorkflowSession workflowSession = session.Should().BeOfType().Subject; + + ChatMessage[] responseMessages = response.Messages.ToArray(); + ChatMessage[] sessionMessages = workflowSession.ChatHistoryProvider.GetAllMessages(workflowSession).ToArray(); + + // Since we never sent an incoming message, the expectation is that there should be nothing in the session + // except the response + responseMessages.Should().BeEquivalentTo(sessionMessages, options => options.WithStrictOrdering()); + } + + [Fact] + public async Task Test_Handoffs_AsAgent_OutgoingMessagesInHistoryAsync() { // Arrange TestReplayAgent agent = new(TestMessages, TestAgentId, TestAgentName); @@ -751,6 +774,7 @@ public class WorkflowHostSmokeTests : AIAgentHostingExecutorTestsBase ChatMessage[] sessionMessages = workflowSession.ChatHistoryProvider.GetAllMessages(workflowSession).ToArray(); // Since we never sent an incoming message, the expectation is that there should be nothing in the session - responseMessages.Should().BeEquivalentTo(sessionMessages); + // except the response + responseMessages.Should().BeEquivalentTo(sessionMessages, options => options.WithStrictOrdering()); } }