fix(serialization): support set and frozenset generic containers in JsonDataConverter

CHANGELOG.md

@@ -14,6 +14,7 @@ by one or more task hubs.
 
 FIXED
 
+- Added deterministic serialization for `set`, `frozenset`, and `collections.abc.Set` containers, and enabled opt-in generic set reconstruction (`enable_generic_sets`) in `JsonDataConverter` to preserve existing-instance replay compatibility.
 - Large-payload externalization now checks every eligible payload's maximum
 size before uploading or replacing any protobuf values, including carryover
 events and entity payloads, for both synchronous and asynchronous calls.

durabletask/serialization.py

@@ -42,7 +42,8 @@
 import types
 import typing
 from abc import ABC, abstractmethod
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Set as AbstractSet
 from types import SimpleNamespace
 from typing import Any, cast
 
@@ -68,7 +69,6 @@ class DataConverter(ABC):
     @abstractmethod
     def serialize(self, value: Any) -> str | None:
         """Serialize ``value`` to a string, or ``None`` when ``value`` is ``None``."""
-        pass
 
     @abstractmethod
     def deserialize(self, data: str | None, target_type: type | None = None) -> Any:
@@ -85,7 +85,6 @@ def deserialize(self, data: str | None, target_type: type | None = None) -> Any:
         :class:`JsonDataConverter` is best-effort and falls back; a validating
         converter may instead raise.
         """
-        pass
 
     @abstractmethod
     def coerce(self, value: Any, target_type: type | None = None) -> Any:
@@ -99,7 +98,6 @@ def coerce(self, value: Any, target_type: type | None = None) -> Any:
         (strict vs. best-effort) that an implementation applies in
         :meth:`deserialize` should apply here.
         """
-        pass
 
     def can_reconstruct(self, target_type: Any) -> bool:
         """Return True if this converter can rebuild ``target_type`` from a payload.
@@ -167,6 +165,18 @@ class JsonDataConverter(DataConverter):
     > payload; the destination type is always supplied by the caller.
     """
 
+    def __init__(
+        self,
+        *,
+        enable_generic_sets: bool = False,
+        enable_set_coercion: bool = False,
+    ) -> None:
+        self.enable_generic_sets: bool = enable_generic_sets or enable_set_coercion
+
+    @property
+    def enable_set_coercion(self) -> bool:
+        return self.enable_generic_sets
+
     def serialize(self, value: Any) -> str | None:
         if value is None:
             return None
@@ -221,15 +231,40 @@ def _log_coercion_fallback(target_type: type, error: Exception) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _is_supported_set_element(converter: DataConverter, elem_type: Any) -> bool:
+    if elem_type is None or elem_type is typing.Any:
+        return True
+    elem_origin = typing.get_origin(elem_type)
+    if elem_origin is not None:
+        if elem_origin in (list, dict, Sequence, Mapping, set, AbstractSet):
+            # Mutable collections are unhashable
+            return False
+        if elem_origin is typing.Union or elem_origin is types.UnionType:
+            non_none = [a for a in typing.get_args(elem_type) if a is not type(None)]
+            return bool(non_none) and all(_is_supported_set_element(converter, a) for a in non_none)
+        if elem_origin is frozenset:
+            args = typing.get_args(elem_type)
+            return not args or _is_supported_set_element(converter, args[0])
+        return False
+    if not isinstance(elem_type, type):
+        return False
+    if issubclass(elem_type, (list, dict, set)):
+        return False
+    return True
+
+
 def _can_reconstruct(converter: DataConverter, target_type: Any) -> bool:
     """:class:`JsonDataConverter`'s reconstruction policy.
 
     Recognizes dataclasses and ``from_json()``-capable types, plus ``Optional``
     / ``list`` / ``Sequence`` hints wrapping them; builtins and unknown
-    annotations are excluded. Recurses through ``converter.can_reconstruct``
-    (not itself) so a :class:`JsonDataConverter` subclass that overrides
-    ``can_reconstruct`` still participates in the element-type checks of
-    ``Optional`` / ``list`` hints.
+    annotations are excluded. When set support is enabled on the converter
+    (via ``enable_generic_sets=True``), ``set`` / ``frozenset`` / ``Set``
+    containers with supported element types are also recognized. Recurses
+    through ``converter.can_reconstruct`` (not itself) so a
+    :class:`JsonDataConverter` subclass that overrides ``can_reconstruct``
+    still participates in the element-type checks of ``Optional`` / ``list``
+    hints.
     """
     origin = typing.get_origin(target_type)
     if origin is not None:
@@ -240,11 +275,19 @@ def _can_reconstruct(converter: DataConverter, target_type: Any) -> bool:
             )
         if origin in (list, Sequence):
             return any(converter.can_reconstruct(a) for a in args)
+        if origin in (set, frozenset, AbstractSet, typing.AbstractSet):
+            if not getattr(converter, "enable_generic_sets", False):
+                return False
+            if not args:
+                return True
+            return _is_supported_set_element(converter, args[0])
         return False
     if not isinstance(target_type, type):
         return False
     if dataclasses.is_dataclass(target_type):
         return True
+    if target_type in (set, frozenset, AbstractSet):
+        return getattr(converter, "enable_generic_sets", False)
     return callable(getattr(cast(Any, target_type), "from_json", None))
 
 
@@ -319,6 +362,22 @@ def _encode_custom_object(o: Any) -> Any:
     to_json_hook = getattr(cast(Any, type(o)), "to_json", None)
     if callable(to_json_hook):
         return to_json_hook(o)
+    if isinstance(o, (set, frozenset, AbstractSet)):
+        items: AbstractSet[Any] = cast("AbstractSet[Any]", o)
+        try:
+            return sorted(items)
+        except TypeError:
+            try:
+                def _set_sort_key(item: Any) -> tuple[str, str, str]:
+                    encoded = json.dumps(item, default=_encode_custom_object, sort_keys=True)
+                    return (encoded, type(item).__qualname__, repr(item))
+
+                return sorted(items, key=_set_sort_key)
+            except Exception as e:
+                raise TypeError(
+                    f"Cannot deterministically serialize set of type '{type(items).__name__}' "
+                    f"because its elements cannot be compared or deterministically encoded: {e}"
+                ) from e
     if isinstance(o, enum.Enum):
         # Emit the member's underlying value (a primitive). Reconstruction is
         # type-directed: passing the enum type to ``deserialize`` rebuilds the
@@ -477,6 +536,19 @@ def _coerce_generic(value: Any, expected_type: Any, origin: Any,
     if origin in (list, Sequence) and isinstance(value, list):
         elem_type = args[0] if args else None
         return [_coerce_to_type(item, elem_type, converter) for item in cast(list[Any], value)]
+    if origin in (set, frozenset, AbstractSet, typing.AbstractSet) and isinstance(
+        value, (list, set, frozenset, AbstractSet)
+    ):
+        if not getattr(converter, "enable_generic_sets", False):
+            return cast(Any, value)
+        elem_type = args[0] if args else None
+        coerced: list[Any] = [
+            _coerce_to_type(item, elem_type, converter)
+            for item in cast("Iterable[Any]", value)
+        ]
+        if origin is frozenset:
+            return frozenset(coerced)
+        return set(coerced)
     if origin in (dict, Mapping) and isinstance(value, dict):
         # JSON object keys are always strings, so only the values can carry a
         # reconstructable type. Keys are passed through unchanged.
@@ -595,12 +667,12 @@ def _resolve_forward_refs(tp: Any, globalns: dict[str, Any]) -> Any:
     """
     if isinstance(tp, str):
         try:
-            tp = eval(tp, globalns)  # noqa: S307 - resolves caller-authored annotations
+            tp = eval(tp, globalns)
         except Exception:
             return tp
     elif isinstance(tp, typing.ForwardRef):
         try:
-            tp = eval(tp.__forward_arg__, globalns)  # noqa: S307
+            tp = eval(tp.__forward_arg__, globalns)
         except Exception:
             return tp
 

tests/durabletask/test_data_converter.py

@@ -12,13 +12,14 @@
 from unittest.mock import patch
 
 import pytest
-
 from durabletask.internal.entity_state_shim import StateShim
 from durabletask.serialization import (
     DEFAULT_DATA_CONVERTER,
     DataConverter,
     JsonDataConverter,
 )
+from durabletask.internal import type_discovery
+from durabletask.worker import _ActivityExecutor, _Registry
 
 
 @dataclass
@@ -219,7 +220,7 @@ def __init__(self, /, **kwargs):
             self.self = kwargs.get("self", 0)
 
     converter = JsonDataConverter()
-    encoded = converter.serialize(PositionalReceiver(**{"self": 42}))
+    encoded = converter.serialize(PositionalReceiver(self=42))
     restored = converter.deserialize(encoded, PositionalReceiver)
     assert isinstance(restored, PositionalReceiver)
     assert restored.self == 42
@@ -375,3 +376,102 @@ def coerce(self, value: Any, target_type: type | None = None) -> Any:
     assert conv.serialize("hello") == '"HELLO"'
     assert conv.deserialize('"HELLO"') == "HELLO"
     assert conv.coerce("HELLO") == "HELLO"
+
+
+# ----- set and frozenset container support -----
+
+
+@dataclass(frozen=True)
+class TaggedOrder:
+    item: str
+    tags: set[str]
+    flags: frozenset[int]
+
+
+def test_roundtrip_dataclass_with_sets():
+    converter = JsonDataConverter(enable_generic_sets=True)
+    original = TaggedOrder(item="laptop", tags={"electronics", "sale"}, flags=frozenset({1, 2}))
+    serialized = converter.serialize(original)
+    assert serialized is not None
+    restored = converter.deserialize(serialized, TaggedOrder)
+    assert isinstance(restored, TaggedOrder)
+    assert restored.item == "laptop"
+    assert isinstance(restored.tags, set)
+    assert restored.tags == {"electronics", "sale"}
+    assert isinstance(restored.flags, frozenset)
+    assert restored.flags == frozenset({1, 2})
+
+
+def test_can_reconstruct_set_and_frozenset():
+    # By default (enable_generic_sets=False), generic sets are not reconstructed to
+    # preserve backward compatibility and existing-instance replay behavior.
+    default_converter = JsonDataConverter()
+    assert default_converter.can_reconstruct(set[PricedOrder]) is False
+    assert default_converter.can_reconstruct(frozenset[PricedOrder]) is False
+    assert default_converter.can_reconstruct(set[int]) is False
+    assert default_converter.can_reconstruct(set[str]) is False
+
+    # When opted in, supported sets (including primitive elements) return True,
+    # while unsupported unhashable element types return False.
+    opt_in_converter = JsonDataConverter(enable_generic_sets=True)
+    assert opt_in_converter.can_reconstruct(set[PricedOrder]) is True
+    assert opt_in_converter.can_reconstruct(frozenset[PricedOrder]) is True
+    assert opt_in_converter.can_reconstruct(set[int]) is True
+    assert opt_in_converter.can_reconstruct(set[str]) is True
+    assert opt_in_converter.can_reconstruct(frozenset[int]) is True
+    assert opt_in_converter.can_reconstruct(set[list[str]]) is False
+    assert opt_in_converter.can_reconstruct(set[dict[str, int]]) is False
+
+
+def test_existing_instance_replay_compatibility_preserves_list():
+    """Preserve existing-instance replay behavior for recorded arrays under set annotations.
+
+    When generic set coercion is not explicitly enabled, historical JSON arrays
+    containing duplicates and order (e.g. {"tags": ["b", "a", "b"]}) remain raw
+    lists, avoiding NonDeterminismError during orchestration replay.
+    """
+    default_converter = JsonDataConverter()
+    recorded_input = '{"item": "laptop", "tags": ["b", "a", "b"], "flags": [1, 2, 1]}'
+    restored = default_converter.deserialize(recorded_input, TaggedOrder)
+    assert isinstance(restored, TaggedOrder)
+    assert isinstance(restored.tags, list)
+    assert restored.tags == ["b", "a", "b"]
+    assert len(restored.tags) == 3
+    assert isinstance(restored.flags, list)
+    assert restored.flags == [1, 2, 1]
+    assert len(restored.flags) == 3
+
+    # With opt-in enabled, set coercion deduplicates and converts to sets
+    opt_in_converter = JsonDataConverter(enable_generic_sets=True)
+    opt_in_restored = opt_in_converter.deserialize(recorded_input, TaggedOrder)
+    assert isinstance(opt_in_restored, TaggedOrder)
+    assert isinstance(opt_in_restored.tags, set)
+    assert opt_in_restored.tags == {"a", "b"}
+    assert len(opt_in_restored.tags) == 2
+    assert isinstance(opt_in_restored.flags, frozenset)
+    assert opt_in_restored.flags == frozenset({1, 2})
+    assert len(opt_in_restored.flags) == 2
+
+
+def test_type_discovery_and_activity_execution_with_sets():
+    """Verify that type discovery recognizes set annotations when opted in,
+    enabling activities to receive sets rather than raw lists."""
+    def tag_activity(ctx: Any, tags: set[str]) -> int:
+        assert isinstance(tags, set)
+        tags.add("processed")
+        return len(tags)
+
+    reg = _Registry()
+    reg.add_activity(tag_activity)
+    logger = logging.getLogger("test_type_discovery_sets")
+
+    # With opt-in, activity input is discovered as set[str] and executed
+    conv = JsonDataConverter(enable_generic_sets=True)
+    assert type_discovery.activity_input_type(tag_activity, conv) == set[str]
+    executor = _ActivityExecutor(reg, logger, conv)
+    output = executor.execute("orch-1", "tag_activity", 1, '["item1", "item2"]')
+    assert output == "3"
+
+    # Default converter preserves input as None (raw JSON), passing list
+    default_conv = JsonDataConverter()
+    assert type_discovery.activity_input_type(tag_activity, default_conv) is None

tests/durabletask/test_data_converter_roundtrip.py

@@ -545,8 +545,6 @@ class HasUnion:
 @pytest.mark.parametrize("value", [
     datetime(2020, 1, 1, 12, 0, 0),
     Decimal("1.5"),
-    {1, 2, 3},
-    frozenset({1, 2}),
     b"bytes",
 ])
 def test_unsupported_types_raise_clear_typeerror(conv, value):