@@ -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.
@@ -136,7 +134,9 @@ class JsonDataConverter(DataConverter):
     ``SimpleNamespace`` handling, and nested values are encoded recursively, so
     a dataclass (or any object) with a ``to_json`` hook -- including one nested
     inside another dataclass, ``list``, or ``dict`` -- round-trips through its
-    own hooks.
+    own hooks. Sets and frozensets serialize to JSON arrays ordered
+    deterministically by their canonical JSON representation (using lexical
+    string order, so numeric sets like ``{2, 10}`` serialize as ``[10, 2]``).
 
     Deserialization (and value-level :meth:`coerce`) is **best-effort**: when a
     ``target_type`` is supplied and the value cannot be coerced to it, the raw
@@ -167,6 +167,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 +233,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 +277,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))
 
 
@@ -253,9 +298,12 @@ def _to_json(obj: Any) -> str:
 
     Builtins serialize to plain JSON. Objects exposing a ``to_json()`` method,
     dataclasses, and ``SimpleNamespace`` instances are serialized to plain JSON
-    as well (without any type marker), recursively. Custom objects can be
-    reconstructed on the receiving side by passing ``expected_type`` to
-    :func:`_from_json`.
+    as well (without any type marker), recursively. Sets and frozensets are
+    serialized as JSON arrays with elements deterministically ordered by their
+    canonical JSON string representation (note that numeric set elements follow
+    lexical JSON ordering, e.g. ``{2, 10}`` serializes as ``[10, 2]``). Custom
+    objects can be reconstructed on the receiving side by passing ``expected_type``
+    to :func:`_from_json`.
     """
     try:
         return json.dumps(obj, default=_encode_custom_object)
@@ -319,6 +367,31 @@ 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)
+        # Sets and frozensets are unordered collections. To ensure deterministic
+        # serialization across processes, worker replays, and PYTHONHASHSEED
+        # variations, elements are individually encoded to canonical JSON
+        # (with sort_keys=True), sorted lexically by their encoded JSON string,
+        # and emitted as normalized parsed values.
+        #
+        # Emitting the normalized parsed values rather than original objects
+        # guarantees identical order and eliminates leaks from dict insertion
+        # order variations, NaN/non-finite comparison anomalies, or partial
+        # ordering in frozensets. A consequence of sorting canonical JSON strings
+        # is lexical ordering for numeric sets (e.g., {2, 10} serializes as
+        # [10, 2]).
+        try:
+            encoded_items = sorted(
+                json.dumps(item, default=_encode_custom_object, sort_keys=True)
+                for item in items
+            )
+            return [json.loads(encoded) for encoded in encoded_items]
+        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 +550,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 +681,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
 
 
@@ -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
@@ -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):