fix: add typing.Annotated support to serialize_type/deserialize_type

haystack/utils/type_serialization.py

@@ -30,6 +30,45 @@
 _SPECIAL_LITERALS: dict[str, Any] = {"None": None, "NoneType": NoneType, "...": ..., "Ellipsis": ...}
 
 
+def _parse_annotated_args(args_str: str) -> list[str]:
+    """
+    Parse Annotated arguments string, handling nested brackets and quoted strings.
+
+    Annotated[int, "doc", 42] -> ["int", '"doc"', "42"]
+    Annotated[int, "a, b", "c"] -> ["int", '"a, b"', '"c"']
+    Annotated[list[int], "meta"] -> ["list[int]", '"meta"']
+    """
+    args = []
+    bracket_count = 0
+    current_arg = ""
+    in_quotes = False
+    quote_char = None
+
+    for char in args_str:
+        if char in ('"', "'") and not in_quotes:
+            in_quotes = True
+            quote_char = char
+        elif char == quote_char and in_quotes:
+            in_quotes = False
+            quote_char = None
+
+        if char == "[" and not in_quotes:
+            bracket_count += 1
+        elif char == "]" and not in_quotes:
+            bracket_count -= 1
+
+        if char == "," and bracket_count == 0 and not in_quotes:
+            args.append(current_arg.strip())
+            current_arg = ""
+        else:
+            current_arg += char
+
+    if current_arg:
+        args.append(current_arg.strip())
+
+    return args
+
+
 def _is_union_type(target: Any) -> bool:
     """
     Check if target is a Union type.
@@ -93,6 +132,16 @@ 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 wraps a type with metadata (e.g. Annotated[int, "doc", 42]).
+    # The first argument is the wrapped type, the rest are metadata values.
+    # Serialize metadata with repr() so strings keep their quotes and other literals
+    # are rendered faithfully.
+    if typing.get_origin(target) is typing.Annotated:
+        args = get_args(target)
+        wrapped_type = serialize_type(args[0])
+        metadata = ", ".join(repr(a) for a in args[1:])
+        return f"typing.Annotated[{wrapped_type}, {metadata}]"
+
     args = get_args(target)
 
     if isinstance(target, UnionType):
@@ -248,6 +297,27 @@ def deserialize_type(type_str: str) -> Any:
         if main_type is typing.Literal:
             return typing.Literal[ast.literal_eval(f"({generics_str},)")]
 
+        # Annotated: first arg is the wrapped type, rest are metadata values.
+        # Use the same safe parsing as Literal, but the first arg is a type (not a literal).
+        if main_type is typing.Annotated:
+            # Parse the full argument list with the quote-aware splitter
+            annotated_args = _parse_annotated_args(generics_str)
+            if not annotated_args:
+                raise DeserializationError("Annotated requires at least one argument (the wrapped type)")
+
+            # First arg is the wrapped type - deserialize it as a type
+            wrapped_type = deserialize_type(annotated_args[0])
+
+            # Remaining args are metadata - parse them safely with ast.literal_eval
+            metadata = []
+            for meta_str in annotated_args[1:]:
+                try:
+                    metadata.append(ast.literal_eval(meta_str))
+                except (ValueError, SyntaxError) as e:
+                    raise DeserializationError(f"Could not parse Annotated metadata: {meta_str}") from e
+
+            return typing.Annotated[wrapped_type, *metadata]
+
         generic_args = [_deserialize_type_arg(arg) for arg in _parse_generic_args(generics_str)]
 
         # Reconstruct

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

@@ -0,0 +1,10 @@
+---
+fixes:
+  - |
+    Fixes typing.Annotated round-trip through serialize_type / deserialize_type.
+    Previously, Annotated types (e.g. Annotated[int, "doc"]) could not be serialized
+    and deserialized correctly - metadata values were rendered without quotes causing
+    deserialization failures or silent corruption (e.g. Annotated[str, "int"] becoming
+    Annotated[str, int]). Now Annotated types with literal metadata (str, int, bool,
+    None, bytes) round-trip correctly, preserving both the wrapped type and metadata values.
+    Mirrors the Literal fix from PR #12286.

test/utils/test_type_serialization.py

@@ -508,3 +508,36 @@ def test_type_de_se_union_and_optional():
         assert serialize_type(Optional[dict]) == "typing.Optional[dict]"
         assert serialize_type(Optional[float]) == "typing.Optional[float]"
         assert serialize_type(Optional[bool]) == "typing.Optional[bool]"
+
+
+def test_output_type_serialization_annotated():
+    from typing import Annotated
+    assert serialize_type(Annotated[int, "doc"]) == "typing.Annotated[int, 'doc']"
+    assert serialize_type(Annotated[str, "int"]) == "typing.Annotated[str, 'int']"
+    assert serialize_type(Annotated[int, "a, b"]) == "typing.Annotated[int, 'a, b']"
+    assert serialize_type(Annotated[int, "doc", 42, True, None, b"bytes"]) == "typing.Annotated[int, 'doc', 42, True, None, b'bytes']"
+    assert serialize_type(Annotated[list[int], "meta"]) == "typing.Annotated[list[int], 'meta']"
+
+
+def test_output_type_deserialization_annotated():
+    from typing import Annotated
+    assert deserialize_type("typing.Annotated[int, 'doc']") == Annotated[int, "doc"]
+    assert deserialize_type("typing.Annotated[str, 'int']") == Annotated[str, "int"]
+    assert deserialize_type("typing.Annotated[int, 'a, b']") == Annotated[int, "a, b"]
+    assert deserialize_type("typing.Annotated[int, 'doc', 42, True, None, b'bytes']") == Annotated[int, "doc", 42, True, None, b"bytes"]
+    assert deserialize_type("typing.Annotated[list[int], 'meta']") == Annotated[list[int], "meta"]
+
+
+def test_output_type_round_trip_annotated():
+    from typing import Annotated
+    for type_ in [
+        Annotated[int, "doc"],
+        Annotated[str, "int"],
+        Annotated[int, "a, b"],
+        Annotated[int, "doc", 42, True, None, b"bytes"],
+        Annotated[list[int], "meta"],
+        Annotated[Dict[str, int], "metadata"],
+        Optional[Annotated[int, "doc"]],
+        Union[Annotated[str, "x"], int],
+    ]:
+        assert deserialize_type(serialize_type(type_)) == type_