fix: Remove duplicate history entry creation and ad test

2026-06-16 21:04:09 +08:00 · 2026-03-17 11:09:38 -04:00
parent e11ca24913
commit d565413a20
3 changed files with 120 additions and 111 deletions
@@ -43,7 +43,7 @@ internal sealed class WorkflowChatHistoryProvider : ChatHistoryProvider
        => this._sessionState.GetOrInitializeState(session).Messages.AddRange(messages);

    protected override ValueTask<IEnumerable<ChatMessage>> 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<ChatMessage> GetAllMessages(AgentSession session)
    {
        var state = this._sessionState.GetOrInitializeState(session);
-        return state.Messages;
+        return state.Messages.AsReadOnly();
    }

    public void UpdateBookmark(AgentSession session)
@@ -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<ResumeRunResult> CreateOrResumeRunAsync(List<ChatMessage> messages, CancellationToken cancellationToken = default)
@@ -328,111 +320,104 @@ internal sealed class WorkflowSession : AgentSession
    IAsyncEnumerable<AgentResponseUpdate> InvokeStageAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
-        try
-        {
-            this.LastResponseId = Guid.NewGuid().ToString("N");
-            List<ChatMessage> messages = this.ChatHistoryProvider.GetFromBookmark(this).ToList();
+        this.LastResponseId = Guid.NewGuid().ToString("N");
+        List<ChatMessage> 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<ChatMessage>? updateMessages = output.Data switch
-                        {
-                            IEnumerable<ChatMessage> 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<ChatMessage>? updateMessages = output.Data switch
+                    {
+                        IEnumerable<ChatMessage> 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;
+            }
        }
    }

@@ -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<WorkflowSession>().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());
    }
 }