Repository: deepset-ai/haystack
Pull request: feat: return common type in SuperComponent type compatibility check
### Related Issues

- fixes #9076 

### Proposed Changes:

- modified _types_are_compatible to return Tuple[bool, Optional[T]]
- updated super_component.py to use the common type returned by _types_are_compatible

### How did you test it?

- added unit tests

### Notes for the reviewer

<!-- E.g. point out section where the reviewer  -->

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:` and added `!` in case the PR includes breaking changes.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
In SuperComponent utils update `_is_compatible(type1, type2)` to return the common type
In SuperComponents during the validation of an `input_mapping` provided by a user we check if the types of the combined inputs are compatible using the `_is_compatible` utility function. `_is_compatible` works by checking if the two types have some overlapping common type rather than using strict type validation.

This is helpful because it can quickly alert a user if a mapping is not possible due to an incompatible type. 

However, after this compatibility check we then assign one of the types (e.g. `type1` or `type2`) as the overall type of the input socket to the SuperComponent. This isn't 100% accurate because we should use the overlapping type of the two types.

So my suggestion would be to expand on `_is_compatible` to also return the detected overlapping type between type1 and type2 which we could use to assign as the overall type for that input socket.

Hints from issue discussion:

Commit message: fix test_utils

Patch to review:
diff --git a/haystack/core/super_component/super_component.py b/haystack/core/super_component/super_component.py
index fd4027eec5..4864b40e1d 100644
--- a/haystack/core/super_component/super_component.py
+++ b/haystack/core/super_component/super_component.py
@@ -210,16 +210,20 @@ def _resolve_input_types_from_mapping(
                         aggregated_inputs[wrapper_input_name]["default"] = _delegate_default
                     continue
 
-                if not _is_compatible(existing_socket_info["type"], socket_info["type"]):
+                is_compatible, common_type = _is_compatible(existing_socket_info["type"], socket_info["type"])
+
+                if not is_compatible:
                     raise InvalidMappingTypeError(
                         f"Type conflict for input '{socket_name}' from component '{comp_name}'. "
                         f"Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}."
                     )
 
+                # Use the common type for the aggregated input
+                aggregated_inputs[wrapper_input_name]["type"] = common_type
+
                 # If any socket requires mandatory inputs then the aggregated input is also considered mandatory.
                 # So we use the type of the mandatory input and remove the default value if it exists.
                 if socket_info["is_mandatory"]:
-                    aggregated_inputs[wrapper_input_name]["type"] = socket_info["type"]
                     aggregated_inputs[wrapper_input_name].pop("default", None)
 
         return aggregated_inputs
diff --git a/haystack/core/super_component/utils.py b/haystack/core/super_component/utils.py
index 2a67ec6804..c8b1a3203c 100644
--- a/haystack/core/super_component/utils.py
+++ b/haystack/core/super_component/utils.py
@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import Annotated, Any, TypeVar, Union, get_args, get_origin
+from typing import Annotated, Any, Optional, Tuple, TypeVar, Union, get_args, get_origin
 
 from haystack.core.component.types import HAYSTACK_GREEDY_VARIADIC_ANNOTATION, HAYSTACK_VARIADIC_ANNOTATION
 
@@ -30,7 +30,7 @@ def _is_compatible(type1: T, type2: T, unwrap_nested: bool = True) -> bool:
     return _types_are_compatible(type1_unwrapped, type2_unwrapped)
 
 
-def _types_are_compatible(type1: T, type2: T) -> bool:
+def _types_are_compatible(type1: T, type2: T) -> Tuple[bool, Optional[T]]:
     """
     Core type compatibility check implementing symmetric matching.
 
@@ -38,9 +38,15 @@ def _types_are_compatible(type1: T, type2: T) -> bool:
     :param type2: Second unwrapped type to compare
     :return: True if types are compatible, False otherwise
     """
-    # Handle Any type and direct equality
-    if type1 is Any or type2 is Any or type1 == type2:
-        return True
+    # Handle Any type
+    if type1 is Any:
+        return True, type2
+    if type2 is Any:
+        return True, type1
+
+    # Direct equality
+    if type1 == type2:
+        return True, type1
 
     type1_origin = get_origin(type1)
     type2_origin = get_origin(type2)
@@ -53,34 +59,72 @@ def _types_are_compatible(type1: T, type2: T) -> bool:
     return _check_non_union_compatibility(type1, type2, type1_origin, type2_origin)
 
 
-def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> bool:
+def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> Tuple[bool, Optional[T]]:
     """Handle all Union type compatibility cases."""
     if type1_origin is Union and type2_origin is not Union:
-        return any(_types_are_compatible(union_arg, type2) for union_arg in get_args(type1))
+        # Find all compatible types from the union
+        compatible_types = []
+        for union_arg in get_args(type1):
+            is_compat, common = _types_are_compatible(union_arg, type2)
+            if is_compat and common is not None:
+                compatible_types.append(common)
+        if compatible_types:
+            return True, Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
+        return False, None
+
     if type2_origin is Union and type1_origin is not Union:
-        return any(_types_are_compatible(type1, union_arg) for union_arg in get_args(type2))
-    # Both are Union types. Check all type combinations are compatible.
-    return any(any(_types_are_compatible(arg1, arg2) for arg2 in get_args(type2)) for arg1 in get_args(type1))
+        # Find all compatible types from the union
+        compatible_types = []
+        for union_arg in get_args(type2):
+            is_compat, common = _types_are_compatible(type1, union_arg)
+            if is_compat and common is not None:
+                compatible_types.append(common)
+        if compatible_types:
+            return True, Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
+        return False, None
+
+    # Both are Union types
+    compatible_types = []
+    for arg1 in get_args(type1):
+        for arg2 in get_args(type2):
+            is_compat, common = _types_are_compatible(arg1, arg2)
+            if is_compat and common is not None:
+                compatible_types.append(common)
+
+    if compatible_types:
+        return True, Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
+    return False, None
 
 
 def _check_non_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> bool:
     """Handle non-Union type compatibility cases."""
     # If no origin, compare types directly
     if not type1_origin and not type2_origin:
-        return type1 == type2
+        if type1 == type2:
+            return True, type1
+        return False, None
 
     # Both must have origins and they must be equal
     if not (type1_origin and type2_origin and type1_origin == type2_origin):
-        return False
+        return False, None
 
     # Compare generic type arguments
     type1_args = get_args(type1)
     type2_args = get_args(type2)
 
     if len(type1_args) != len(type2_args):
-        return False
-
-    return all(_types_are_compatible(t1_arg, t2_arg) for t1_arg, t2_arg in zip(type1_args, type2_args))
+        return False, None
+
+        # Check if all arguments are compatible
+    common_args = []
+    for t1_arg, t2_arg in zip(type1_args, type2_args):
+        is_compat, common = _types_are_compatible(t1_arg, t2_arg)
+        if not is_compat:
+            return False, None
+        common_args.append(common)
+
+    # Reconstruct the type with common arguments
+    return True, type1_origin[tuple(common_args)]
 
 
 def _unwrap_all(t: T, recursive: bool) -> T:
diff --git a/releasenotes/notes/super-component-common-type-30ee9b5522e97ec6.yaml b/releasenotes/notes/super-component-common-type-30ee9b5522e97ec6.yaml
new file mode 100644
index 0000000000..f3ce62de38
--- /dev/null
+++ b/releasenotes/notes/super-component-common-type-30ee9b5522e97ec6.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Enhance SuperComponent's type compatibility check to return the detected common type between two input types.
diff --git a/test/core/super_component/test_super_component.py b/test/core/super_component/test_super_component.py
index 59ecd55695..34c7dcc155 100644
--- a/test/core/super_component/test_super_component.py
+++ b/test/core/super_component/test_super_component.py
@@ -1,7 +1,7 @@
 # SPDX-FileCopyrightText: 2024-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
-from typing import List
+from typing import Any, List, Optional, Union
 
 import pytest
 from haystack import Document, SuperComponent, Pipeline, AsyncPipeline, component, super_component
@@ -314,3 +314,70 @@ def test_custom_super_component_from_dict(self):
         assert isinstance(custom_super_component, CustomSuperComponent)
         assert custom_super_component.var1 == 1
         assert custom_super_component.var2 == "test"
+
+    def test_input_types_reconciliation(self):
+        """Test that input types are properly reconciled when they are compatible but not identical."""
+
+        @component
+        class TypeTestComponent:
+            @component.output_types(result_int=int, result_any=Any)
+            def run(self, input_int: int, input_any: Any):
+                return {"result_int": input_int, "result_any": input_any}
+
+        pipeline = Pipeline()
+        pipeline.add_component("test1", TypeTestComponent())
+        pipeline.add_component("test2", TypeTestComponent())
+
+        input_mapping = {"number": ["test1.input_int", "test2.input_any"]}
+        output_mapping = {"test2.result_int": "result_int"}
+        wrapper = SuperComponent(pipeline=pipeline, input_mapping=input_mapping, output_mapping=output_mapping)
+
+        input_sockets = wrapper.__haystack_input__._sockets_dict
+        assert "number" in input_sockets
+        assert input_sockets["number"].type == int
+
+    def test_union_type_reconciliation(self):
+        """Test that Union types are properly reconciled when creating a SuperComponent."""
+
+        @component
+        class UnionTypeComponent1:
+            @component.output_types(result=Union[int, str])
+            def run(self, input: Union[int, str]):
+                return {"result": input}
+
+        @component
+        class UnionTypeComponent2:
+            @component.output_types(result=Union[float, str])
+            def run(self, input: Union[float, str]):
+                return {"result": input}
+
+        pipeline = Pipeline()
+        pipeline.add_component("test1", UnionTypeComponent1())
+        pipeline.add_component("test2", UnionTypeComponent2())
+
+        input_mapping = {"data": ["test1.input", "test2.input"]}
+        output_mapping = {"test2.result": "result"}
+        wrapper = SuperComponent(pipeline=pipeline, input_mapping=input_mapping, output_mapping=output_mapping)
+
+        input_sockets = wrapper.__haystack_input__._sockets_dict
+        assert "data" in input_sockets
+        assert input_sockets["data"].type == Union[str]
+
+    def test_input_types_with_any(self):
+        """Test that Any type is properly handled when reconciling types."""
+
+        @component
+        class AnyTypeComponent:
+            @component.output_types(result=str)
+            def run(self, specific: str, generic: Any):
+                return {"result": specific}
+
+        pipeline = Pipeline()
+        pipeline.add_component("test", AnyTypeComponent())
+
+        input_mapping = {"text": ["test.specific", "test.generic"]}
+        wrapper = SuperComponent(pipeline=pipeline, input_mapping=input_mapping)
+
+        input_sockets = wrapper.__haystack_input__._sockets_dict
+        assert "text" in input_sockets
+        assert input_sockets["text"].type == str
diff --git a/test/core/super_component/test_utils.py b/test/core/super_component/test_utils.py
index 0a2933f4f6..53a15fa316 100644
--- a/test/core/super_component/test_utils.py
+++ b/test/core/super_component/test_utils.py
@@ -11,10 +11,17 @@ class TestTypeCompatibility:
 
     def test_basic_types(self):
         """Test compatibility of basic Python types."""
-        assert _is_compatible(str, str)
-        assert _is_compatible(int, int)
-        assert not _is_compatible(str, int)
-        assert not _is_compatible(float, int)
+        is_compat, common = _is_compatible(str, str)
+        assert is_compat and common == str
+
+        is_compat, common = _is_compatible(int, int)
+        assert is_compat and common == int
+
+        is_compat, common = _is_compatible(str, int)
+        assert not is_compat and common is None
+
+        is_compat, common = _is_compatible(float, int)
+        assert not is_compat and common is None
 
     def test_any_type(self):
         """Test Any type compatibility."""
@@ -26,12 +33,24 @@ def test_any_type(self):
 
     def test_union_types(self):
         """Test Union type compatibility."""
-        assert _is_compatible(int, Union[int, str])
-        assert _is_compatible(Union[int, str], int)
-        assert _is_compatible(Union[int, str], Union[str, int])
-        assert _is_compatible(str, Union[int, str])
-        assert not _is_compatible(bool, Union[int, str])
-        assert not _is_compatible(float, Union[int, str])
+        is_compat, common = _is_compatible(int, Union[int, str])
+        assert is_compat and common == int
+
+        is_compat, common = _is_compatible(Union[int, str], int)
+        assert is_compat and common == int
+
+        is_compat, common = _is_compatible(Union[int, str], Union[str, int])
+        assert is_compat
+        assert common == Union[int, str] or common == Union[str, int]
+
+        is_compat, common = _is_compatible(str, Union[int, str])
+        assert is_compat and common == str
+
+        is_compat, common = _is_compatible(bool, Union[int, str])
+        assert not is_compat and common is None
+
+        is_compat, common = _is_compatible(float, Union[int, str])
+        assert not is_compat and common is None
 
     def test_variadic_type_compatibility(self):
         """Test compatibility with Variadic and GreedyVariadic types."""
@@ -71,12 +90,21 @@ def test_complex_nested_types(self):
         target_type = List[List[int]]
 
         # With unwrap_nested=True
-        assert _is_compatible(complex_type, target_type)
-        assert _is_compatible(target_type, complex_type)
+        is_compat, common = _is_compatible(complex_type, target_type)
+        assert is_compat
+        # Handle both typing.List and built-in list notation
+        assert common == List[List[int]] or common == list[list[int]]
+
+        is_compat, common = _is_compatible(target_type, complex_type)
+        assert is_compat
+        assert common == List[List[int]] or common == list[list[int]]
 
         # With unwrap_nested=False
-        assert not _is_compatible(complex_type, target_type, unwrap_nested=False)
-        assert not _is_compatible(target_type, complex_type, unwrap_nested=False)
+        is_compat, common = _is_compatible(complex_type, target_type, unwrap_nested=False)
+        assert not is_compat and common is None
+
+        is_compat, common = _is_compatible(target_type, complex_type, unwrap_nested=False)
+        assert not is_compat and common is None
 
     def test_mixed_variadic_types(self):
         """Test mixing Variadic and GreedyVariadic with other type constructs."""


Code review comments:
diff_hunk: @@ -53,34 +59,72 @@ def _types_are_compatible(type1: T, type2: T) -> bool:
     return _check_non_union_compatibility(type1, type2, type1_origin, type2_origin)
 
 
-def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> bool:
+def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> Tuple[bool, Optional[T]]:
     """Handle all Union type compatibility cases."""
     if type1_origin is Union and type2_origin is not Union:
-        return any(_types_are_compatible(union_arg, type2) for union_arg in get_args(type1))
+        # Find all compatible types from the union
+        compatible_types = []
+        for union_arg in get_args(type1):
+            is_compat, common = _types_are_compatible(union_arg, type2)
+            if is_compat and common is not None:
+                compatible_types.append(common)
+        if compatible_types:
+            return True, Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
path: haystack/core/super_component/utils.py
text: @user1:
```suggestion
            # The constructed Union or single type must be cast to Optional[T]
            # to satisfy mypy, as T is specific to this function's call context.
            result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
            return True, cast(Optional[T], result_type)
```

diff_hunk: @@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import Annotated, Any, TypeVar, Union, get_args, get_origin
+from typing import Annotated, Any, Optional, Tuple, TypeVar, Union, get_args, get_origin
path: haystack/core/super_component/utils.py
text: @user1:
```suggestion
from typing import Annotated, Any, Optional, Tuple, TypeVar, Union, cast, get_args, get_origin
```

diff_hunk: @@ -53,34 +59,72 @@ def _types_are_compatible(type1: T, type2: T) -> bool:
     return _check_non_union_compatibility(type1, type2, type1_origin, type2_origin)
 
 
-def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> bool:
+def _check_union_compatibility(type1: T, type2: T, type1_origin: Any, type2_origin: Any) -> Tuple[bool, Optional[T]]:
     """Handle all Union type compatibility cases."""
     if type1_origin is Union and type2_origin is not Union:
-        return any(_types_are_compatible(union_arg, type2) for union_arg in get_args(type1))
+        # Find all compatible types from the union
+        compatible_types = []
+        for union_arg in get_args(type1):
+            is_compat, common = _types_are_compatible(union_arg, type2)
+            if is_compat and common is not None:
+                compatible_types.append(common)
+        if compatible_types:
+            return True, Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
+        return False, None
+
     if type2_origin is Union and type1_origin is not Union:
-        return any(_types_are_compatible(type1, union_arg) for union_arg in get_args(type2))
-    # Both are Union types. Check all type combinations are compatible.
-    return any(any(_types_are_compatible(arg1, arg2) for arg2 in get_args(type2)) for arg1 in get_args(type1))
+        # Find all compatible types from the union
+        compatible_types = []
+        for union_arg in get_args(type2):
+            is_compat, common = _types_are_compatible(type1, union_arg)
+            if is_compat and common is not None:
+                compatible_types.append(common)
+        if compatible_types:
+            return True, Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
path: haystack/core/super_component/utils.py
text: @user1:
```suggestion
            # The constructed Union or single type must be cast to Optional[T]
            # to satisfy mypy, as T is specific to this function's call context.
            result_type = Union[tuple(compatible_types)] if len(compatible_types) > 1 else compatible_types[0]
            return True, cast(Optional[T], result_type)
```