fix: round-trip serialization of typing.Annotated

haystack/core/super_component/utils.py

@@ -13,9 +13,7 @@ class _delegate_default:
     """Custom object for delegating filling of default values to the underlying components."""
 
 
-def _is_compatible(
-    type1: type | UnionType, type2: type | UnionType, unwrap_nested: bool = True
-) -> tuple[bool, type | UnionType | None]:
+def _is_compatible(type1: Any, type2: Any, unwrap_nested: bool = True) -> tuple[bool, Any]:
     """
     Check if two types are compatible (bidirectional/symmetric check).
 
@@ -147,6 +145,13 @@ def _unwrap_all(t: type | UnionType, recursive: bool) -> type | UnionType:
     if _is_variadic_type(t):
         t = _unwrap_variadics(t, recursive=recursive)
     else:
+        # `Annotated[T, m1, m2, ...]` is the same type as `T` for compatibility — the metadata is
+        # just an annotation, not a type modifier. Unwrap to T (recursively, for nested Annotated).
+        # Haystack Variadic markers are also expressed as Annotated[T, HAYSTACK_VARIADIC_ANNOTATION]
+        # — the `_is_variadic_type` branch above handles those, so this unwrap only fires for plain
+        # user-supplied Annotated.
+        if get_origin(t) is Annotated:
+            return _unwrap_all(get_args(t)[0], recursive)
         # If it's a generic type and we're unwrapping recursively
         origin = get_origin(t)
         if recursive and origin is not None and (args := get_args(t)):

haystack/core/type_utils.py

@@ -7,7 +7,7 @@
 from collections.abc import Callable
 from enum import Enum
 from types import NoneType, UnionType
-from typing import Any, Union, get_args, get_origin, get_type_hints
+from typing import Annotated, Any, Union, get_args, get_origin, get_type_hints
 
 from haystack.dataclasses import ChatMessage
 
@@ -108,7 +108,7 @@ def _type_name(type_: Any) -> str:
     return f"{name}"
 
 
-def _safe_get_origin(_type: type | UnionType) -> Any:
+def _safe_get_origin(_type: Any) -> Any:
     """
     Safely retrieves the origin type of a generic alias or returns the type itself if it's a built-in.
 
@@ -124,9 +124,31 @@ def _safe_get_origin(_type: type | UnionType) -> Any:
     # So we convert UnionType to Union if it is detected.
     if origin is UnionType:
         origin = Union
+    # `Annotated[T, m1, m2, ...]` is the same type as `T` for type-checking purposes — the metadata
+    # is just an annotation, not a type modifier. `get_origin(Annotated[T, ...])` returns `Annotated`
+    # (not `T`), so unwrap to the wrapped type (recursively, for nested Annotated). Without this,
+    # `Annotated[int, "doc"]` would not be considered compatible with `int` and a component socket
+    # declared as `Annotated[int, "doc"]` could not be connected to one declared as `int`.
+    while origin is Annotated:
+        _type = get_args(_type)[0]
+        origin = get_origin(_type) or (_type if isinstance(_type, type) else None)
     return origin
 
 
+def _unwrap_annotated(_type: Any) -> Any:
+    """
+    Strip one or more ``Annotated[...]`` wrappers and return the underlying type.
+
+    ``Annotated[T, m1, m2, ...]`` is the same type as ``T`` for type-checking purposes — the metadata
+    is annotation, not a type modifier. This is a small helper that does the same unwrap as
+    ``_safe_get_origin`` but returns the full type object (not just the origin), so callers that need
+    to use ``get_args`` on the result can do so without seeing the metadata as a type arg.
+    """
+    while get_origin(_type) is Annotated:
+        _type = get_args(_type)[0]
+    return _type
+
+
 def _contains_type(container: Any, target: Any) -> bool:
     """Checks if the container type includes the target type"""
     if container == target:
@@ -145,6 +167,15 @@ def _strict_types_are_compatible(sender: Any, receiver: Any) -> bool:  # noqa: P
     :param receiver: The receiver type.
     :return: True if the sender type is strictly compatible with the receiver type, False otherwise.
     """
+    # `Annotated[T, m1, m2, ...]` is the same type as `T` for type-checking purposes — the metadata is
+    # just an annotation, not a type modifier. Unwrap both sides so the rest of the function works on
+    # the underlying types. Without this, `get_args(Annotated[int, "x"])` returns `(int, "x")` and the
+    # generic-args comparison below would see a 2-arg sender against a 0-arg receiver (int) and reject
+    # the pair. _safe_get_origin does the same unwrap; we also need it here because the rest of this
+    # function uses `get_args` directly.
+    sender = _unwrap_annotated(sender)
+    receiver = _unwrap_annotated(receiver)
+
     if sender == receiver or receiver is Any:
         return True
 

haystack/utils/type_serialization.py

@@ -69,6 +69,23 @@ def _serialize_type_arg(arg: Any) -> str:
     return serialize_type(arg)
 
 
+def _serialize_annotated_meta(meta: Any) -> str:
+    """
+    Serialize a single metadata value of a ``typing.Annotated[...]``.
+
+    Literal values (str, int, bool, None, bytes) are rendered with ``repr()`` so they keep their
+    quotes and round-trip cleanly through ``ast.literal_eval`` on the deserialize side. Type-like
+    values (classes, typing forms) are rendered through ``serialize_type`` so the full module path
+    is preserved and the deserialize side can resolve them back to the same object. ``repr`` on a
+    class produces ``<class 'foo.Bar'>`` which is not a valid Python literal and would not round-trip.
+    """
+    try:
+        ast.literal_eval(repr(meta))
+    except (ValueError, SyntaxError):
+        return serialize_type(meta)
+    return repr(meta)
+
+
 def serialize_type(target: Any) -> str:
     """
     Serializes a type or an instance to its string representation, including the module name.
@@ -96,6 +113,18 @@ def serialize_type(target: Any) -> str:
     if typing.get_origin(target) is typing.Literal:
         return f"typing.Literal[{', '.join(repr(a) for a in get_args(target))}]"
 
+    # Annotated[T, m1, m2, ...] holds a type T followed by metadata values. Python normalizes the bare
+    # `Annotated[T]` form back to `T` (no metadata), so the metadata list is always non-empty here.
+    # Render the type through serialize_type (so nested generics and module paths are preserved) and
+    # each metadata value with repr() so strings keep their quotes — same approach as the Literal
+    # branch above, since metadata values are not types and would otherwise be misread as type names on
+    # deserialize (e.g. Annotated[str, "int"] silently corrupting to Annotated[str, int]).
+    if typing.get_origin(target) is typing.Annotated:
+        ann_args = get_args(target)
+        type_str = serialize_type(ann_args[0])
+        meta_str = ", ".join(_serialize_annotated_meta(m) for m in ann_args[1:])
+        return f"typing.Annotated[{type_str}, {meta_str}]"
+
     args = get_args(target)
 
     if isinstance(target, UnionType):
@@ -206,7 +235,66 @@ def _deserialize_type_arg(arg_str: str) -> Any:
 
 
 @mark_deserialization_internal
-def deserialize_type(type_str: str) -> Any:
+def _split_annotated_args(args_str: str) -> tuple[str, str]:
+    """
+    Split the inside of a serialized ``typing.Annotated[...]`` into ``(type_str, metadata_str)``.
+
+    Splits on the first comma at the top level — outside any ``[...]`` brackets and outside any string
+    literal — so a comma inside a string metadata value (``Annotated[int, "a, b"]``) is not treated as a
+    separator. ``_parse_generic_args`` is bracket-aware but not quote-aware and would mis-split such a
+    payload. If no top-level comma is found the whole string is the type and the metadata is empty
+    (defensive: ``Annotated`` always has at least one metadata value in practice because Python
+    normalizes ``Annotated[T]`` back to ``T``).
+
+    :param args_str: The contents between the brackets of a serialized ``typing.Annotated[...]``.
+    :returns: A ``(type_str, metadata_str)`` tuple. Both are stripped. ``metadata_str`` is empty when no
+        top-level comma is present.
+    """
+    bracket_count = 0
+    in_string = False
+    string_char: str | None = None
+    i = 0
+    while i < len(args_str):
+        c = args_str[i]
+        if in_string:
+            if c == "\\" and i + 1 < len(args_str):
+                # Skip the escaped character so a backslash-escaped quote does not end the string.
+                i += 2
+                continue
+            if c == string_char:
+                in_string = False
+        elif c in ('"', "'"):
+            in_string = True
+            string_char = c
+        elif c == "[":
+            bracket_count += 1
+        elif c == "]":
+            bracket_count -= 1
+        elif c == "," and bracket_count == 0:
+            return args_str[:i].strip(), args_str[i + 1 :].strip()
+        i += 1
+    return args_str.strip(), ""
+
+
+@mark_deserialization_internal
+def _deserialize_annotated_metadata(arg_str: str) -> Any:
+    """
+    Deserialize a single metadata value from a serialized ``typing.Annotated[...]``.
+
+    Tries ``ast.literal_eval`` first (which safely parses Python literals — str, int, bool, None, bytes —
+    and is quote-aware). Falls back to ``deserialize_type`` for non-literal metadata such as a validator
+    class (``Annotated[int, MyValidator]``), which is rendered with its full module path by
+    ``serialize_type``. Ellipsis (``...``) is parsed through the fallback to ``_SPECIAL_LITERALS``.
+    """
+    stripped = arg_str.strip()
+    try:
+        return ast.literal_eval(stripped)
+    except (ValueError, SyntaxError):
+        return deserialize_type(stripped)
+
+
+@mark_deserialization_internal
+def deserialize_type(type_str: str) -> Any:  # noqa: PLR0911 - dispatch on wire-form (PEP 604 / generic / Annotated / module path / builtin / typing)
     """
     Deserializes a type given its full import path as a string, including nested generic types.
 
@@ -252,6 +340,30 @@ def deserialize_type(type_str: str) -> Any:
         if main_type is typing.Literal:
             return typing.Literal[ast.literal_eval(f"({generics_str},)")]
 
+        # Annotated[T, m1, m2, ...] has a type as the first argument and metadata values (typically
+        # Python literals) as the rest. The split is quote-aware because a string metadata value can
+        # contain a comma — `_parse_generic_args` is bracket-aware but not quote-aware and would
+        # mis-split such a payload. The first chunk is the type and is resolved via deserialize_type;
+        # each metadata chunk is resolved via `_deserialize_annotated_metadata` (literal_eval with a
+        # deserialize_type fallback for type metadata like validators).
+        if main_type is typing.Annotated:
+            type_str, meta_str = _split_annotated_args(generics_str)
+            deserialized_type = deserialize_type(type_str)
+            if not meta_str:
+                raise DeserializationError(
+                    f"Annotated requires at least one metadata value: typing.Annotated[{generics_str}]"
+                )
+            # Try the all-literals path first (the common case): wrap the metadata in `(...)` with a
+            # trailing comma so `ast.literal_eval` parses it as a tuple. This is quote-aware, so a
+            # comma inside a string metadata value is not treated as a separator. If any metadata is
+            # a type (e.g. `Annotated[int, MyValidator]`), `ast.literal_eval` raises and we fall back
+            # to the per-arg path that mixes literal_eval with deserialize_type.
+            try:
+                deserialized_metadata = list(ast.literal_eval(f"({meta_str},)"))
+            except (ValueError, SyntaxError):
+                deserialized_metadata = [_deserialize_annotated_metadata(a) for a in _parse_generic_args(meta_str)]
+            return typing.Annotated[(deserialized_type, *deserialized_metadata)]
+
         generic_args = [_deserialize_type_arg(arg) for arg in _parse_generic_args(generics_str)]
 
         # Reconstruct

releasenotes/notes/fix-annotated-type-serialization-7d2c4f1e3a8b9450.yaml

@@ -0,0 +1,24 @@
+---
+fixes:
+  - |
+    Fixed round-trip serialization and Pipeline type-checking for ``typing.Annotated``. Two related gaps:
+
+    * ``serialize_type``/``deserialize_type`` had no handling for ``Annotated``: metadata values were
+      rendered as bare tokens (e.g. ``typing.Annotated[int, doc]``) which failed to deserialize, and
+      values that happened to look like a type name (e.g. ``Annotated[str, "int"]``) were silently
+      turned into types on the round-trip. Literal metadata (str, int, bool, None, bytes) is now
+      serialized with ``repr()`` and read back with ``ast.literal_eval``; type metadata is serialized
+      through ``serialize_type`` and resolved through ``deserialize_type`` on read. A quote-aware
+      split is used on the deserialize side so a comma inside a string metadata value (e.g.
+      ``Annotated[int, "a, b"]``) is not treated as an argument separator.
+    * ``_safe_get_origin`` and ``_strict_types_are_compatible`` in ``haystack.core.type_utils`` did
+      not unwrap ``Annotated``, so a component socket declared as ``Annotated[int, "doc"]`` was
+      treated as a different type from ``int`` and could not be connected. ``Annotated`` is now
+      treated as the underlying type for type-checking purposes (the metadata is annotation, not a
+      type modifier), so a socket declared as ``Annotated[int, "doc"]`` can be connected to one
+      declared as ``int``.
+
+    An ``Annotated`` type used by a component (such as an ``OutputAdapter`` or any tool whose schema
+    uses Pydantic ``Field`` metadata) now round-trips correctly through ``Pipeline.dumps()`` /
+    ``Pipeline.loads()`` and can be connected across components. Mirrors the ``Literal`` fix from
+    ``#12286``.

test/core/super_component/test_utils.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import Any, Optional, Union
+from typing import Annotated, Any, Optional, Union
 
 import pytest
 
@@ -242,10 +242,10 @@ def test_mixed_variadic_types():
     # Variadic with Union
     var_union = Variadic[Union[int, str]]
 
-    is_compat, common = _is_compatible(var_union, Union[int, str])  # type: ignore[arg-type]
+    is_compat, common = _is_compatible(var_union, Union[int, str])
     assert is_compat and common == Union[int, str]
 
-    is_compat, common = _is_compatible(Union[int, str], var_union)  # type: ignore[arg-type]
+    is_compat, common = _is_compatible(Union[int, str], var_union)
     assert is_compat and common == Union[int, str]
 
     # GreedyVariadic with Optional
@@ -262,3 +262,39 @@ def test_mixed_variadic_types():
 
     is_compat, common = _is_compatible(nested_var, list[int])
     assert is_compat and common == list[int]
+
+
+# `Annotated[T, m1, m2, ...]` is the same type as `T` for compatibility — the metadata is just an
+# annotation, not a type modifier. These tests verify that the super_component type checker treats
+# Annotated the same as the wrapped type. Without this, a SuperComponent socket declared as
+# `Annotated[int, "doc"]` could not be aligned with one declared as `int`.
+def test_annotated_type_compatibility():
+    # Annotated[int, ...] is compatible with int in both directions.
+    is_compat, common = _is_compatible(Annotated[int, "doc"], int)
+    assert is_compat and common is int
+
+    is_compat, common = _is_compatible(int, Annotated[int, "doc"])
+    assert is_compat and common is int
+
+    # Two Annotated types with the same wrapped type are compatible (metadata is ignored).
+    is_compat, common = _is_compatible(Annotated[int, "x"], Annotated[int, "y"])
+    assert is_compat and common is int
+
+    # Nested Annotated: unwrap recursively.
+    is_compat, common = _is_compatible(Annotated[Annotated[int, "inner"], "outer"], int)
+    assert is_compat and common is int
+
+    # Annotated with a generic wrapped type is compatible with the bare generic.
+    is_compat, common = _is_compatible(Annotated[list[int], "doc"], list[int])
+    assert is_compat and common == list[int]
+
+    is_compat, common = _is_compatible(list[int], Annotated[list[int], "doc"])
+    assert is_compat and common == list[int]
+
+    # Annotated[Optional[int], ...] is compatible with Optional[int].
+    is_compat, common = _is_compatible(Annotated[Optional[int], "doc"], Optional[int])
+    assert is_compat and common == Optional[int]
+
+    # Annotated[str, ...] is not compatible with int.
+    is_compat, _ = _is_compatible(Annotated[str, "doc"], int)
+    assert not is_compat

test/core/test_type_utils.py

@@ -1233,3 +1233,45 @@ def test_keeps_annotations_of_target_that_cannot_carry_them(self):
             "documents": "list[Document]",
             "top_k": "int | None",
         }
+
+
+# `Annotated[T, m1, m2, ...]` is the same type as `T` for type-checking purposes — the metadata is
+# just an annotation, not a type modifier. These tests verify that the Pipeline type checker treats
+# Annotated the same as the wrapped type so a component socket declared as `Annotated[int, "doc"]`
+# can be connected to a socket declared as `int` (this is what enables `Annotated` annotations in
+# component input/output socket types to round-trip through Pipeline serialization).
+def test_annotated_same_as_wrapped_type_strict():
+    from typing import Annotated
+
+    from haystack.core.type_utils import _strict_types_are_compatible
+
+    # Annotated[int, ...] is compatible with int in both directions.
+    assert _strict_types_are_compatible(Annotated[int, "doc"], int)
+    assert _strict_types_are_compatible(int, Annotated[int, "doc"])
+    # Two Annotated types with the same wrapped type are compatible (metadata is ignored).
+    assert _strict_types_are_compatible(Annotated[int, "x"], Annotated[int, "y"])
+    # Nested Annotated: unwrap recursively.
+    assert _strict_types_are_compatible(Annotated[Annotated[int, "inner"], "outer"], int)
+    # Annotated with a generic wrapped type is compatible with the bare generic.
+    assert _strict_types_are_compatible(Annotated[List[int], "doc"], List[int])
+    assert _strict_types_are_compatible(List[int], Annotated[List[int], "doc"])
+    # Annotated[str, ...] is not compatible with int.
+    assert not _strict_types_are_compatible(Annotated[str, "doc"], int)
+    assert not _strict_types_are_compatible(int, Annotated[str, "doc"])
+
+
+def test_safe_get_origin_unwraps_annotated():
+    from typing import Annotated
+
+    from haystack.core.type_utils import _safe_get_origin
+
+    # _safe_get_origin(Annotated[T, ...]) returns the origin of T (not Annotated), so the rest of
+    # the type checker can treat Annotated[T, ...] as T. Without this unwrap, the type checker
+    # would see `Annotated` as the origin and reject the type as incompatible with anything else.
+    assert _safe_get_origin(Annotated[int, "doc"]) is int
+    assert _safe_get_origin(Annotated[str, "x", "y"]) is str
+    # Nested Annotated: unwrap recursively.
+    assert _safe_get_origin(Annotated[Annotated[int, "inner"], "outer"]) is int
+    # Bare types are unchanged.
+    assert _safe_get_origin(int) is int
+    assert _safe_get_origin(List[int]) is list