Improve e2e integration tests and isolate tests from other things; includes patch to Serializer

dotnet/AutoGen.sln

@@ -140,7 +140,13 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AutoGen.AgentChat
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AutoGen.AgentChat.Tests", "test\Microsoft.AutoGen.AgentChat.Tests\Microsoft.AutoGen.AgentChat.Tests.csproj", "{217A4F86-8ADD-4998-90BA-880092A019F5}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "HelloAgent.AppHost", "test\Microsoft.AutoGen.Integration.Tests.AppHosts\HelloAgent.AppHost\HelloAgent.AppHost.csproj", "{0C371D65-7EF9-44EA-8128-A105DA82A80E}"
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Microsoft.AutoGen.Integration.Tests.AppHosts", "Microsoft.AutoGen.Integration.Tests.AppHosts", "{D1C2B0BB-1276-4146-A699-D1983AE8ED04}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "HelloAgentTests", "test\Microsoft.AutoGen.Integration.Tests.AppHosts\HelloAgentTests\HelloAgentTests.csproj", "{CD10E29A-725E-4BEF-9CFF-6C0E0A652926}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "InMemoryTests.AppHost", "test\Microsoft.AutoGen.Integration.Tests.AppHosts\InMemoryTests.AppHost\InMemoryTests.AppHost.csproj", "{1E4E1ED4-7701-4A05-A861-64461C3B1EE3}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "XlangTests.AppHost", "test\Microsoft.AutoGen.Integration.Tests.AppHosts\XLangTests.AppHost\XlangTests.AppHost.csproj", "{62CDFB27-3B02-4D4B-B789-8AAD5E20688A}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -372,6 +378,18 @@ Global
 		{0C371D65-7EF9-44EA-8128-A105DA82A80E}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{0C371D65-7EF9-44EA-8128-A105DA82A80E}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{0C371D65-7EF9-44EA-8128-A105DA82A80E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{CD10E29A-725E-4BEF-9CFF-6C0E0A652926}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{CD10E29A-725E-4BEF-9CFF-6C0E0A652926}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{CD10E29A-725E-4BEF-9CFF-6C0E0A652926}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{CD10E29A-725E-4BEF-9CFF-6C0E0A652926}.Release|Any CPU.Build.0 = Release|Any CPU
+		{1E4E1ED4-7701-4A05-A861-64461C3B1EE3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{1E4E1ED4-7701-4A05-A861-64461C3B1EE3}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{1E4E1ED4-7701-4A05-A861-64461C3B1EE3}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{1E4E1ED4-7701-4A05-A861-64461C3B1EE3}.Release|Any CPU.Build.0 = Release|Any CPU
+		{62CDFB27-3B02-4D4B-B789-8AAD5E20688A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{62CDFB27-3B02-4D4B-B789-8AAD5E20688A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{62CDFB27-3B02-4D4B-B789-8AAD5E20688A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{62CDFB27-3B02-4D4B-B789-8AAD5E20688A}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -436,7 +454,10 @@ Global
 		{EF954ED3-87D5-40F1-8557-E7179F43EA0E} = {18BF8DD7-0585-48BF-8F97-AD333080CE06}
 		{7F828599-56E8-4597-8F68-EE26FD631417} = {18BF8DD7-0585-48BF-8F97-AD333080CE06}
 		{217A4F86-8ADD-4998-90BA-880092A019F5} = {F823671B-3ECA-4AE6-86DA-25E920D3FE64}
-		{0C371D65-7EF9-44EA-8128-A105DA82A80E} = {F42F9C8E-7BD9-4687-9B63-AFFA461AF5C1}
+		{D1C2B0BB-1276-4146-A699-D1983AE8ED04} = {F823671B-3ECA-4AE6-86DA-25E920D3FE64}
+		{CD10E29A-725E-4BEF-9CFF-6C0E0A652926} = {D1C2B0BB-1276-4146-A699-D1983AE8ED04}
+		{1E4E1ED4-7701-4A05-A861-64461C3B1EE3} = {D1C2B0BB-1276-4146-A699-D1983AE8ED04}
+		{62CDFB27-3B02-4D4B-B789-8AAD5E20688A} = {D1C2B0BB-1276-4146-A699-D1983AE8ED04}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {93384647-528D-46C8-922C-8DB36A382F0B}

dotnet/samples/Hello/HelloAgent/HelloAgent.csproj

@@ -23,6 +23,7 @@
     <ProjectReference Include="..\..\..\src\Microsoft.AutoGen\Contracts\Microsoft.AutoGen.Contracts.csproj" />
     <ProjectReference Include="..\..\..\src\Microsoft.AutoGen\Core.Grpc\Microsoft.AutoGen.Core.Grpc.csproj" />
     <ProjectReference Include="..\..\..\src\Microsoft.AutoGen\Core\Microsoft.AutoGen.Core.csproj" />
+    <ProjectReference Include="..\..\..\src\Microsoft.AutoGen\Core.Grpc\Microsoft.AutoGen.Core.Grpc.csproj" />
   </ItemGroup>
 
   <ItemGroup>

dotnet/samples/Hello/HelloAgent/Program.cs

@@ -4,14 +4,22 @@
 using Microsoft.AutoGen.Agents;
 using Microsoft.AutoGen.Contracts;
 using Microsoft.AutoGen.Core;
+using Microsoft.AutoGen.Core.Grpc;
 using Samples;
-
-// Set up app builder for in-process runtime, allow message delivery to self, and add the Hello agent
-AgentsAppBuilder appBuilder = new AgentsAppBuilder()
-    .UseInProcessRuntime(deliverToSelf: true)
-    .AddAgent<HelloAgent>("HelloAgent");
+var appBuilder = new AgentsAppBuilder(); // Create app builder
+// if we are using distributed, we need the AGENT_HOST var defined and then we will use the grpc runtime
+if (Environment.GetEnvironmentVariable("AGENT_HOST") is string agentHost)
+{
+    appBuilder.AddGrpcAgentWorker(agentHost)
+        .AddAgent<HelloAgent>("HelloAgent");
+}
+else
+{
+    // Set up app builder for in-process runtime, allow message delivery to self, and add the Hello agent
+    appBuilder.UseInProcessRuntime(deliverToSelf: true).AddAgent<HelloAgent>("HelloAgent");
+}
 var app = await appBuilder.BuildAsync(); // Build the app
 // Create a custom message type from proto and define message
-NewMessageReceived message = new NewMessageReceived { Message = "Hello World!" };
-await app.PublishMessageAsync(message, new TopicId("HelloTopic")); // Publish custom message (handler has been set in HelloAgent)
-await app.WaitForShutdownAsync(); // Wait for shutdown from agent
+var message = new NewMessageReceived { Message = "Hello World!" };
+await app.PublishMessageAsync(message, new TopicId("HelloTopic")).ConfigureAwait(false); // Publish custom message (handler has been set in HelloAgent)
+await app.WaitForShutdownAsync().ConfigureAwait(false); // Wait for shutdown from agent

dotnet/src/Microsoft.AutoGen/Agents/IOAgent/ConsoleAgent/IHandleConsole.cs

@@ -1,6 +1,5 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // IHandleConsole.cs
-using Google.Protobuf;
 using Microsoft.AutoGen.Contracts;
 
 namespace Microsoft.AutoGen.Agents;
@@ -14,13 +13,12 @@
     /// <summary>
     /// Prototype for Publish Message Async method
     /// </summary>
-    /// <typeparam name="T"></typeparam>
     /// <param name="message"></param>
     /// <param name="topic"></param>
     /// <param name="messageId"></param>
-    /// <param name="token"></param>
+    /// <param name="cancellationToken"></param>
     /// <returns>ValueTask</returns>
-    ValueTask PublishMessageAsync<T>(T message, TopicId topic, string? messageId, CancellationToken token = default) where T : IMessage;
+    ValueTask PublishMessageAsync(object message, TopicId topic, string? messageId = null, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Receives events of type Output and writes them to the console
@@ -39,7 +37,7 @@
         {
             Route = "console"
         };
-        await PublishMessageAsync(evt, new TopicId("OutputWritten"), null, token: CancellationToken.None).ConfigureAwait(false);
+        await PublishMessageAsync(evt, new TopicId("OutputWritten"), null, cancellationToken: CancellationToken.None).ConfigureAwait(false);
     }
 
     /// <summary>
@@ -60,6 +58,6 @@
         {
             Route = "console"
         };
-        await PublishMessageAsync(evt, new TopicId("InputProcessed"), null, token: CancellationToken.None).ConfigureAwait(false);
+        await PublishMessageAsync(evt, new TopicId("InputProcessed"), null, cancellationToken: CancellationToken.None).ConfigureAwait(false);
     }
 }

dotnet/src/Microsoft.AutoGen/Agents/IOAgent/FileAgent/IHandleFileIO.cs

@@ -1,7 +1,5 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // IHandleFileIO.cs
-
-using Google.Protobuf;
 using Microsoft.AutoGen.Contracts;
 using Microsoft.Extensions.Logging;
 
@@ -25,13 +23,12 @@
     /// <summary>
     /// Prototype for Publish Message Async method
     /// </summary>
-    /// <typeparam name="T"></typeparam>
     /// <param name="message"></param>
     /// <param name="topic"></param>
     /// <param name="messageId"></param>
-    /// <param name="token"></param>
+    /// <param name="cancellationToken"></param>
     /// <returns>ValueTask</returns>
-    ValueTask PublishMessageAsync<T>(T message, TopicId topic, string? messageId, CancellationToken token = default) where T : IMessage;
+    ValueTask PublishMessageAsync(object message, TopicId topic, string? messageId = null, CancellationToken cancellationToken = default);
     async ValueTask IHandle<Input>.HandleAsync(Input item, MessageContext messageContext)
     {
 
@@ -45,7 +42,7 @@
             {
                 Message = errorMessage
             };
-            await PublishMessageAsync(err, new TopicId("IOError"), null, token: CancellationToken.None).ConfigureAwait(false);
+            await PublishMessageAsync(err, new TopicId("IOError"), null, cancellationToken: CancellationToken.None).ConfigureAwait(false);
             return;
         }
         string content;
@@ -58,7 +55,7 @@
         {
             Route = Route
         };
-        await PublishMessageAsync(evt, new TopicId("InputProcessed"), null, token: CancellationToken.None).ConfigureAwait(false);
+        await PublishMessageAsync(evt, new TopicId("InputProcessed"), null, cancellationToken: CancellationToken.None).ConfigureAwait(false);
     }
     async ValueTask IHandle<Output>.HandleAsync(Output item, MessageContext messageContext)
     {
@@ -70,6 +67,6 @@
         {
             Route = Route
         };
-        await PublishMessageAsync(evt, new TopicId("OutputWritten"), null, token: CancellationToken.None).ConfigureAwait(false);
+        await PublishMessageAsync(evt, new TopicId("OutputWritten"), null, cancellationToken: CancellationToken.None).ConfigureAwait(false);
     }
 }

dotnet/src/Microsoft.AutoGen/Core.Grpc/ProtobufSerializationRegistry.cs

@@ -28,10 +28,7 @@ public bool Exists(Type type)
 
     public void RegisterSerializer(Type type, IProtobufMessageSerializer serializer)
     {
-        if (_serializers.ContainsKey(TypeNameResolver.ResolveTypeName(type)))
-        {
-            throw new InvalidOperationException($"Serializer already registered for {type.FullName}");
-        }
+        _serializers.TryAdd(TypeNameResolver.ResolveTypeName(type), serializer);
         _serializers[TypeNameResolver.ResolveTypeName(type)] = serializer;
     }
 }

dotnet/src/Microsoft.AutoGen/RuntimeGateway.Grpc/Abstractions/AgentsRegistryState.cs

@@ -4,12 +4,40 @@
 using Microsoft.AutoGen.Protobuf;
 
 namespace Microsoft.AutoGen.RuntimeGateway.Grpc.Abstractions;
+
+/// <summary>
+/// Stores agent subscription information such as topic and prefix mappings,
+/// and maintains an ETag for concurrency checks.
+/// </summary>
 public class AgentsRegistryState
 {
+    /// <summary>
+    /// Maps each agent ID to the set of topics they subscribe to.
+    /// </summary>
     public ConcurrentDictionary<string, HashSet<string>> AgentsToTopicsMap { get; set; } = [];
+
+    /// <summary>
+    /// Maps each agent ID to the set of topic prefixes they subscribe to.
+    /// </summary>
     public ConcurrentDictionary<string, HashSet<string>> AgentsToTopicsPrefixMap { get; set; } = [];
+
+    /// <summary>
+    /// Maps each topic name to the set of agent types subscribed to it.
+    /// </summary>
     public ConcurrentDictionary<string, HashSet<string>> TopicToAgentTypesMap { get; set; } = [];
+
+    /// <summary>
+    /// Maps each topic prefix to the set of agent types subscribed to it.
+    /// </summary>
     public ConcurrentDictionary<string, HashSet<string>> TopicPrefixToAgentTypesMap { get; set; } = [];
+
+    /// <summary>
+    /// Stores subscriptions by GUID
+    /// </summary>
     public ConcurrentDictionary<string, HashSet<Subscription>> GuidSubscriptionsMap { get; set; } = [];
+
+    /// <summary>
+    /// The concurrency ETag for identifying the registry's version or state.
+    /// </summary>
     public string Etag { get; set; } = Guid.NewGuid().ToString();
 }

dotnet/src/Microsoft.AutoGen/RuntimeGateway.Grpc/Abstractions/IMessageRegistryGrain.cs

@@ -0,0 +1,22 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// IMessageRegistryGrain.cs
+
+using Microsoft.AutoGen.Contracts;
+
+namespace Microsoft.AutoGen.RuntimeGateway.Grpc.Abstractions;
+
+public interface IMessageRegistryGrain : IGrainWithIntegerKey
+{
+    /// <summary>
+    /// Writes a message to the dead-letter queue for the given topic.
+    /// </summary>
+    Task WriteMessageAsync(string topic, CloudEvent message);
+
+    /// <summary>
+    /// Removes all messages for the given topic from the dead-letter queue.
+    /// </summary>
+    /// <param name="topic">The topic to remove messages for.</param>
+    /// <returns>A task representing the asynchronous operation, with the list of removed messages as the result.</returns>
+    Task<List<CloudEvent>> RemoveMessagesAsync(string topic);
+}
+

dotnet/src/Microsoft.AutoGen/RuntimeGateway.Grpc/Abstractions/MessageRegistryState.cs

@@ -0,0 +1,18 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// MessageRegistryState.cs
+
+using System.Collections.Concurrent;
+using Microsoft.AutoGen.Contracts;
+
+namespace Microsoft.AutoGen.RuntimeGateway.Grpc.Abstractions;
+
+/// <summary>
+/// Holds a dead-letter queue by topic type.
+/// </summary>
+public class MessageRegistryState
+{
+    /// <summary>
+    /// Dictionary mapping topic types to a list of CloudEvents that failed delivery.
+    /// </summary>
+    public ConcurrentDictionary<string, List<CloudEvent>> DeadLetterQueue { get; set; } = new();
+}