.Net Agent Runtime Fix: Allow multiple subscribers to handle events concurrently (#12713)

### 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.
-->

Enable multiple subscribers to to handle events concurrently (instead of
sequentially)

Fixes: #12701

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

The event handling loop of the `InProcessRuntime` is waiting on each
subscriber to handle the event. This conflicts with performance
expectations.

### 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/src/Agents/Runtime/InProcess.Tests/PublishMessageTests.cs

@@ -1,7 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System;
-using System.Linq;
 using System.Threading.Tasks;
 using FluentAssertions;
 using Xunit;
@@ -56,10 +55,7 @@ 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<AggregateException>())
-                                   .Which.Should().Match<AggregateException>(
-                                        exception => exception.InnerExceptions.Count == 2 &&
-                                              exception.InnerExceptions.All(exception => exception is TestException));
+        await publishTask.Should().ThrowAsync<TestException>();
 
         fixture.GetAgentInstances<ErrorAgent>().Values
             .Should().HaveCount(2)
@@ -81,11 +77,7 @@ 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<AggregateException>())
-                                   .Which.Should().Match<AggregateException>(
-                                        exception => exception.InnerExceptions.Count == 2 &&
-                                              exception.InnerExceptions.All(
-                                                exception => exception is TestException));
+        await publicTask.Should().ThrowAsync<TestException>();
 
         fixture.GetAgentInstances<ReceiverAgent>().Values
             .Should().HaveCount(2, "Two ReceiverAgents should have been created")

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

@@ -352,45 +352,41 @@ private async ValueTask PublishMessageServicerAsync(MessageEnvelope envelope, Ca
             throw new InvalidOperationException("Message must have a topic to be published.");
         }
 
-        List<Exception> exceptions = [];
+        List<Task>? tasks = null;
         TopicId topic = envelope.Topic.Value;
         foreach (ISubscriptionDefinition subscription in this._subscriptions.Values.Where(subscription => subscription.Matches(topic)))
         {
-            try
-            {
-                deliveryToken.ThrowIfCancellationRequested();
+            (tasks ??= []).Add(ProcessSubscriptionAsync(envelope, topic, subscription, deliveryToken));
+        }
 
-                AgentId? sender = envelope.Sender;
+        if (tasks is not null)
+        {
+            await Task.WhenAll(tasks).ConfigureAwait(false);
+        }
 
-                using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(envelope.Cancellation, deliveryToken);
-                MessageContext messageContext = new(envelope.MessageId, combinedSource.Token)
-                {
-                    Sender = sender,
-                    Topic = topic,
-                    IsRpc = false
-                };
+        async Task ProcessSubscriptionAsync(MessageEnvelope envelope, TopicId topic, ISubscriptionDefinition subscription, CancellationToken deliveryToken)
+        {
+            deliveryToken.ThrowIfCancellationRequested();
 
-                AgentId agentId = subscription.MapToAgent(topic);
-                if (!this.DeliverToSelf && sender.HasValue && sender == agentId)
-                {
-                    continue;
-                }
+            AgentId? sender = envelope.Sender;
 
-                IHostableAgent agent = await this.EnsureAgentAsync(agentId).ConfigureAwait(false);
+            using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(envelope.Cancellation, deliveryToken);
+            MessageContext messageContext = new(envelope.MessageId, combinedSource.Token)
+            {
+                Sender = sender,
+                Topic = topic,
+                IsRpc = false
+            };
 
-                // TODO: Cancellation propagation!
-                await agent.OnMessageAsync(envelope.Message, messageContext).ConfigureAwait(false);
-            }
-            catch (Exception ex) when (!ex.IsCriticalException())
+            AgentId agentId = subscription.MapToAgent(topic);
+            if (!this.DeliverToSelf && sender.HasValue && sender == agentId)
             {
-                exceptions.Add(ex);
+                return;
             }
-        }
 
-        if (exceptions.Count > 0)
-        {
-            // TODO: Unwrap TargetInvocationException?
-            throw new AggregateException("One or more exceptions occurred while processing the message.", exceptions);
+            IHostableAgent agent = await this.EnsureAgentAsync(agentId).ConfigureAwait(false);
+
+            await agent.OnMessageAsync(envelope.Message, messageContext).ConfigureAwait(false);
         }
     }