.Net Agent Orchestration - Fix GroupChat Samples (#12794)

### Motivation and Context
<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->

Fixes: #12731
Fixes: #12646
Fixes: #12592
Fixes: #12626

Orchestration not providing the expected result.

### Description
<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

Undefined termination condition results in orchestration always running
until the maximum number of turns. This result in blank messages present
in the history as the agents have no completed the task.

In addition, the orchestration result for the sample is not the last
message (which is the default for `RoundRobinGroupChatManager`). The
result is always the writer output.

Also:
- Addresses bizarre issue in `AgentActor` where response variable
assignment from function closure evaluate to null.
- Convert `PublishMessageAsync` to true "fire and forget"

### Contribution Checklist
<!-- Before submitting this PR, please make sure: -->

- [X] The code builds clean without any errors or warnings
- [X] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [X] All unit tests pass, and I have added new tests where possible
- [X] I didn't break anyone 😄

Author: crickman

dotnet/samples/GettingStartedWithAgents/Orchestration/Step03_GroupChat.cs

@@ -5,6 +5,7 @@
 using Microsoft.SemanticKernel.Agents.Orchestration;
 using Microsoft.SemanticKernel.Agents.Orchestration.GroupChat;
 using Microsoft.SemanticKernel.Agents.Runtime.InProcess;
+using Microsoft.SemanticKernel.ChatCompletion;
 
 namespace GettingStarted.Orchestration;
 
@@ -47,7 +48,7 @@ Consider suggestions when refining an idea.
                 """
                 You are an art director who has opinions about copywriting born of a love for David Ogilvy.
                 The goal is to determine if the given copy is acceptable to print.
-                If so, state that it is approved.
+                If so, state: "I Approve".
                 If not, provide insight on how to refine suggested copy without example.
                 """);
 
@@ -57,7 +58,7 @@ Consider suggestions when refining an idea.
         OrchestrationMonitor monitor = new();
         // Define the orchestration
         GroupChatOrchestration orchestration =
-            new(new RoundRobinGroupChatManager()
+            new(new AuthorCriticManager(writer.Name!, editor.Name!)
             {
                 MaximumInvocationCount = 5
             },
@@ -87,4 +88,31 @@ Consider suggestions when refining an idea.
             this.WriteAgentChatMessage(message);
         }
     }
+
+    private sealed class AuthorCriticManager(string authorName, string criticName) : RoundRobinGroupChatManager
+    {
+        public override ValueTask<GroupChatManagerResult<string>> FilterResults(ChatHistory history, CancellationToken cancellationToken = default)
+        {
+            ChatMessageContent finalResult = history.Last(message => message.AuthorName == authorName);
+            return ValueTask.FromResult(new GroupChatManagerResult<string>($"{finalResult}") { Reason = "The approved copy." });
+        }
+
+        /// <inheritdoc/>
+        public override async ValueTask<GroupChatManagerResult<bool>> ShouldTerminate(ChatHistory history, CancellationToken cancellationToken = default)
+        {
+            // Has the maximum invocation count been reached?
+            GroupChatManagerResult<bool> result = await base.ShouldTerminate(history, cancellationToken);
+            if (!result.Value)
+            {
+                // If not, check if the reviewer has approved the copy.
+                ChatMessageContent? lastMessage = history.LastOrDefault();
+                if (lastMessage is not null && lastMessage.AuthorName == criticName && $"{lastMessage}".Contains("I Approve", StringComparison.OrdinalIgnoreCase))
+                {
+                    // If the reviewer approves, we terminate the chat.
+                    result = new GroupChatManagerResult<bool>(true) { Reason = "The reviewer has approved the copy." };
+                }
+            }
+            return result;
+        }
+    }
 }

dotnet/samples/GettingStartedWithAgents/Orchestration/Step03a_GroupChatWithHumanInTheLoop.cs

@@ -39,7 +39,7 @@ Consider suggestions when refining an idea.
                 """
                 You are an art director who has opinions about copywriting born of a love for David Ogilvy.
                 The goal is to determine if the given copy is acceptable to print.
-                If so, state that it is approved.
+                If so, state: "I Approve".
                 If not, provide insight on how to refine suggested copy without example.
                 """);
 
@@ -49,14 +49,17 @@ Consider suggestions when refining an idea.
         OrchestrationMonitor monitor = new();
 
         // Define the orchestration
+        bool didUserRespond = false;
         GroupChatOrchestration orchestration =
             new(
-                new CustomRoundRobinGroupChatManager()
+                new HumanInTheLoopChatManager(writer.Name!, editor.Name!)
                 {
                     MaximumInvocationCount = 5,
                     InteractiveCallback = () =>
                     {
-                        ChatMessageContent input = new(AuthorRole.User, "I like it");
+                        // Simlulate user input that first replies "No" and then "Yes"
+                        ChatMessageContent input = new(AuthorRole.User, didUserRespond ? "Yes" : "More pizzazz");
+                        didUserRespond = true;
                         Console.WriteLine($"\n# INPUT: {input.Content}\n");
                         return ValueTask.FromResult(input);
                     }
@@ -89,18 +92,42 @@ Consider suggestions when refining an idea.
     /// User input is achieved by overriding the default round robin manager
     /// to allow user input after the reviewer agent's message.
     /// </remarks>
-    private sealed class CustomRoundRobinGroupChatManager : RoundRobinGroupChatManager
+    private sealed class HumanInTheLoopChatManager(string authorName, string criticName) : RoundRobinGroupChatManager
     {
+        public override ValueTask<GroupChatManagerResult<string>> FilterResults(ChatHistory history, CancellationToken cancellationToken = default)
+        {
+            ChatMessageContent finalResult = history.Last(message => message.AuthorName == authorName);
+            return ValueTask.FromResult(new GroupChatManagerResult<string>($"{finalResult}") { Reason = "The approved copy." });
+        }
+
+        /// <inheritdoc/>
+        public override async ValueTask<GroupChatManagerResult<bool>> ShouldTerminate(ChatHistory history, CancellationToken cancellationToken = default)
+        {
+            // Has the maximum invocation count been reached?
+            GroupChatManagerResult<bool> result = await base.ShouldTerminate(history, cancellationToken);
+            if (!result.Value)
+            {
+                // If not, check if the reviewer has approved the copy.
+                ChatMessageContent? lastMessage = history.LastOrDefault();
+                if (lastMessage is not null && lastMessage.Role == AuthorRole.User && $"{lastMessage}".Contains("Yes", StringComparison.OrdinalIgnoreCase))
+                {
+                    // If the reviewer approves, we terminate the chat.
+                    result = new GroupChatManagerResult<bool>(true) { Reason = "The user is satisfied with the copy." };
+                }
+            }
+            return result;
+        }
+
         public override ValueTask<GroupChatManagerResult<bool>> ShouldRequestUserInput(ChatHistory history, CancellationToken cancellationToken = default)
         {
-            string? lastAgent = history.LastOrDefault()?.AuthorName;
+            ChatMessageContent? lastMessage = history.LastOrDefault();
 
-            if (lastAgent is null)
+            if (lastMessage is null)
             {
                 return ValueTask.FromResult(new GroupChatManagerResult<bool>(false) { Reason = "No agents have spoken yet." });
             }
 
-            if (lastAgent == "Reviewer")
+            if (lastMessage is not null && lastMessage.AuthorName == criticName && $"{lastMessage}".Contains("I Approve", StringComparison.OrdinalIgnoreCase))
             {
                 return ValueTask.FromResult(new GroupChatManagerResult<bool>(true) { Reason = "User input is needed after the reviewer's message." });
             }

dotnet/samples/GettingStartedWithAgents/Orchestration/Step06_DifferentAgentTypes.cs

@@ -161,7 +161,7 @@ await this.CreateOpenAIAssistantAgentAsync(
                 """
                 You are an art director who has opinions about copywriting born of a love for David Ogilvy.
                 The goal is to determine if the given copy is acceptable to print.
-                If so, state that it is approved.
+                If so, state: "I Approve".
                 If not, provide insight on how to refine suggested copy without example.
                 """);
 
@@ -171,7 +171,7 @@ await this.CreateOpenAIAssistantAgentAsync(
         OrchestrationMonitor monitor = new();
         // Define the orchestration
         GroupChatOrchestration orchestration =
-            new(new RoundRobinGroupChatManager()
+            new(new AuthorCriticManager(writer.Name!, editor.Name!)
             {
                 MaximumInvocationCount = 5
             },
@@ -285,7 +285,6 @@ public async Task HandoffOrchestrationAsync()
         }
     }
 
-    #region private
     private sealed class OrderStatusPlugin
     {
         [KernelFunction]
@@ -303,5 +302,31 @@ private sealed class OrderRefundPlugin
         [KernelFunction]
         public string ProcessReturn(string orderId, string reason) => $"Refund for order {orderId} has been processed successfully.";
     }
-    #endregion
+
+    private sealed class AuthorCriticManager(string authorName, string criticName) : RoundRobinGroupChatManager
+    {
+        public override ValueTask<GroupChatManagerResult<string>> FilterResults(ChatHistory history, CancellationToken cancellationToken = default)
+        {
+            ChatMessageContent finalResult = history.Last(message => message.AuthorName == authorName);
+            return ValueTask.FromResult(new GroupChatManagerResult<string>($"{finalResult}") { Reason = "The approved copy." });
+        }
+
+        /// <inheritdoc/>
+        public override async ValueTask<GroupChatManagerResult<bool>> ShouldTerminate(ChatHistory history, CancellationToken cancellationToken = default)
+        {
+            // Has the maximum invocation count been reached?
+            GroupChatManagerResult<bool> result = await base.ShouldTerminate(history, cancellationToken);
+            if (!result.Value)
+            {
+                // If not, check if the reviewer has approved the copy.
+                ChatMessageContent? lastMessage = history.LastOrDefault();
+                if (lastMessage is not null && lastMessage.AuthorName == criticName && $"{lastMessage}".Contains("I Approve", StringComparison.OrdinalIgnoreCase))
+                {
+                    // If the reviewer approves, we terminate the chat.
+                    result = new GroupChatManagerResult<bool>(true) { Reason = "The reviewer has approved the copy." };
+                }
+            }
+            return result;
+        }
+    }
 }

dotnet/src/Agents/Orchestration/AgentActor.cs

@@ -17,6 +17,7 @@ namespace Microsoft.SemanticKernel.Agents.Orchestration;
 public abstract class AgentActor : OrchestrationActor
 {
     private AgentInvokeOptions? _options;
+    private ChatMessageContent? _lastResponse;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="AgentActor"/> class.
@@ -94,9 +95,9 @@ protected async ValueTask<ChatMessageContent> InvokeAsync(IList<ChatMessageConte
     {
         this.Context.Cancellation.ThrowIfCancellationRequested();
 
-        ChatMessageContent? response = null;
+        this._lastResponse = null;
 
-        AgentInvokeOptions options = this.GetInvokeOptions(HandleMessage);
+        AgentInvokeOptions options = this.GetInvokeOptions(HandleMessageAsync);
         if (this.Context.StreamingResponseCallback == null)
         {
             // No need to utilize streaming if no callback is provided
@@ -107,11 +108,11 @@ protected async ValueTask<ChatMessageContent> InvokeAsync(IList<ChatMessageConte
             await this.InvokeStreamingAsync(input, options, cancellationToken).ConfigureAwait(false);
         }
 
-        return response ?? new ChatMessageContent(AuthorRole.Assistant, string.Empty);
+        return this._lastResponse ?? new ChatMessageContent(AuthorRole.Assistant, string.Empty);
 
-        async Task HandleMessage(ChatMessageContent message)
+        async Task HandleMessageAsync(ChatMessageContent message)
         {
-            response = message; // Keep track of most recent response for both invocation modes
+            this._lastResponse = message; // Keep track of most recent response for both invocation modes
 
             if (this.Context.ResponseCallback is not null && !this.ResponseCallbackFilter(message))
             {

dotnet/src/Agents/Orchestration/GroupChat/RoundRobinGroupChatManager.cs

@@ -11,7 +11,7 @@ namespace Microsoft.SemanticKernel.Agents.Orchestration.GroupChat;
 /// A <see cref="GroupChatManager"/> that selects agents in a round-robin fashion.
 /// </summary>
 /// <remarks>
-/// Subclass this class to customize filter and user interaction behavior.
+/// Subclass this class to customize filter, termination, and user interaction behaviors.
 /// </remarks>
 public class RoundRobinGroupChatManager : GroupChatManager
 {

dotnet/src/Agents/Runtime/Core.Tests/AgentRuntimeExtensionsTests.cs

@@ -86,6 +86,7 @@ public async Task RegisterImplicitAgentSubscriptionsAsync()
             string messageText2 = "Test message #1";
             await runtime.PublishMessageAsync(messageText1, topic1);
             await runtime.PublishMessageAsync(messageText2, topic2);
+            await runtime.RunUntilIdleAsync();
 
             // Get agent and verify it received messages
             AgentId registeredId = await runtime.GetAgentAsync(agentTypeName, lazy: false);

dotnet/src/Agents/Runtime/InProcess.Tests/PublishMessageTests.cs

@@ -37,8 +37,8 @@ public async Task Test_PublishMessage_SingleFailure()
 
         Func<Task> publishTask = async () => await fixture.RunPublishTestAsync(new TopicId("TestTopic"), new BasicMessage { Content = "1" });
 
-        // Test that we wrap single errors appropriately
-        await publishTask.Should().ThrowAsync<TestException>();
+        // Publish is fire and forget, so we expect no exception to be thrown
+        await publishTask.Should().NotThrowAsync();
 
         fixture.GetAgentInstances<ErrorAgent>().Values.Should().ContainSingle()
                                                 .Which.DidThrow.Should().BeTrue("Agent should have thrown an exception");
@@ -54,8 +54,8 @@ public async Task Test_PublishMessage_MultipleFailures()
 
         Func<Task> publishTask = async () => await fixture.RunPublishTestAsync(new TopicId("TestTopic"), new BasicMessage { Content = "1" });
 
-        // What we are really testing here is that a single exception does not prevent sending to the remaining agents
-        await publishTask.Should().ThrowAsync<TestException>();
+        // Publish is fire and forget, so we expect no exception to be thrown
+        await publishTask.Should().NotThrowAsync();
 
         fixture.GetAgentInstances<ErrorAgent>().Values
             .Should().HaveCount(2)
@@ -76,8 +76,8 @@ public async Task Test_PublishMessage_MixedSuccessFailure()
 
         Func<Task> publicTask = async () => await fixture.RunPublishTestAsync(new TopicId("TestTopic"), new BasicMessage { Content = "1" });
 
-        // What we are really testing here is that raising exceptions does not prevent sending to the remaining agents
-        await publicTask.Should().ThrowAsync<TestException>();
+        // Publish is fire and forget, so we expect no exception to be thrown
+        await publicTask.Should().NotThrowAsync();
 
         fixture.GetAgentInstances<ReceiverAgent>().Values
             .Should().HaveCount(2, "Two ReceiverAgents should have been created")

dotnet/src/Agents/Runtime/InProcess/InProcessRuntime.cs

@@ -102,7 +102,7 @@ public async Task RunUntilIdleAsync()
     /// <inheritdoc/>
     public ValueTask PublishMessageAsync(object message, TopicId topic, AgentId? sender = null, string? messageId = null, CancellationToken cancellationToken = default)
     {
-        return this.ExecuteTracedAsync(async () =>
+        return this.ExecuteTracedAsync(() =>
         {
             MessageDelivery delivery =
                 new MessageEnvelope(message, messageId, cancellationToken)
@@ -112,7 +112,11 @@ public ValueTask PublishMessageAsync(object message, TopicId topic, AgentId? sen
             this._messageDeliveryQueue.Enqueue(delivery);
             Interlocked.Increment(ref this.messageQueueCount);
 
-            await delivery.ResultSink.Future.ConfigureAwait(false);
+#if !NETCOREAPP
+            return Task.CompletedTask.AsValueTask();
+#else
+            return ValueTask.CompletedTask;
+#endif
         });
     }