diff --git a/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs b/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs index 30665fecf3..685dac8488 100644 --- a/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs @@ -204,6 +204,8 @@ public override async IAsyncEnumerable RunStreamingAsync (ChatClientAgentThread safeThread, ChatOptions? chatOptions, List inputMessagesForChatClient, IList? aiContextProviderMessages) = await this.PrepareThreadAndMessagesAsync(thread, inputMessages, options, cancellationToken).ConfigureAwait(false); + ValidateStreamResumptionAllowed(chatOptions?.ContinuationToken, safeThread); + var chatClient = this.ChatClient; chatClient = ApplyRunOptionsTransformations(options, chatClient); @@ -621,6 +623,12 @@ await thread.AIContextProvider.InvokedAsync(new(inputMessages, aiContextProvider { throw new InvalidOperationException("Input messages are not allowed when continuing a background response using a continuation token."); } + + if (chatOptions?.ContinuationToken is not null && typedThread.ConversationId is null && typedThread.MessageStore is null) + { + throw new InvalidOperationException("Continuation tokens are not allowed to be used for initial runs."); + } + List inputMessagesForChatClient = []; IList? aiContextProviderMessages = null; @@ -731,6 +739,28 @@ private static Task NotifyMessageStoreOfNewMessagesAsync(ChatClientAgentThread t return Task.CompletedTask; } + private static void ValidateStreamResumptionAllowed(ResponseContinuationToken? continuationToken, ChatClientAgentThread safeThread) + { + if (continuationToken is null) + { + return; + } + + // Streaming resumption is only supported with chat history managed by the agent service because, currently, there's no good solution + // to collect updates received in failed runs and pass them to the last successful run so it can store them to the message store. + if (safeThread.ConversationId is null) + { + throw new NotSupportedException("Streaming resumption is only supported when chat history is stored and managed by the underlying AI service."); + } + + // Similarly, streaming resumption is not supported when a context provider is used because, currently, there's no good solution + // to collect updates received in failed runs and pass them to the last successful run so it can notify the context provider of the updates. + if (safeThread.AIContextProvider is not null) + { + throw new NotSupportedException("Using context provider with streaming resumption is not supported."); + } + } + private string GetLoggingAgentName() => this.Name ?? "UnnamedAgent"; #endregion } diff --git a/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgentTests.cs b/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgentTests.cs index 920f9f82ee..8b46d7b57c 100644 --- a/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgentTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgentTests.cs @@ -2132,499 +2132,6 @@ await Assert.ThrowsAsync(async () => #endregion - #region Background Responses Tests - - [Theory] - [InlineData(true)] - [InlineData(false)] - public async Task RunAsyncPropagatesBackgroundResponsesPropertiesToChatClientAsync(bool providePropsViaChatOptions) - { - // Arrange - var continuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - ChatOptions? capturedChatOptions = null; - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) - .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "response")]) { ContinuationToken = null }); - - AgentRunOptions agentRunOptions; - - if (providePropsViaChatOptions) - { - ChatOptions chatOptions = new() - { - AllowBackgroundResponses = true, - ContinuationToken = continuationToken - }; - - agentRunOptions = new ChatClientAgentRunOptions(chatOptions); - } - else - { - agentRunOptions = new AgentRunOptions() - { - AllowBackgroundResponses = true, - ContinuationToken = continuationToken - }; - } - - ChatClientAgent agent = new(mockChatClient.Object); - - ChatClientAgentThread thread = new(); - - // Act - await agent.RunAsync(thread, options: agentRunOptions); - - // Assert - Assert.NotNull(capturedChatOptions); - Assert.True(capturedChatOptions.AllowBackgroundResponses); - Assert.Same(continuationToken, capturedChatOptions.ContinuationToken); - } - - [Fact] - public async Task RunAsyncPrioritizesBackgroundResponsesPropertiesFromAgentRunOptionsOverOnesFromChatOptionsAsync() - { - // Arrange - var continuationToken1 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - var continuationToken2 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - ChatOptions? capturedChatOptions = null; - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) - .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "response")]) { ContinuationToken = null }); - - ChatOptions chatOptions = new() - { - AllowBackgroundResponses = true, - ContinuationToken = continuationToken1 - }; - - ChatClientAgentRunOptions agentRunOptions = new(chatOptions) - { - AllowBackgroundResponses = false, - ContinuationToken = continuationToken2 - }; - - ChatClientAgent agent = new(mockChatClient.Object); - - // Act - await agent.RunAsync(options: agentRunOptions); - - // Assert - Assert.NotNull(capturedChatOptions); - Assert.False(capturedChatOptions.AllowBackgroundResponses); - Assert.Same(continuationToken2, capturedChatOptions.ContinuationToken); - } - - [Theory] - [InlineData(true)] - [InlineData(false)] - public async Task RunStreamingAsyncPropagatesBackgroundResponsesPropertiesToChatClientAsync(bool providePropsViaChatOptions) - { - // Arrange - ChatResponseUpdate[] returnUpdates = - [ - new ChatResponseUpdate(role: ChatRole.Assistant, content: "wh"), - new ChatResponseUpdate(role: ChatRole.Assistant, content: "at?"), - ]; - - var continuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - ChatOptions? capturedChatOptions = null; - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetStreamingResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) - .Returns(ToAsyncEnumerableAsync(returnUpdates)); - - AgentRunOptions agentRunOptions; - - if (providePropsViaChatOptions) - { - ChatOptions chatOptions = new() - { - AllowBackgroundResponses = true, - ContinuationToken = continuationToken - }; - - agentRunOptions = new ChatClientAgentRunOptions(chatOptions); - } - else - { - agentRunOptions = new AgentRunOptions() - { - AllowBackgroundResponses = true, - ContinuationToken = continuationToken - }; - } - - ChatClientAgent agent = new(mockChatClient.Object); - - ChatClientAgentThread thread = new(); - - // Act - await foreach (var _ in agent.RunStreamingAsync(thread, options: agentRunOptions)) - { - } - - // Assert - Assert.NotNull(capturedChatOptions); - - Assert.True(capturedChatOptions.AllowBackgroundResponses); - Assert.Same(continuationToken, capturedChatOptions.ContinuationToken); - } - - [Fact] - public async Task RunStreamingAsyncPrioritizesBackgroundResponsesPropertiesFromAgentRunOptionsOverOnesFromChatOptionsAsync() - { - // Arrange - ChatResponseUpdate[] returnUpdates = - [ - new ChatResponseUpdate(role: ChatRole.Assistant, content: "wh"), - ]; - - var continuationToken1 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - var continuationToken2 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - ChatOptions? capturedChatOptions = null; - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetStreamingResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) - .Returns(ToAsyncEnumerableAsync(returnUpdates)); - - ChatOptions chatOptions = new() - { - AllowBackgroundResponses = true, - ContinuationToken = continuationToken1 - }; - - ChatClientAgentRunOptions agentRunOptions = new(chatOptions) - { - AllowBackgroundResponses = false, - ContinuationToken = continuationToken2 - }; - - ChatClientAgent agent = new(mockChatClient.Object); - - // Act - await foreach (var _ in agent.RunStreamingAsync(options: agentRunOptions)) - { - } - - // Assert - Assert.NotNull(capturedChatOptions); - Assert.False(capturedChatOptions.AllowBackgroundResponses); - Assert.Same(continuationToken2, capturedChatOptions.ContinuationToken); - } - - [Fact] - public async Task RunAsyncPropagatesContinuationTokenFromChatResponseToAgentRunResponseAsync() - { - // Arrange - var continuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "partial")]) { ContinuationToken = continuationToken }); - - ChatClientAgent agent = new(mockChatClient.Object); - var runOptions = new ChatClientAgentRunOptions(new ChatOptions { AllowBackgroundResponses = true }); - - ChatClientAgentThread thread = new(); - - // Act - var response = await agent.RunAsync([new(ChatRole.User, "hi")], thread, options: runOptions); - - // Assert - Assert.Same(continuationToken, response.ContinuationToken); - } - - [Fact] - public async Task RunStreamingAsyncPropagatesContinuationTokensFromUpdatesAsync() - { - // Arrange - var token1 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); - ChatResponseUpdate[] expectedUpdates = - [ - new ChatResponseUpdate(ChatRole.Assistant, "pa") { ContinuationToken = token1 }, - new ChatResponseUpdate(ChatRole.Assistant, "rt") { ContinuationToken = null } // terminal - ]; - - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetStreamingResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Returns(ToAsyncEnumerableAsync(expectedUpdates)); - - ChatClientAgent agent = new(mockChatClient.Object); - - ChatClientAgentThread thread = new(); - - // Act - var actualUpdates = new List(); - await foreach (var u in agent.RunStreamingAsync([new(ChatRole.User, "hi")], thread, options: new ChatClientAgentRunOptions(new ChatOptions { AllowBackgroundResponses = true }))) - { - actualUpdates.Add(u); - } - - // Assert - Assert.Equal(2, actualUpdates.Count); - Assert.Same(token1, actualUpdates[0].ContinuationToken); - Assert.Null(actualUpdates[1].ContinuationToken); // last update has null token - } - - [Fact] - public async Task RunAsyncThrowsWhenMessagesProvidedWithContinuationTokenAsync() - { - // Arrange - Mock mockChatClient = new(); - - ChatClientAgent agent = new(mockChatClient.Object); - - AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; - - IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; - - // Act & Assert - await Assert.ThrowsAsync(() => agent.RunAsync(inputMessages, options: runOptions)); - - // Verify that the IChatClient was never called due to early validation - mockChatClient.Verify( - c => c.GetResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny()), - Times.Never); - } - - [Fact] - public async Task RunStreamingAsyncThrowsWhenMessagesProvidedWithContinuationTokenAsync() - { - // Arrange - Mock mockChatClient = new(); - - ChatClientAgent agent = new(mockChatClient.Object); - - AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; - - IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; - - // Act & Assert - await Assert.ThrowsAsync(async () => - { - await foreach (var update in agent.RunStreamingAsync(inputMessages, options: runOptions)) - { - // Should not reach here - } - }); - - // Verify that the IChatClient was never called due to early validation - mockChatClient.Verify( - c => c.GetStreamingResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny()), - Times.Never); - } - - [Fact] - public async Task RunAsyncSkipsThreadMessagePopulationWithContinuationTokenAsync() - { - // Arrange - List capturedMessages = []; - - // Create a mock message store that would normally provide messages - var mockMessageStore = new Mock(); - mockMessageStore - .Setup(ms => ms.GetMessagesAsync(It.IsAny())) - .ReturnsAsync([new(ChatRole.User, "Message from message store")]); - - // Create a mock AI context provider that would normally provide context - var mockContextProvider = new Mock(); - mockContextProvider - .Setup(p => p.InvokingAsync(It.IsAny(), It.IsAny())) - .ReturnsAsync(new AIContext - { - Messages = [new(ChatRole.System, "Message from AI context")], - Instructions = "context instructions" - }); - - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Callback, ChatOptions, CancellationToken>((msgs, opts, ct) => - capturedMessages.AddRange(msgs)) - .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "continued response")])); - - ChatClientAgent agent = new(mockChatClient.Object); - - // Create a thread with both message store and AI context provider - ChatClientAgentThread thread = new() - { - MessageStore = mockMessageStore.Object, - AIContextProvider = mockContextProvider.Object - }; - - AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; - - // Act - await agent.RunAsync([], thread, options: runOptions); - - // Assert - - // With continuation token, thread message population should be skipped - Assert.Empty(capturedMessages); - - // Verify that message store was never called due to continuation token - mockMessageStore.Verify( - ms => ms.GetMessagesAsync(It.IsAny()), - Times.Never); - - // Verify that AI context provider was never called due to continuation token - mockContextProvider.Verify( - p => p.InvokingAsync(It.IsAny(), It.IsAny()), - Times.Never); - } - - [Fact] - public async Task RunStreamingAsyncSkipsThreadMessagePopulationWithContinuationTokenAsync() - { - // Arrange - List capturedMessages = []; - - // Create a mock message store that would normally provide messages - var mockMessageStore = new Mock(); - mockMessageStore - .Setup(ms => ms.GetMessagesAsync(It.IsAny())) - .ReturnsAsync([new(ChatRole.User, "Message from message store")]); - - // Create a mock AI context provider that would normally provide context - var mockContextProvider = new Mock(); - mockContextProvider - .Setup(p => p.InvokingAsync(It.IsAny(), It.IsAny())) - .ReturnsAsync(new AIContext - { - Messages = [new(ChatRole.System, "Message from AI context")], - Instructions = "context instructions" - }); - - Mock mockChatClient = new(); - mockChatClient - .Setup(c => c.GetStreamingResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny())) - .Callback, ChatOptions, CancellationToken>((msgs, opts, ct) => - capturedMessages.AddRange(msgs)) - .Returns(ToAsyncEnumerableAsync([new ChatResponseUpdate(role: ChatRole.Assistant, content: "continued response")])); - - ChatClientAgent agent = new(mockChatClient.Object); - - // Create a thread with both message store and AI context provider - ChatClientAgentThread thread = new() - { - MessageStore = mockMessageStore.Object, - AIContextProvider = mockContextProvider.Object - }; - - AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; - - // Act - await agent.RunStreamingAsync([], thread, options: runOptions).ToListAsync(); - - // Assert - - // With continuation token, thread message population should be skipped - Assert.Empty(capturedMessages); - - // Verify that message store was never called due to continuation token - mockMessageStore.Verify( - ms => ms.GetMessagesAsync(It.IsAny()), - Times.Never); - - // Verify that AI context provider was never called due to continuation token - mockContextProvider.Verify( - p => p.InvokingAsync(It.IsAny(), It.IsAny()), - Times.Never); - } - - [Fact] - public async Task RunAsyncThrowsWhenNoThreadProvideForBackgroundResponsesAsync() - { - // Arrange - Mock mockChatClient = new(); - - ChatClientAgent agent = new(mockChatClient.Object); - - AgentRunOptions runOptions = new() { AllowBackgroundResponses = true }; - - IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; - - // Act & Assert - await Assert.ThrowsAsync(() => agent.RunAsync(inputMessages, options: runOptions)); - - // Verify that the IChatClient was never called due to early validation - mockChatClient.Verify( - c => c.GetResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny()), - Times.Never); - } - - [Fact] - public async Task RunStreamingAsyncThrowsWhenNoThreadProvideForBackgroundResponsesAsync() - { - // Arrange - Mock mockChatClient = new(); - - ChatClientAgent agent = new(mockChatClient.Object); - - AgentRunOptions runOptions = new() { AllowBackgroundResponses = true }; - - IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; - - // Act & Assert - await Assert.ThrowsAsync(async () => - { - await foreach (var update in agent.RunStreamingAsync(inputMessages, options: runOptions)) - { - // Should not reach here - } - }); - - // Verify that the IChatClient was never called due to early validation - mockChatClient.Verify( - c => c.GetStreamingResponseAsync( - It.IsAny>(), - It.IsAny(), - It.IsAny()), - Times.Never); - } - - #endregion - private static async IAsyncEnumerable ToAsyncEnumerableAsync(IEnumerable values) { await Task.Yield(); diff --git a/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgent_BackgroundResponsesTests.cs b/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgent_BackgroundResponsesTests.cs new file mode 100644 index 0000000000..583a0815ca --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.UnitTests/ChatClient/ChatClientAgent_BackgroundResponsesTests.cs @@ -0,0 +1,643 @@ +// 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; +using Moq; + +namespace Microsoft.Agents.AI.UnitTests; + +/// +/// Contains unit tests for ChatClientAgent background responses functionality. +/// +public class ChatClientAgent_BackgroundResponsesTests +{ + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task RunAsyncPropagatesBackgroundResponsesPropertiesToChatClientAsync(bool providePropsViaChatOptions) + { + // Arrange + var continuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + ChatOptions? capturedChatOptions = null; + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) + .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "response")]) { ContinuationToken = null, ConversationId = "conversation-id" }); + + AgentRunOptions agentRunOptions; + + if (providePropsViaChatOptions) + { + ChatOptions chatOptions = new() + { + AllowBackgroundResponses = true, + ContinuationToken = continuationToken + }; + + agentRunOptions = new ChatClientAgentRunOptions(chatOptions); + } + else + { + agentRunOptions = new AgentRunOptions() + { + AllowBackgroundResponses = true, + ContinuationToken = continuationToken + }; + } + + ChatClientAgent agent = new(mockChatClient.Object); + + ChatClientAgentThread thread = new() { ConversationId = "conversation-id" }; + + // Act + await agent.RunAsync(thread, options: agentRunOptions); + + // Assert + Assert.NotNull(capturedChatOptions); + Assert.True(capturedChatOptions.AllowBackgroundResponses); + Assert.Same(continuationToken, capturedChatOptions.ContinuationToken); + } + + [Fact] + public async Task RunAsyncPrioritizesBackgroundResponsesPropertiesFromAgentRunOptionsOverOnesFromChatOptionsAsync() + { + // Arrange + var continuationToken1 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + var continuationToken2 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + ChatOptions? capturedChatOptions = null; + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) + .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "response")]) { ContinuationToken = null, ConversationId = "conversation-id" }); + + ChatOptions chatOptions = new() + { + AllowBackgroundResponses = true, + ContinuationToken = continuationToken1 + }; + + ChatClientAgentRunOptions agentRunOptions = new(chatOptions) + { + AllowBackgroundResponses = false, + ContinuationToken = continuationToken2 + }; + + ChatClientAgentThread thread = new() { ConversationId = "conversation-id" }; + + ChatClientAgent agent = new(mockChatClient.Object); + + // Act + await agent.RunAsync(thread, options: agentRunOptions); + + // Assert + Assert.NotNull(capturedChatOptions); + Assert.False(capturedChatOptions.AllowBackgroundResponses); + Assert.Same(continuationToken2, capturedChatOptions.ContinuationToken); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task RunStreamingAsyncPropagatesBackgroundResponsesPropertiesToChatClientAsync(bool providePropsViaChatOptions) + { + // Arrange + ChatResponseUpdate[] returnUpdates = + [ + new ChatResponseUpdate(role: ChatRole.Assistant, content: "wh") { ConversationId = "conversation-id" }, + new ChatResponseUpdate(role: ChatRole.Assistant, content: "at?") { ConversationId = "conversation-id" }, + ]; + + var continuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + ChatOptions? capturedChatOptions = null; + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) + .Returns(ToAsyncEnumerableAsync(returnUpdates)); + + AgentRunOptions agentRunOptions; + + if (providePropsViaChatOptions) + { + ChatOptions chatOptions = new() + { + AllowBackgroundResponses = true, + ContinuationToken = continuationToken + }; + + agentRunOptions = new ChatClientAgentRunOptions(chatOptions); + } + else + { + agentRunOptions = new AgentRunOptions() + { + AllowBackgroundResponses = true, + ContinuationToken = continuationToken + }; + } + + ChatClientAgent agent = new(mockChatClient.Object); + + ChatClientAgentThread thread = new() { ConversationId = "conversation-id" }; + + // Act + await foreach (var _ in agent.RunStreamingAsync(thread, options: agentRunOptions)) + { + } + + // Assert + Assert.NotNull(capturedChatOptions); + + Assert.True(capturedChatOptions.AllowBackgroundResponses); + Assert.Same(continuationToken, capturedChatOptions.ContinuationToken); + } + + [Fact] + public async Task RunStreamingAsyncPrioritizesBackgroundResponsesPropertiesFromAgentRunOptionsOverOnesFromChatOptionsAsync() + { + // Arrange + ChatResponseUpdate[] returnUpdates = + [ + new ChatResponseUpdate(role: ChatRole.Assistant, content: "wh") { ConversationId = "conversation-id" }, + ]; + + var continuationToken1 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + var continuationToken2 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + ChatOptions? capturedChatOptions = null; + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback, ChatOptions, CancellationToken>((m, co, ct) => capturedChatOptions = co) + .Returns(ToAsyncEnumerableAsync(returnUpdates)); + + ChatOptions chatOptions = new() + { + AllowBackgroundResponses = true, + ContinuationToken = continuationToken1 + }; + + ChatClientAgentRunOptions agentRunOptions = new(chatOptions) + { + AllowBackgroundResponses = false, + ContinuationToken = continuationToken2 + }; + + ChatClientAgent agent = new(mockChatClient.Object); + + var thread = new ChatClientAgentThread() { ConversationId = "conversation-id" }; + + // Act + await foreach (var _ in agent.RunStreamingAsync(thread, options: agentRunOptions)) + { + } + + // Assert + Assert.NotNull(capturedChatOptions); + Assert.False(capturedChatOptions.AllowBackgroundResponses); + Assert.Same(continuationToken2, capturedChatOptions.ContinuationToken); + } + + [Fact] + public async Task RunAsyncPropagatesContinuationTokenFromChatResponseToAgentRunResponseAsync() + { + // Arrange + var continuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "partial")]) { ContinuationToken = continuationToken }); + + ChatClientAgent agent = new(mockChatClient.Object); + var runOptions = new ChatClientAgentRunOptions(new ChatOptions { AllowBackgroundResponses = true }); + + ChatClientAgentThread thread = new(); + + // Act + var response = await agent.RunAsync([new(ChatRole.User, "hi")], thread, options: runOptions); + + // Assert + Assert.Same(continuationToken, response.ContinuationToken); + } + + [Fact] + public async Task RunStreamingAsyncPropagatesContinuationTokensFromUpdatesAsync() + { + // Arrange + var token1 = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }); + ChatResponseUpdate[] expectedUpdates = + [ + new ChatResponseUpdate(ChatRole.Assistant, "pa") { ContinuationToken = token1 }, + new ChatResponseUpdate(ChatRole.Assistant, "rt") { ContinuationToken = null } // terminal + ]; + + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns(ToAsyncEnumerableAsync(expectedUpdates)); + + ChatClientAgent agent = new(mockChatClient.Object); + + ChatClientAgentThread thread = new(); + + // Act + var actualUpdates = new List(); + await foreach (var u in agent.RunStreamingAsync([new(ChatRole.User, "hi")], thread, options: new ChatClientAgentRunOptions(new ChatOptions { AllowBackgroundResponses = true }))) + { + actualUpdates.Add(u); + } + + // Assert + Assert.Equal(2, actualUpdates.Count); + Assert.Same(token1, actualUpdates[0].ContinuationToken); + Assert.Null(actualUpdates[1].ContinuationToken); // last update has null token + } + + [Fact] + public async Task RunAsyncThrowsWhenMessagesProvidedWithContinuationTokenAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; + + // Act & Assert + await Assert.ThrowsAsync(() => agent.RunAsync(inputMessages, options: runOptions)); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunStreamingAsyncThrowsWhenMessagesProvidedWithContinuationTokenAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; + + // Act & Assert + await Assert.ThrowsAsync(async () => + { + await foreach (var update in agent.RunStreamingAsync(inputMessages, options: runOptions)) + { + // Should not reach here + } + }); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunAsyncSkipsThreadMessagePopulationWithContinuationTokenAsync() + { + // Arrange + List capturedMessages = []; + + // Create a mock message store that would normally provide messages + var mockMessageStore = new Mock(); + mockMessageStore + .Setup(ms => ms.GetMessagesAsync(It.IsAny())) + .ReturnsAsync([new(ChatRole.User, "Message from message store")]); + + // Create a mock AI context provider that would normally provide context + var mockContextProvider = new Mock(); + mockContextProvider + .Setup(p => p.InvokingAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new AIContext + { + Messages = [new(ChatRole.System, "Message from AI context")], + Instructions = "context instructions" + }); + + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback, ChatOptions, CancellationToken>((msgs, opts, ct) => + capturedMessages.AddRange(msgs)) + .ReturnsAsync(new ChatResponse([new(ChatRole.Assistant, "continued response")])); + + ChatClientAgent agent = new(mockChatClient.Object); + + // Create a thread with both message store and AI context provider + ChatClientAgentThread thread = new() + { + MessageStore = mockMessageStore.Object, + AIContextProvider = mockContextProvider.Object + }; + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + // Act + await agent.RunAsync([], thread, options: runOptions); + + // Assert + + // With continuation token, thread message population should be skipped + Assert.Empty(capturedMessages); + + // Verify that message store was never called due to continuation token + mockMessageStore.Verify( + ms => ms.GetMessagesAsync(It.IsAny()), + Times.Never); + + // Verify that AI context provider was never called due to continuation token + mockContextProvider.Verify( + p => p.InvokingAsync(It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunStreamingAsyncSkipsThreadMessagePopulationWithContinuationTokenAsync() + { + // Arrange + List capturedMessages = []; + + // Create a mock message store that would normally provide messages + var mockMessageStore = new Mock(); + mockMessageStore + .Setup(ms => ms.GetMessagesAsync(It.IsAny())) + .ReturnsAsync([new(ChatRole.User, "Message from message store")]); + + // Create a mock AI context provider that would normally provide context + var mockContextProvider = new Mock(); + mockContextProvider + .Setup(p => p.InvokingAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new AIContext + { + Messages = [new(ChatRole.System, "Message from AI context")], + Instructions = "context instructions" + }); + + Mock mockChatClient = new(); + mockChatClient + .Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback, ChatOptions, CancellationToken>((msgs, opts, ct) => + capturedMessages.AddRange(msgs)) + .Returns(ToAsyncEnumerableAsync([new ChatResponseUpdate(role: ChatRole.Assistant, content: "continued response")])); + + ChatClientAgent agent = new(mockChatClient.Object); + + // Create a thread with both message store and AI context provider + ChatClientAgentThread thread = new() + { + MessageStore = mockMessageStore.Object, + AIContextProvider = mockContextProvider.Object + }; + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + // Act + var exception = await Assert.ThrowsAsync(async () => await agent.RunStreamingAsync(thread, options: runOptions).ToListAsync()); + + // Assert + Assert.Equal("Streaming resumption is only supported when chat history is stored and managed by the underlying AI service.", exception.Message); + + // With continuation token, thread message population should be skipped + Assert.Empty(capturedMessages); + + // Verify that message store was never called due to continuation token + mockMessageStore.Verify( + ms => ms.GetMessagesAsync(It.IsAny()), + Times.Never); + + // Verify that AI context provider was never called due to continuation token + mockContextProvider.Verify( + p => p.InvokingAsync(It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunAsyncThrowsWhenNoThreadProvideForBackgroundResponsesAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + AgentRunOptions runOptions = new() { AllowBackgroundResponses = true }; + + IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; + + // Act & Assert + await Assert.ThrowsAsync(() => agent.RunAsync(inputMessages, options: runOptions)); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunStreamingAsyncThrowsWhenNoThreadProvideForBackgroundResponsesAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + AgentRunOptions runOptions = new() { AllowBackgroundResponses = true }; + + IEnumerable inputMessages = [new ChatMessage(ChatRole.User, "test message")]; + + // Act & Assert + await Assert.ThrowsAsync(async () => + { + await foreach (var update in agent.RunStreamingAsync(inputMessages, options: runOptions)) + { + // Should not reach here + } + }); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunAsyncThrowsWhenContinuationTokenProvidedForInitialRunAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + // Create a new thread with no ConversationId and no MessageStore (initial run state) + ChatClientAgentThread thread = new(); + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + // Act & Assert + var exception = await Assert.ThrowsAsync(() => agent.RunAsync(thread: thread, options: runOptions)); + Assert.Equal("Continuation tokens are not allowed to be used for initial runs.", exception.Message); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunStreamingAsyncThrowsWhenContinuationTokenProvidedForInitialRunAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + // Create a new thread with no ConversationId and no MessageStore (initial run state) + ChatClientAgentThread thread = new(); + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + // Act & Assert + var exception = await Assert.ThrowsAsync(async () => await agent.RunStreamingAsync(thread: thread, options: runOptions).ToListAsync()); + Assert.Equal("Continuation tokens are not allowed to be used for initial runs.", exception.Message); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunStreamingAsyncThrowsWhenContinuationTokenUsedWithClientSideManagedChatHistoryAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + // Create a thread with a MessageStore + ChatClientAgentThread thread = new() + { + MessageStore = new InMemoryChatMessageStore(), // Setting a message store to skip checking the continuation token in the initial run + ConversationId = null, // No conversation ID to simulate client-side managed chat history + }; + + // Create run options with a continuation token + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + // Act & Assert + var exception = await Assert.ThrowsAsync(async () => await agent.RunStreamingAsync(thread: thread, options: runOptions).ToListAsync()); + Assert.Equal("Streaming resumption is only supported when chat history is stored and managed by the underlying AI service.", exception.Message); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [Fact] + public async Task RunStreamingAsyncThrowsWhenContinuationTokenUsedWithAIContextProviderAsync() + { + // Arrange + Mock mockChatClient = new(); + + ChatClientAgent agent = new(mockChatClient.Object); + + // Create a mock AIContextProvider + var mockContextProvider = new Mock(); + mockContextProvider + .Setup(p => p.InvokingAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new AIContext()); + mockContextProvider + .Setup(p => p.InvokedAsync(It.IsAny(), It.IsAny())) + .Returns(new ValueTask()); + + // Create a thread with an AIContextProvider and conversation ID to simulate non-initial run + ChatClientAgentThread thread = new() + { + ConversationId = "existing-conversation-id", + AIContextProvider = mockContextProvider.Object + }; + + AgentRunOptions runOptions = new() { ContinuationToken = ResponseContinuationToken.FromBytes(new byte[] { 1, 2, 3 }) }; + + // Act & Assert + var exception = await Assert.ThrowsAsync(async () => await agent.RunStreamingAsync(thread: thread, options: runOptions).ToListAsync()); + + Assert.Equal("Using context provider with streaming resumption is not supported.", exception.Message); + + // Verify that the IChatClient was never called due to early validation + mockChatClient.Verify( + c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + private static async IAsyncEnumerable ToAsyncEnumerableAsync(IEnumerable values) + { + await Task.Yield(); + foreach (var update in values) + { + yield return update; + } + } +}