Python: Allow union types in FanIn edge group (#868)

* Improve type utils

* Add sample

* Add Union

* Add more test cases

* Add more test cases

* Fix RequestResponse typing to only coerce mapping original_request

---------

Co-authored-by: Evan Mattson <evan.mattson@microsoft.com>

python/packages/main/tests/workflow/test_edge.py

@@ -79,6 +79,33 @@ class MockAggregator(Executor):
         self.call_count += 1
         self.last_message = message
 
+    @handler
+    async def mock_aggregator_handler_secondary(
+        self,
+        message: list[MockMessageSecondary],
+        ctx: WorkflowContext,
+    ) -> None:
+        """A mock aggregator handler that does nothing."""
+        self.call_count += 1
+        self.last_message = message
+
+
+class MockAggregatorSecondary(Executor):
+    """A mock aggregator that has a handler for a union type for testing purposes."""
+
+    call_count: int = 0
+    last_message: Any = None
+
+    @handler
+    async def mock_aggregator_handler_combine(
+        self,
+        message: list[MockMessage | MockMessageSecondary],
+        ctx: WorkflowContext,
+    ) -> None:
+        """A mock aggregator handler that does nothing."""
+        self.call_count += 1
+        self.last_message = message
+
 
 # region Edge
 
@@ -1062,6 +1089,73 @@ async def test_fan_in_edge_group_tracing_type_mismatch(span_exporter) -> None:
     assert span.attributes.get("edge_group.delivery_status") == EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH.value
 
 
+async def test_fan_in_edge_group_with_multiple_message_types() -> None:
+    source1 = MockExecutor(id="source_executor_1")
+    source2 = MockExecutor(id="source_executor_2")
+    target = MockAggregatorSecondary(id="target_executor")
+
+    edge_group = FanInEdgeGroup(source_ids=[source1.id, source2.id], target_id=target.id)
+
+    executors: dict[str, Executor] = {source1.id: source1, source2.id: source2, target.id: target}
+    edge_runner = create_edge_runner(edge_group, executors)
+
+    shared_state = SharedState()
+    ctx = InProcRunnerContext()
+
+    data = MockMessage(data="test")
+
+    success = await edge_runner.send_message(
+        Message(data=data, source_id=source1.id),
+        shared_state,
+        ctx,
+    )
+    assert success
+
+    data2 = MockMessageSecondary(data="test")
+    success = await edge_runner.send_message(
+        Message(data=data2, source_id=source2.id),
+        shared_state,
+        ctx,
+    )
+    assert success
+
+
+async def test_fan_in_edge_group_with_multiple_message_types_failed() -> None:
+    source1 = MockExecutor(id="source_executor_1")
+    source2 = MockExecutor(id="source_executor_2")
+    target = MockAggregator(id="target_executor")
+
+    edge_group = FanInEdgeGroup(source_ids=[source1.id, source2.id], target_id=target.id)
+
+    executors: dict[str, Executor] = {source1.id: source1, source2.id: source2, target.id: target}
+    edge_runner = create_edge_runner(edge_group, executors)
+
+    shared_state = SharedState()
+    ctx = InProcRunnerContext()
+
+    data = MockMessage(data="test")
+
+    success = await edge_runner.send_message(
+        Message(data=data, source_id=source1.id),
+        shared_state,
+        ctx,
+    )
+    assert success
+
+    with pytest.raises(RuntimeError):
+        # Although `MockAggregator` can handle `list[MockMessage]` and `list[MockMessageSecondary]`
+        # separately (i.e., it has handlers for each type individually), it cannot handle
+        # `list[MockMessage | MockMessageSecondary]` (a list containing a mix of both types).
+        # With the fan-in edge group, the target executor must handle all message types from the
+        # source executors as a union.
+        data2 = MockMessageSecondary(data="test")
+        _ = await edge_runner.send_message(
+            Message(data=data2, source_id=source2.id),
+            shared_state,
+            ctx,
+        )
+
+
 # endregion FanInEdgeGroup
 
 # region SwitchCaseEdgeGroup