fix: round-trip serialization of Callable types with an explicit parameter list (#12122)

Co-authored-by: David S. Batista <dsbatista@gmail.com>

Author: vidigoat

haystack/utils/type_serialization.py

@@ -50,6 +50,21 @@ def _build_pep604_union_type(types: list[type | UnionType]) -> type | UnionType:
     return result
 
 
+def _serialize_type_arg(arg: Any) -> str:
+    """
+    Serialize a single generic argument.
+
+    Almost all arguments are plain types and are handed off to ``serialize_type``. The exception is the
+    parameter list of a ``Callable[[X, Y], R]``, which ``typing.get_args`` returns as a Python ``list``
+    (e.g. ``[X, Y]``) rather than a type. Such a list is rendered as ``[X, Y]`` so it round-trips back to the
+    same ``Callable`` on deserialization. Without this, the list would fall through to ``serialize_type`` and
+    be dropped (its ``str()`` starts with ``[``, which the name handling truncates to an empty string).
+    """
+    if isinstance(arg, list):
+        return f"[{', '.join(serialize_type(a) for a in arg)}]"
+    return serialize_type(arg)
+
+
 def serialize_type(target: Any) -> str:
     """
     Serializes a type or an instance to its string representation, including the module name.
@@ -100,7 +115,7 @@ def serialize_type(target: Any) -> str:
         # Union with more than two members) NoneType is a regular argument and must be kept.
         skip_nonetype = name == "Optional"
         args_str = ", ".join(
-            serialize_type(Union[tuple(get_args(a))] if is_typing_generic and isinstance(a, UnionType) else a)  # noqa: UP007
+            _serialize_type_arg(Union[tuple(get_args(a))] if is_typing_generic and isinstance(a, UnionType) else a)  # noqa: UP007
             for a in args
             if not (skip_nonetype and a is NoneType)
         )
@@ -163,6 +178,24 @@ def _parse_pep604_union_args(union_str: str) -> list[str]:
     return args
 
 
+def _deserialize_type_arg(arg_str: str) -> Any:
+    """
+    Deserialize a single generic argument produced by ``_parse_generic_args``.
+
+    Mirrors ``_serialize_type_arg``: a ``Callable`` parameter list is emitted as ``[X, Y]`` and must be read
+    back as a Python ``list`` of types (so ``Callable`` can be re-subscripted as ``Callable[[X, Y], R]``),
+    not as a single type. An empty parameter list ``[]`` becomes ``[]``. Every other argument is a normal
+    type and is delegated to ``deserialize_type``.
+    """
+    stripped = arg_str.strip()
+    if stripped.startswith("[") and stripped.endswith("]"):
+        inner = stripped[1:-1].strip()
+        if not inner:
+            return []
+        return [deserialize_type(a) for a in _parse_generic_args(inner)]
+    return deserialize_type(arg_str)
+
+
 def deserialize_type(type_str: str) -> Any:
     """
     Deserializes a type given its full import path as a string, including nested generic types.
@@ -202,7 +235,7 @@ def deserialize_type(type_str: str) -> Any:
         generics_str = generics_str[:-1]
 
         main_type = deserialize_type(main_type_str)
-        generic_args = [deserialize_type(arg) for arg in _parse_generic_args(generics_str)]
+        generic_args = [_deserialize_type_arg(arg) for arg in _parse_generic_args(generics_str)]
 
         # Reconstruct
         try:

releasenotes/notes/fix-callable-parameter-list-type-serialization-0c6f96661cf773db.yaml

@@ -0,0 +1,10 @@
+---
+fixes:
+  - |
+    Fixed ``serialize_type`` and ``deserialize_type`` to correctly round-trip
+    ``Callable`` types that declare an explicit parameter list, such as
+    ``Callable[[int, str], bool]``. Previously the parameter list was dropped
+    during serialization (producing a malformed string like
+    ``typing.Callable[, bool]``) and could no longer be deserialized. This
+    affected components that serialize type annotations, for example
+    ``ConditionalRouter`` and ``OutputAdapter`` using a ``Callable`` output type.

test/components/converters/test_output_adapter.py

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import json
-from typing import Any, List
+from typing import Any, Callable, List
 
 import pytest
 
@@ -99,6 +99,18 @@ def test_sede(self):
         assert adapter.template == deserialized_adapter.template
         assert adapter.output_type == deserialized_adapter.output_type
 
+    def test_sede_with_callable_output_type(self):
+        # Regression test: `output_type=Callable[[int, str], bool]` used to lose its parameter list on
+        # `to_dict` (producing "typing.Callable[, bool]") and then fail to deserialize entirely.
+        adapter = OutputAdapter(template="{{ callback }}", output_type=Callable[[int, str], bool])
+        adapter_dict = adapter.to_dict()
+
+        assert adapter_dict["init_parameters"]["output_type"] == "typing.Callable[[int, str], bool]"
+
+        deserialized_adapter = OutputAdapter.from_dict(adapter_dict)
+        assert adapter.template == deserialized_adapter.template
+        assert adapter.output_type == deserialized_adapter.output_type
+
     # OutputAdapter can be serialized to a dictionary and deserialized along with custom filters
     def test_sede_with_custom_filters(self):
         # NOTE: filters need to be declared in a namespace visible to the deserialization function

test/components/routers/test_conditional_router.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import copy
+from collections.abc import Callable
 from unittest import mock
 
 import pytest
@@ -279,6 +280,26 @@ def test_router_de_serialization(self):
         # check that the result is the same and correct
         assert result1 == result2 and result1 == {"streams": [1, 2, 3]}
 
+    def test_router_de_serialization_with_callable_output_type(self):
+        # Regression test: `output_type=Callable[[int, str], bool]` used to lose its parameter list on
+        # `to_dict` (producing "typing.Callable[, bool]") and then fail to deserialize entirely.
+        routes = [
+            {
+                "condition": "{{True}}",
+                "output": "{{callback}}",
+                "output_type": Callable[[int, str], bool],
+                "output_name": "callback",
+            }
+        ]
+        router = ConditionalRouter(routes)
+        router_dict = router.to_dict()
+
+        serialized_output_type = router_dict["init_parameters"]["routes"][0]["output_type"]
+        assert serialized_output_type == "collections.abc.Callable[[int, str], bool]"
+
+        new_router = ConditionalRouter.from_dict(router_dict)
+        assert router.routes == new_router.routes
+
     def test_router_de_serialization_with_none_argument(self):
         new_router = ConditionalRouter.from_dict(
             {

test/utils/test_type_serialization.py

@@ -259,6 +259,42 @@ def test_output_type_deserialization_legacy_ellipsis_literal():
     assert deserialize_type("typing.Callable[Ellipsis, int]") == Callable[..., int]
 
 
+def test_output_type_serialization_callable_with_parameter_list():
+    # `Callable[[X, Y], R]` returns its parameter list from `typing.get_args` as a Python list ([X, Y]),
+    # not as a type. It must be serialized as "[X, Y]" so the parameters survive the round-trip.
+    assert serialize_type(Callable[[int, str], bool]) == "typing.Callable[[int, str], bool]"
+    assert serialize_type(Callable[[], int]) == "typing.Callable[[], int]"
+    assert serialize_type(Callable[[int], List[str]]) == "typing.Callable[[int], typing.List[str]]"
+    assert serialize_type(Callable[[Union[str, int]], bool]) == "typing.Callable[[typing.Union[str, int]], bool]"
+
+
+def test_output_type_deserialization_callable_with_parameter_list():
+    assert deserialize_type("typing.Callable[[int, str], bool]") == Callable[[int, str], bool]
+    assert deserialize_type("typing.Callable[[], int]") == Callable[[], int]
+    assert deserialize_type("typing.Callable[[int], typing.List[str]]") == Callable[[int], List[str]]
+    assert deserialize_type("typing.Callable[[typing.Union[str, int]], bool]") == Callable[[Union[str, int]], bool]
+
+
+def test_output_type_round_trip_callable_with_parameter_list():
+    for type_ in [
+        Callable[[int, str], bool],
+        Callable[[], int],
+        Callable[[int], List[str]],
+        Callable[[Dict[str, int], str], List[bool]],
+        Callable[[Union[str, int]], bool],
+        List[Callable[[int], str]],
+        Dict[str, Callable[[int, str], bool]],
+        Optional[Callable[[int], str]],
+        Callable[[Callable[[int], str]], bool],
+        # The Ellipsis form (no parameter list) must still round-trip unaffected by the parameter-list
+        # handling above: `_serialize_type_arg`/`_deserialize_type_arg` only special-case a `list` argument,
+        # so `...` still falls through to the existing Ellipsis handling in serialize_type/deserialize_type.
+        Callable[..., int],
+        Callable[[int], Callable[..., str]],
+    ]:
+        assert deserialize_type(serialize_type(type_)) == type_
+
+
 def test_output_type_serialization_haystack_dataclasses():
     # typing
     # Answer