Refine message handling in MessageReceived activity.

Ensure type validation for incoming messages and clean up workflow input to prevent unintended data propagation. Added a dedicated ResumeAsync method to handle bookmark resumption logic consistently.

Fixes #6294

src/modules/Elsa.MassTransit/Activities/MessageReceived.cs

@@ -32,20 +32,32 @@ public class MessageReceived : Trigger<object>
     protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
         // If we did not receive external input, it means we are just now encountering this activity and we need to block execution by creating a bookmark.
-        if (!context.TryGetWorkflowInput<object>(InputKey, out var message))
+        if (!context.TryGetWorkflowInput<object>(InputKey, out var message) || message.GetType() != MessageType)
         {
             // Create bookmarks for when we receive the expected HTTP request.
-            context.CreateBookmark(GetBookmarkPayload(context.ExpressionExecutionContext));
+            context.CreateBookmark(GetBookmarkPayload(context.ExpressionExecutionContext), ResumeAsync, includeActivityInstanceId: false);
             return;
         }
 
         // Provide the received message as output.
         context.Set(Result, message);
         
+        // Remove the input to prevent it from being passed to the next activity.
+        context.WorkflowInput.Remove(InputKey);
+        
         // Complete.
         await context.CompleteActivityAsync();
     }
 
+    private ValueTask ResumeAsync(ActivityExecutionContext context)
+    {
+        // Remove the input to prevent it from being passed to the next activity.
+        context.WorkflowInput.Remove(InputKey);
+        
+        // Complete.
+        return context.CompleteActivityAsync();
+    }
+
     private object GetBookmarkPayload(ExpressionExecutionContext context)
     {
         // Generate bookmark data for message type.