diff --git a/dotnet/src/Microsoft.Extensions.AI.Agents.Hosting/AgentActor.cs b/dotnet/src/Microsoft.Extensions.AI.Agents.Hosting/AgentActor.cs index 9502e2c062..31d91c1c6d 100644 --- a/dotnet/src/Microsoft.Extensions.AI.Agents.Hosting/AgentActor.cs +++ b/dotnet/src/Microsoft.Extensions.AI.Agents.Hosting/AgentActor.cs @@ -33,17 +33,19 @@ internal sealed class AgentActor( cancellationToken).ConfigureAwait(false); this._etag = response.ETag; + var hasExistingThread = false; if (response.Results[0] is GetValueResult threadResult) { if (threadResult.Value is { } threadJson) { // Deserialize the thread state if it exists - await agent.DeserializeThreadAsync(threadJson, cancellationToken: cancellationToken).ConfigureAwait(false); + this._thread = await agent.DeserializeThreadAsync(threadJson, cancellationToken: cancellationToken).ConfigureAwait(false); + hasExistingThread = true; } } this._thread ??= agent.GetNewThread(); - Log.ThreadStateRestored(logger, context.ActorId.ToString(), response.Results[0] is GetValueResult { Value: not null }); + Log.ThreadStateRestored(logger, context.ActorId.ToString(), hasExistingThread); while (!cancellationToken.IsCancellationRequested) { @@ -120,10 +122,9 @@ internal sealed class AgentActor( Log.AgentStreamingUpdate(logger, requestId, i); } - var serializedRunResponse = JsonSerializer.SerializeToElement( - updates.ToAgentRunResponse(), - AIJsonUtilities.DefaultOptions.GetTypeInfo(typeof(AgentRunResponse))); - var updatedThread = JsonSerializer.SerializeToElement(this._thread, AgentHostingJsonUtilities.DefaultOptions.GetTypeInfo(typeof(AgentThread))); + var serializedRunResponse = JsonSerializer.SerializeToElement(updates.ToAgentRunResponse(), AIJsonUtilities.DefaultOptions.GetTypeInfo(typeof(AgentRunResponse))); + var updatedThread = await this._thread.SerializeAsync(AgentHostingJsonUtilities.DefaultOptions, cancellationToken).ConfigureAwait(false); + var writeResponse = await context.WriteAsync( new(this._etag, [ diff --git a/dotnet/tests/Microsoft.Extensions.AI.Agents.Hosting.UnitTests/AgentActorTests.cs b/dotnet/tests/Microsoft.Extensions.AI.Agents.Hosting.UnitTests/AgentActorTests.cs index d4d8901c57..1584960e9b 100644 --- a/dotnet/tests/Microsoft.Extensions.AI.Agents.Hosting.UnitTests/AgentActorTests.cs +++ b/dotnet/tests/Microsoft.Extensions.AI.Agents.Hosting.UnitTests/AgentActorTests.cs @@ -1,5 +1,9 @@ // Copyright (c) Microsoft. All rights reserved. +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.AI.Agents.Runtime; using Microsoft.Extensions.Logging; @@ -19,17 +23,174 @@ public class AgentActorTests [Fact] public async Task DisposeAsync_NoException_CompletesSuccessfullyAsync() { - // Arrange var mockAgent = new Mock(); var mockContext = new Mock(); var mockLogger = NullLoggerFactory.Instance.CreateLogger(); var actor = new AgentActor(mockAgent.Object, mockContext.Object, mockLogger); - // Act var valueTask = actor.DisposeAsync(); - // Assert Assert.True(valueTask.IsCompleted, "DisposeAsync should return a completed ValueTask."); await valueTask; } + + /// + /// Verifies that when no thread state exists, GetNewThread is called. + /// + [Fact] + public async Task RunAsync_WithNoExistingThreadState_CallsGetNewThreadAsync() + { + var expectedThread = new AgentThread { ConversationId = "new-thread-id" }; + + var mockAgent = new Mock(); + mockAgent.Setup(a => a.GetNewThread()).Returns(expectedThread); + + var mockContext = new Mock(); + var actorId = new ActorId("TestAgent", "test-instance"); + mockContext.Setup(c => c.ActorId).Returns(actorId); + + // Setup ReadAsync to return no existing thread state + var readResponse = new ReadResponse("test-etag", [new GetValueResult(null)]); + mockContext.Setup(c => c.ReadAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(readResponse); + + // Setup WatchMessagesAsync to return empty sequence to prevent infinite loop + mockContext.Setup(c => c.WatchMessagesAsync(It.IsAny())) + .Returns(CreateEmptyAsyncEnumerableAsync()); + + var mockLogger = NullLoggerFactory.Instance.CreateLogger(); + await using var actor = new AgentActor(mockAgent.Object, mockContext.Object, mockLogger); + + using var cts = new CancellationTokenSource(); + cts.CancelAfter(TimeSpan.FromMilliseconds(100)); // Cancel quickly to exit the loop + + await actor.RunAsync(cts.Token); + + mockAgent.Verify(a => a.GetNewThread(), Times.Once); + } + + /// + /// Verifies that when ReadAsync throws an exception, the actor handles it gracefully. + /// + [Fact] + public async Task RunAsync_WhenReadAsyncThrows_HandlesExceptionGracefullyAsync() + { + var mockAgent = new Mock(); + var mockContext = new Mock(); + var actorId = new ActorId("TestAgent", "test-instance"); + mockContext.Setup(c => c.ActorId).Returns(actorId); + + mockContext.Setup(c => c.ReadAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("Read failed")); + + var mockLogger = NullLoggerFactory.Instance.CreateLogger(); + await using var actor = new AgentActor(mockAgent.Object, mockContext.Object, mockLogger); + + using var cts = new CancellationTokenSource(); + + await Assert.ThrowsAsync(async () => + await actor.RunAsync(cts.Token)); + + mockAgent.Verify(a => a.GetNewThread(), Times.Never); + } + + /// + /// Verifies that the thread assignment works correctly when processing an agent request. + /// This test checks that the thread used in the agent request is properly assigned. + /// + [Fact] + public async Task HandleAgentRequest_UsesCorrectThreadAsync() + { + var threadJson = JsonSerializer.SerializeToElement(new { conversationId = "expected-thread-id" }); + + var testAgent = new TestAgent(); + + var mockContext = new Mock(); + var actorId = new ActorId("TestAgent", "test-instance"); + mockContext.Setup(c => c.ActorId).Returns(actorId); + + var readResponse = new ReadResponse("test-etag", [new GetValueResult(threadJson)]); + mockContext.Setup(c => c.ReadAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(readResponse); + + // Create a request message + var requestMessage = new ActorRequestMessage("test-message-id") + { + SenderId = actorId, + Method = AgentActorConstants.RunMethodName, + Params = JsonSerializer.SerializeToElement(new AgentRunRequest + { + Messages = [new ChatMessage(ChatRole.User, "Test message")] + }) + }; + + var messageSequence = CreateAsyncEnumerableAsync(new List { requestMessage }); + mockContext.Setup(c => c.WatchMessagesAsync(It.IsAny())) + .Returns(messageSequence); + + mockContext.Setup(c => c.WriteAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new WriteResponse("new-etag", true)); + + var mockLogger = NullLoggerFactory.Instance.CreateLogger(); + await using var actor = new AgentActor(testAgent, mockContext.Object, mockLogger); + + using var cts = new CancellationTokenSource(); + + cts.CancelAfter(TimeSpan.FromSeconds(1)); + + await actor.RunAsync(cts.Token); + + Assert.True(testAgent.RunStreamingAsyncCalled, "RunStreamingAsync should have been called"); + + // Verify the thread was used in RunStreamingAsync and has the expected ID + Assert.NotNull(testAgent.ThreadUsedInRunStreamingAsync); + Assert.Equal("expected-thread-id", testAgent.ThreadUsedInRunStreamingAsync.ConversationId); + } + + /// + /// Helper method to create an empty async enumerable. + /// + private static async IAsyncEnumerable CreateEmptyAsyncEnumerableAsync() + { + await Task.CompletedTask; + yield break; + } + + /// + /// Helper method to create an async enumerable from a list. + /// + private static async IAsyncEnumerable CreateAsyncEnumerableAsync(IEnumerable items) + { + foreach (var item in items) + { + yield return item; + } + } + + /// + /// Test agent implementation to track method calls. + /// + private sealed class TestAgent : AIAgent + { + public bool RunStreamingAsyncCalled { get; private set; } + public AgentThread? ThreadUsedInRunStreamingAsync { get; private set; } + + public override Task RunAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + { + this.ThreadUsedInRunStreamingAsync = thread; + return Task.FromResult(new AgentRunResponse + { + Messages = [new ChatMessage(ChatRole.Assistant, "Test response")] + }); + } + + public override async IAsyncEnumerable RunStreamingAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) + { + this.RunStreamingAsyncCalled = true; + this.ThreadUsedInRunStreamingAsync = thread; + + yield return new AgentRunResponseUpdate(ChatRole.Assistant, "Test response"); + await Task.CompletedTask; + } + } }