From 4dad104d8ecee54babe87dc033700d2f2c359134 Mon Sep 17 00:00:00 2001
From: Andy Staples <andystaples@microsoft.com>
Date: Fri, 30 Oct 2026 10:28:02 -0600
Subject: [PATCH 1/4] Optimize deterministic set serialization without JSON
 parse round trips

Normalize set subtrees once, reuse normalized values for output, and skip singleton ordering keys while preserving legacy coercion defaults.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                             |   4 +
 durabletask/serialization.py             |  66 +++++++---
 tests/durabletask/test_data_converter.py |  28 +++-
 tests/durabletask/test_serialization.py  | 158 ++++++++++++++++++++++-
 4 files changed, 236 insertions(+), 20 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e2914aff..958bdf4a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,10 @@ by one or more task hubs.
 
 FIXED
 
+- Optimized set and frozenset serialization, particularly for deeply nested
+frozensets and large singleton elements. JSON object names within sets are now
+ordered consistently after conversion to strings, including numeric names.
+Set reconstruction remains opt-in.
 - Implemented `__hash__` and `__repr__` for `EntityInstanceId`, making identity attributes immutable to guarantee hash stability when used in sets or dict keys.
 - Large-payload externalization now checks every eligible payload's maximum
 size before uploading or replacing any protobuf values, including carryover
diff --git a/durabletask/serialization.py b/durabletask/serialization.py
index c2d916f0..724da76e 100644
--- a/durabletask/serialization.py
+++ b/durabletask/serialization.py
@@ -137,6 +137,8 @@ class JsonDataConverter(DataConverter):
     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]``).
+    Within sets, object keys are normalized to strings and sorted, and custom
+    hooks run once per occurrence. Singleton sets need no ordering key.
 
     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
@@ -368,25 +370,9 @@ def _encode_custom_object(o: Any) -> Any:
     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]).
+        items = cast(AbstractSet[Any], o)
         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]
+            return _normalize_set_value(items, set())
         except Exception as e:
             raise TypeError(
                 f"Cannot deterministically serialize set of type '{type(items).__name__}' "
@@ -415,6 +401,50 @@ def _encode_custom_object(o: Any) -> Any:
     raise TypeError(f"Object of type '{type(o).__name__}' is not JSON serializable")
 
 
+def _normalize_set_value(value: Any, active: set[int]) -> Any:
+    """Normalize only set subtrees; leave ordinary payload encoding unchanged."""
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    identity = id(value)
+    if identity in active:
+        raise ValueError("Circular reference detected")
+    active.add(identity)
+    try:
+        if isinstance(value, dict):
+            # Match sort_keys=True, including rejection of incomparable keys.
+            # JSON object names are strings; duplicate names keep the last value,
+            # just as the previous dumps/loads normalization did.
+            result: dict[str, Any] = {}
+            string_names = True
+            for key, item in sorted(cast(dict[Any, Any], value).items()):
+                if key is not None and not isinstance(key, (str, int, float, bool)):
+                    raise TypeError(f"Unsupported JSON object key: {type(key).__name__}")
+                if isinstance(key, str):
+                    name = key
+                else:
+                    name = json.dumps(key)
+                    string_names = False
+                result[name] = _normalize_set_value(item, active)
+            return result if string_names else dict(sorted(result.items()))
+        if isinstance(value, (list, tuple)):
+            return [
+                item if item is None or isinstance(item, (str, int, float))
+                else _normalize_set_value(item, active)
+                for item in cast(Sequence[Any], value)
+            ]
+        to_json_hook = getattr(cast(Any, type(value)), "to_json", None)
+        if isinstance(value, AbstractSet) and not callable(to_json_hook):
+            items = [_normalize_set_value(item, active) for item in cast(AbstractSet[Any], value)]
+            if len(items) > 1:
+                # list.sort evaluates each key once and discards the temporary
+                # documents afterwards. Emit these same normalized values.
+                items.sort(key=json.dumps)
+            return items
+        return _normalize_set_value(_encode_custom_object(value), active)
+    finally:
+        active.remove(identity)
+
+
 def _legacy_object_hook(d: dict[str, Any]) -> Any:
     # If the object carries the legacy marker, deserialize it as a SimpleNamespace.
     if d.pop(_AUTO_SERIALIZED, False):
diff --git a/tests/durabletask/test_data_converter.py b/tests/durabletask/test_data_converter.py
index 8186280f..1f6b9ff1 100644
--- a/tests/durabletask/test_data_converter.py
+++ b/tests/durabletask/test_data_converter.py
@@ -12,6 +12,9 @@
 from unittest.mock import patch
 
 import pytest
+from durabletask import task
+from durabletask.internal import helpers
+from durabletask.internal import orchestrator_service_pb2 as pb
 from durabletask.internal.entity_state_shim import StateShim
 from durabletask.serialization import (
     DEFAULT_DATA_CONVERTER,
@@ -19,7 +22,7 @@
     JsonDataConverter,
 )
 from durabletask.internal import type_discovery
-from durabletask.worker import _ActivityExecutor, _Registry
+from durabletask.worker import _ActivityExecutor, _OrchestrationExecutor, _Registry
 
 
 @dataclass
@@ -453,6 +456,29 @@ def test_existing_instance_replay_compatibility_preserves_list():
     assert len(opt_in_restored.flags) == 2
 
 
+def test_default_converter_replays_recorded_arrays_under_set_annotations():
+    def orchestrator(ctx: task.OrchestrationContext, data: TaggedOrder):
+        yield ctx.call_activity(f"count-{len(data.tags)}")
+        return {"tags": data.tags, "flags": data.flags}
+
+    registry = _Registry()
+    name = registry.add_orchestrator(orchestrator)
+    history = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(
+            name, "legacy-sets",
+            encoded_input='{"item": "laptop", "tags": ["b", "a", "b"], "flags": [1, 2, 1]}',
+        ),
+        helpers.new_task_scheduled_event(1, "count-3"),
+    ]
+    executor = _OrchestrationExecutor(registry, logging.getLogger("test_set_replay"), JsonDataConverter())
+    result = executor.execute("legacy-sets", history, [helpers.new_task_completed_event(1, "null")])
+    assert len(result.actions) == 1
+    completion = result.actions[0].completeOrchestration
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert completion.result.value == '{"tags": ["b", "a", "b"], "flags": [1, 2, 1]}'
+
+
 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."""
diff --git a/tests/durabletask/test_serialization.py b/tests/durabletask/test_serialization.py
index 5bd7851a..ce42fd09 100644
--- a/tests/durabletask/test_serialization.py
+++ b/tests/durabletask/test_serialization.py
@@ -21,6 +21,7 @@
 import pytest
 from durabletask.serialization import _AUTO_SERIALIZED as AUTO_SERIALIZED
 from durabletask.serialization import JsonDataConverter
+from durabletask.serialization import _encode_custom_object
 from durabletask.serialization import _coerce_to_type as coerce_to_type
 from durabletask.serialization import _from_json as from_json
 from durabletask.serialization import _resolve_forward_refs as resolve_forward_refs
@@ -585,6 +586,16 @@ class FrozenAddress:
     city: str
 
 
+class SetItem:
+    """Identity-hashable value exposing a custom JSON representation."""
+
+    def __init__(self, value: Any) -> None:
+        self.value = value
+
+    def to_json(self) -> Any:
+        return self.value
+
+
 def test_to_json_serializes_set_and_frozenset():
     encoded_set = to_json({"b", "a"})
     assert json.loads(encoded_set) == ["a", "b"]
@@ -656,6 +667,143 @@ def __eq__(self, other: object) -> bool:
     assert parsed[1] == {"a": 1, "m": 50, "z": 100}
 
 
+@pytest.mark.parametrize("depth", [0, 1, 10])
+def test_set_serialization_calls_nested_hooks_once(depth: int):
+    class CountingItem:
+        calls = 0
+
+        def to_json(self):
+            self.calls += 1
+            return {"nested": [StaticWidget("value")]}
+
+    item = CountingItem()
+    value: Any = item
+    expected: Any = {"nested": ["value"]}
+    for _ in range(depth):
+        value = frozenset({value})
+        expected = [expected]
+    assert json.loads(to_json({value})) == [expected]
+    assert item.calls == 1
+
+
+def test_set_serialization_does_not_parse_or_encode_singleton_sort_keys(monkeypatch: pytest.MonkeyPatch):
+    import durabletask.serialization as serialization
+
+    dumps = json.dumps
+    encoded_values: list[Any] = []
+
+    def record_dumps(value: Any, **kwargs: Any) -> str:
+        encoded_values.append(value)
+        return dumps(value, **kwargs)
+
+    def unexpected_loads(*args: Any, **kwargs: Any) -> Any:
+        pytest.fail("Set serialization must not parse intermediate JSON")
+
+    monkeypatch.setattr(serialization.json, "loads", unexpected_loads)
+    monkeypatch.setattr(serialization.json, "dumps", record_dumps)
+    value = frozenset({frozenset({SetItem("x" * 100000)})})
+    assert to_json(value) == dumps([["x" * 100000]])
+    assert encoded_values == [value]
+
+
+def test_set_serialization_sorts_each_normalized_element_once(monkeypatch: pytest.MonkeyPatch):
+    import durabletask.serialization as serialization
+
+    dumps = json.dumps
+    sort_values: list[Any] = []
+
+    def record_dumps(value: Any, **kwargs: Any) -> str:
+        if not kwargs:
+            sort_values.append(value)
+        return dumps(value, **kwargs)
+
+    monkeypatch.setattr(serialization.json, "dumps", record_dumps)
+    assert to_json({SetItem("b"), SetItem("a")}) == '["a", "b"]'
+    assert sorted(sort_values) == ["a", "b"]
+
+
+def test_set_serialization_matches_canonical_json_order():
+    values = [
+        None, True, False, -1, 1, 10, 1.0, 10.0, float("nan"),
+        float("-inf"), float("inf"), "", '"', "\\", "\n", "\u2603",
+        [], [1], [10], [1, 2], [1.0], [{}], [[]],
+        {}, {"a": 1}, {"a": 10}, {"a": 1, "b": []},
+        {"z": [1, {"b": 2, "a": 1}], "a": [[], {}]},
+    ]
+    expected = "[" + ", ".join(sorted(json.dumps(value, sort_keys=True) for value in values)) + "]"
+    assert to_json({SetItem(value) for value in values}) == expected
+
+
+def test_set_serialization_normalizes_json_object_names_before_ordering():
+    # Numeric names become strings on the wire, just like string names.
+    numeric = SetItem({2: "two", 10: "ten"})
+    strings = SetItem({"2": "two", "10": "ten"})
+    expected = '{"10": "ten", "2": "two"}'
+    assert to_json({numeric, strings}) == f"[{expected}, {expected}]"
+    assert to_json({frozenset({numeric}), frozenset({strings})}) == f"[[{expected}], [{expected}]]"
+
+
+@pytest.mark.parametrize("shape", ["list", "dict", "hook", "set"])
+def test_set_serialization_rejects_cycles(shape: str):
+    class Item:
+        def to_json(self):
+            return value
+
+    item = Item()
+    value: Any
+    if shape == "list":
+        value = [item]
+    elif shape == "dict":
+        value = {"item": item}
+    elif shape == "hook":
+        value = item
+    else:
+        value = {item}
+    with pytest.raises(TypeError, match="Circular reference detected"):
+        to_json({item})
+
+
+def test_set_serialization_allows_shared_non_circular_values():
+    shared = {"b": 2, "a": 1}
+    assert to_json({SetItem([shared, shared])}) == '[[{"a": 1, "b": 2}, {"a": 1, "b": 2}]]'
+
+
+def test_set_serialization_preserves_native_and_custom_hook_precedence():
+    class HookedTuple(tuple):
+        def to_json(self):
+            pytest.fail("JSON-native containers must not invoke to_json")
+
+    class HookedFrozenSet(frozenset):
+        def to_json(self):
+            return "custom set"
+
+    class Item:
+        def to_json(self):
+            return SimpleNamespace(
+                z=HookedTuple([Temperature(100.0)]),
+                a=HookedFrozenSet({1}),
+            )
+
+    assert to_json({Item()}) == '[{"a": "custom set", "z": [{"fahrenheit": 212.0}]}]'
+
+
+@pytest.mark.parametrize("value", [object(), {object(): 1}, {"a": 1, 2: 3}])
+def test_set_serialization_preserves_encoding_errors(value: Any):
+    class Item:
+        def to_json(self):
+            return value
+
+    with pytest.raises(TypeError, match="Cannot deterministically serialize set") as error:
+        to_json({Item()})
+    assert error.value.__cause__ is not None
+
+
+def test_serialization_outside_sets_matches_legacy_json_encoder():
+    value = {"z": [Temperature(100.0), {"b": 2, "a": 1}], "a": (1, 2)}
+    assert to_json(value) == json.dumps(value, default=_encode_custom_object)
+    assert to_json(value) == '{"z": [{"fahrenheit": 212.0}, {"b": 2, "a": 1}], "a": [1, 2]}'
+
+
 def test_set_serialization_deterministic_across_hash_seeds():
     """Verify that sets of disjoint frozensets, nested frozensets, unorderable custom objects,
     non-finite numbers, and equal canonical keys with differently ordered dictionaries
@@ -715,7 +863,13 @@ def __eq__(self, other: object) -> bool:
     DictItem({"a": 1, "z": 10}, "item2"),
 })
 
-print(json.dumps({"a": case_a, "b": case_b, "c": case_c, "d": case_d, "e": case_e}))
+# Case F: Different Python key types representing identical JSON object names.
+case_f = conv.serialize({
+    DictItem({2: "two", 10: "ten"}, "numeric"),
+    DictItem({"2": "two", "10": "ten"}, "strings"),
+})
+
+print(json.dumps({"a": case_a, "b": case_b, "c": case_c, "d": case_d, "e": case_e, "f": case_f}))
 """
     seeds = ["0", "1", "2", "42", "999"]
     outputs: dict[str, dict[str, str]] = {}
@@ -754,12 +908,14 @@ def __eq__(self, other: object) -> bool:
     )
     expected_d = '[-Infinity, 10.0, 2.0, Infinity, NaN]'
     expected_e = '[{"a": 1, "z": 10}, {"a": 1, "z": 10}]'
+    expected_f = '[{"10": "ten", "2": "two"}, {"10": "ten", "2": "two"}]'
 
     assert base["a"] == expected_a
     assert base["b"] == expected_b
     assert base["c"] == expected_c
     assert base["d"] == expected_d
     assert base["e"] == expected_e
+    assert base["f"] == expected_f
 
 
 def test_deserialize_coerces_set_when_opted_in():

From 4e83deeaeb03988ec1aeb2fd68ca9db279df266c Mon Sep 17 00:00:00 2001
From: Andy Staples <andystaples@microsoft.com>
Date: Fri, 30 Oct 2026 10:36:54 -0600
Subject: [PATCH 2/4] Preserve Unicode object-name normalization in set
 payloads

Collapse equivalent UTF-16 surrogate-pair and non-BMP names before dictionary insertion, preserve lone surrogates, and consolidate test imports.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                            |  2 ++
 durabletask/serialization.py            |  7 ++++++
 tests/durabletask/test_serialization.py | 31 +++++++++++++++++++++----
 3 files changed, 36 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 958bdf4a..2f4dd412 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,8 @@ FIXED
 - Optimized set and frozenset serialization, particularly for deeply nested
 frozensets and large singleton elements. JSON object names within sets are now
 ordered consistently after conversion to strings, including numeric names.
+Equivalent surrogate-pair and non-BMP object names retain the last value
+rather than emitting duplicate JSON properties.
 Set reconstruction remains opt-in.
 - Implemented `__hash__` and `__repr__` for `EntityInstanceId`, making identity attributes immutable to guarantee hash stability when used in sets or dict keys.
 - Large-payload externalization now checks every eligible payload's maximum
diff --git a/durabletask/serialization.py b/durabletask/serialization.py
index 724da76e..ffe35aed 100644
--- a/durabletask/serialization.py
+++ b/durabletask/serialization.py
@@ -421,6 +421,13 @@ def _normalize_set_value(value: Any, active: set[int]) -> Any:
                     raise TypeError(f"Unsupported JSON object key: {type(key).__name__}")
                 if isinstance(key, str):
                     name = key
+                    if not str.isascii(name):
+                        # JSON decoding combines UTF-16 surrogate pairs, but
+                        # leaves lone surrogates intact. Match its object names
+                        # without parsing a document or losing duplicate names.
+                        name = str.encode(name, "utf-16-le", "surrogatepass").decode(
+                            "utf-16-le", "surrogatepass")
+                        string_names = string_names and name == key
                 else:
                     name = json.dumps(key)
                     string_names = False
diff --git a/tests/durabletask/test_serialization.py b/tests/durabletask/test_serialization.py
index ce42fd09..4ae1cc68 100644
--- a/tests/durabletask/test_serialization.py
+++ b/tests/durabletask/test_serialization.py
@@ -19,6 +19,7 @@
 from typing import Any, List, Optional, Union, get_args
 
 import pytest
+from durabletask import serialization
 from durabletask.serialization import _AUTO_SERIALIZED as AUTO_SERIALIZED
 from durabletask.serialization import JsonDataConverter
 from durabletask.serialization import _encode_custom_object
@@ -687,8 +688,6 @@ def to_json(self):
 
 
 def test_set_serialization_does_not_parse_or_encode_singleton_sort_keys(monkeypatch: pytest.MonkeyPatch):
-    import durabletask.serialization as serialization
-
     dumps = json.dumps
     encoded_values: list[Any] = []
 
@@ -707,8 +706,6 @@ def unexpected_loads(*args: Any, **kwargs: Any) -> Any:
 
 
 def test_set_serialization_sorts_each_normalized_element_once(monkeypatch: pytest.MonkeyPatch):
-    import durabletask.serialization as serialization
-
     dumps = json.dumps
     sort_values: list[Any] = []
 
@@ -743,6 +740,30 @@ def test_set_serialization_normalizes_json_object_names_before_ordering():
     assert to_json({frozenset({numeric}), frozenset({strings})}) == f"[[{expected}], [{expected}]]"
 
 
+@pytest.mark.parametrize("reverse", [False, True])
+def test_set_serialization_collapses_equivalent_unicode_object_names(reverse: bool):
+    pairs = [("\ud83d\ude00", 1), ("\U0001f600", 2)]
+    value = dict(reversed(pairs) if reverse else pairs)
+    expected = '[{"\\ud83d\\ude00": 2}]'
+    assert to_json({SetItem(value)}) == expected
+    assert to_json({frozenset({SetItem(value)})}) == f"[{expected}]"
+
+
+@pytest.mark.parametrize("name", [
+    "ascii", "\u2603", "\U0001f600", "\ud83d\ude00", "\ud83d",
+    "\ude00", "\ud83d\ud83d\ude00", "\ude00\ud83d", "\ud83dA\ude00",
+])
+def test_set_serialization_unicode_names_match_json_normalization(name: str):
+    value = {name: 1}
+    expected = json.dumps([json.loads(json.dumps(value, sort_keys=True))])
+    assert to_json({SetItem(value)}) == expected
+
+
+def test_set_serialization_orders_normalized_unicode_names():
+    value = {"\ud83d\ude00": 1, "\ue000": 2}
+    assert to_json({SetItem(value)}) == '[{"\\ue000": 2, "\\ud83d\\ude00": 1}]'
+
+
 @pytest.mark.parametrize("shape", ["list", "dict", "hook", "set"])
 def test_set_serialization_rejects_cycles(shape: str):
     class Item:
@@ -802,6 +823,8 @@ def test_serialization_outside_sets_matches_legacy_json_encoder():
     value = {"z": [Temperature(100.0), {"b": 2, "a": 1}], "a": (1, 2)}
     assert to_json(value) == json.dumps(value, default=_encode_custom_object)
     assert to_json(value) == '{"z": [{"fahrenheit": 212.0}, {"b": 2, "a": 1}], "a": [1, 2]}'
+    names = {"\ud83d\ude00": 1, "\U0001f600": 2}
+    assert to_json(names) == json.dumps(names)
 
 
 def test_set_serialization_deterministic_across_hash_seeds():

From 90d87038aad1e2c448ca41bf7ee3bdb1961c177e Mon Sep 17 00:00:00 2001
From: Andy Staples <andystaples@microsoft.com>
Date: Fri, 30 Oct 2026 10:40:43 -0600
Subject: [PATCH 3/4] Align opt-in set discovery with tuple and abstract-set
 reconstruction

Recognize hashable typed tuples recursively and reconstruct bare collections.abc.Set annotations as concrete sets only when enabled. Cover both opt-in names, discovery, activity execution, and unchanged legacy defaults.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                             |  3 ++
 durabletask/serialization.py             | 11 ++++-
 tests/durabletask/test_data_converter.py | 57 +++++++++++++++++++++++-
 3 files changed, 69 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2f4dd412..56778154 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,6 +20,9 @@ ordered consistently after conversion to strings, including numeric names.
 Equivalent surrogate-pair and non-BMP object names retain the last value
 rather than emitting duplicate JSON properties.
 Set reconstruction remains opt-in.
+- With set reconstruction enabled, annotated sets of typed tuples are now
+recognized during input and activity-result discovery, and bare
+`collections.abc.Set` annotations reconstruct JSON arrays as concrete sets.
 - Implemented `__hash__` and `__repr__` for `EntityInstanceId`, making identity attributes immutable to guarantee hash stability when used in sets or dict keys.
 - Large-payload externalization now checks every eligible payload's maximum
 size before uploading or replacing any protobuf values, including carryover
diff --git a/durabletask/serialization.py b/durabletask/serialization.py
index ffe35aed..66698a00 100644
--- a/durabletask/serialization.py
+++ b/durabletask/serialization.py
@@ -249,6 +249,11 @@ def _is_supported_set_element(converter: DataConverter, elem_type: Any) -> bool:
         if elem_origin is frozenset:
             args = typing.get_args(elem_type)
             return not args or _is_supported_set_element(converter, args[0])
+        if elem_origin is tuple:
+            args = typing.get_args(elem_type)
+            if len(args) == 2 and args[1] is Ellipsis:
+                args = args[:1]
+            return all(_is_supported_set_element(converter, arg) for arg in args)
         return False
     if not isinstance(elem_type, type):
         return False
@@ -495,6 +500,10 @@ def _coerce_to_type(value: Any, expected_type: Any, converter: DataConverter | N
     if isinstance(value, expected_type):
         return value
 
+    if (expected_type is AbstractSet and isinstance(value, list)
+            and getattr(converter, "enable_generic_sets", False)):
+        return _coerce_generic(value, expected_type, AbstractSet, converter)
+
     from_json_hook = getattr(expected_type, "from_json", None)
     if callable(from_json_hook):
         return _invoke_from_json(from_json_hook, value, converter)
@@ -508,7 +517,7 @@ def _coerce_to_type(value: Any, expected_type: Any, converter: DataConverter | N
     except Exception as e:
         type_name = getattr(type_ctor, "__name__", None) or str(type_ctor)
         raise TypeError(
-            f"Could not coerce value of type '{type(value).__name__}' to "
+            f"Could not coerce value of type '{type(cast(object, value)).__name__}' to "
             f"'{type_name}'"
         ) from e
 
diff --git a/tests/durabletask/test_data_converter.py b/tests/durabletask/test_data_converter.py
index 1f6b9ff1..1ed5c552 100644
--- a/tests/durabletask/test_data_converter.py
+++ b/tests/durabletask/test_data_converter.py
@@ -6,9 +6,10 @@
 import inspect
 import json
 import logging
+from collections.abc import Set as AbstractSet
 from dataclasses import dataclass, field
 from functools import wraps
-from typing import Any
+from typing import AbstractSet as TypingAbstractSet, Any
 from unittest.mock import patch
 
 import pytest
@@ -426,6 +427,60 @@ def test_can_reconstruct_set_and_frozenset():
     assert opt_in_converter.can_reconstruct(set[dict[str, int]]) is False
 
 
+@pytest.mark.parametrize("option", ["enable_generic_sets", "enable_set_coercion"])
+@pytest.mark.parametrize("target_type,payload,expected", [
+    (set[tuple[int, int]], "[[1, 2], [3, 4]]", {(1, 2), (3, 4)}),
+    (frozenset[tuple[int, ...]], "[[1], [2, 3]]", frozenset({(1,), (2, 3)})),
+    (set[tuple[int, frozenset[str]]], '[[1, ["b", "a"]]]', {(1, frozenset({"a", "b"}))}),
+    (frozenset[frozenset[tuple[int, int]]], "[[[1, 2], [3, 4]]]",
+     frozenset({frozenset({(1, 2), (3, 4)})})),
+    (set[tuple[()]], "[[]]", {()}),
+    (set[tuple[int, str | None]], '[[1, null], [2, "x"]]', {(1, None), (2, "x")}),
+    (AbstractSet, "[2, 1, 2]", {1, 2}),
+    (TypingAbstractSet, "[2, 1, 2]", {1, 2}),
+])
+def test_opt_in_set_discovery_matches_reconstruction(
+    option: str, target_type: Any, payload: str, expected: Any,
+):
+    converter = JsonDataConverter(**{option: True})
+
+    def activity(ctx: Any, values: Any) -> Any:
+        assert isinstance(values, type(expected))
+        assert values == expected
+        return values
+
+    activity.__annotations__["values"] = target_type
+    activity.__annotations__["return"] = target_type
+    assert converter.can_reconstruct(target_type)
+    assert converter.deserialize(payload, target_type) == expected
+    assert converter.coerce(json.loads(payload), target_type) == expected
+    assert type_discovery.activity_input_type(activity, converter) == target_type
+    assert type_discovery.orchestrator_input_type(activity, converter) == target_type
+    assert type_discovery.activity_output_type(activity, converter) == target_type
+    registry = _Registry()
+    name = registry.add_activity(activity)
+    executor = _ActivityExecutor(registry, logging.getLogger("test_set_discovery"), converter)
+    output = executor.execute("instance", name, 1, payload)
+    assert converter.deserialize(output, target_type) == expected
+
+    default = JsonDataConverter()
+    assert not default.can_reconstruct(target_type)
+    assert type_discovery.activity_input_type(activity, default) is None
+    assert type_discovery.orchestrator_input_type(activity, default) is None
+    assert type_discovery.activity_output_type(activity, default) is None
+    assert default.deserialize(payload, target_type) == json.loads(payload)
+
+
+@pytest.mark.parametrize("target_type", [
+    set[tuple[list[int], int]],
+    set[tuple[dict[str, int], ...]],
+    set[tuple[frozenset[list[int]], int]],
+    frozenset[tuple[set[int], ...]],
+])
+def test_set_tuple_discovery_rejects_unhashable_elements(target_type: Any):
+    assert not JsonDataConverter(enable_generic_sets=True).can_reconstruct(target_type)
+
+
 def test_existing_instance_replay_compatibility_preserves_list():
     """Preserve existing-instance replay behavior for recorded arrays under set annotations.
 

From 943516d8f044a2095bee00972048c6cda54864a8 Mon Sep 17 00:00:00 2001
From: Andy Staples <andystaples@microsoft.com>
Date: Fri, 30 Oct 2026 12:51:41 -0600
Subject: [PATCH 4/4] Reduce dictionary-heavy set normalization overhead

Compose only actual surrogate pairs, bypass recursive normalization for scalar dictionary fields, and avoid cycle-tracking writes for scalar-only records. Cover reviewer payload shapes and retain per-occurrence hooks and cycle detection.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 durabletask/serialization.py            | 64 +++++++++++++++----------
 tests/durabletask/test_serialization.py | 44 +++++++++++++++++
 2 files changed, 82 insertions(+), 26 deletions(-)

diff --git a/durabletask/serialization.py b/durabletask/serialization.py
index 66698a00..22d09b22 100644
--- a/durabletask/serialization.py
+++ b/durabletask/serialization.py
@@ -38,6 +38,7 @@
 import inspect
 import json
 import logging
+import re
 import sys
 import types
 import typing
@@ -56,6 +57,8 @@
 # replay. Internal detail; re-exported from ``durabletask.internal.shared`` only
 # for backwards compatibility.
 _AUTO_SERIALIZED = "__durabletask_autoobject__"
+_SURROGATE_PAIR = re.compile("[\ud800-\udbff][\udc00-\udfff]")
+_JSON_SCALAR_TYPES = (int, str, float, type(None))
 
 
 class DataConverter(ABC):
@@ -408,39 +411,48 @@ def _encode_custom_object(o: Any) -> Any:
 
 def _normalize_set_value(value: Any, active: set[int]) -> Any:
     """Normalize only set subtrees; leave ordinary payload encoding unchanged."""
-    if value is None or isinstance(value, (str, int, float, bool)):
-        return value
     identity = id(value)
+    is_mapping = isinstance(value, dict)
+    if not is_mapping and isinstance(value, _JSON_SCALAR_TYPES):
+        return value
     if identity in active:
         raise ValueError("Circular reference detected")
+    if is_mapping:
+        # Match sort_keys=True, including rejection of incomparable keys.
+        # JSON object names are strings; duplicate names keep the last value,
+        # just as the previous dumps/loads normalization did.
+        result: dict[str, Any] = {}
+        string_names = True
+        for key, item in sorted(cast(dict[Any, Any], value).items()):
+            if isinstance(key, str):
+                name = key
+                if not str.isascii(name) and _SURROGATE_PAIR.search(name):
+                    # Ordinary Unicode names and lone surrogates need no codec
+                    # round trip; only a high/low pair changes after decoding.
+                    name = str.encode(name, "utf-16-le", "surrogatepass").decode(
+                        "utf-16-le", "surrogatepass")
+                    string_names = False
+            else:
+                if key is not None and not isinstance(key, (int, float)):
+                    raise TypeError(f"Unsupported JSON object key: {type(key).__name__}")
+                name = json.dumps(key)
+                string_names = False
+            if isinstance(item, _JSON_SCALAR_TYPES):
+                result[name] = item
+            else:
+                # Scalar-only records cannot cycle. Track the parent only
+                # while descending into a child that could refer back to it.
+                active.add(identity)
+                try:
+                    result[name] = _normalize_set_value(item, active)
+                finally:
+                    active.remove(identity)
+        return result if string_names else dict(sorted(result.items()))
     active.add(identity)
     try:
-        if isinstance(value, dict):
-            # Match sort_keys=True, including rejection of incomparable keys.
-            # JSON object names are strings; duplicate names keep the last value,
-            # just as the previous dumps/loads normalization did.
-            result: dict[str, Any] = {}
-            string_names = True
-            for key, item in sorted(cast(dict[Any, Any], value).items()):
-                if key is not None and not isinstance(key, (str, int, float, bool)):
-                    raise TypeError(f"Unsupported JSON object key: {type(key).__name__}")
-                if isinstance(key, str):
-                    name = key
-                    if not str.isascii(name):
-                        # JSON decoding combines UTF-16 surrogate pairs, but
-                        # leaves lone surrogates intact. Match its object names
-                        # without parsing a document or losing duplicate names.
-                        name = str.encode(name, "utf-16-le", "surrogatepass").decode(
-                            "utf-16-le", "surrogatepass")
-                        string_names = string_names and name == key
-                else:
-                    name = json.dumps(key)
-                    string_names = False
-                result[name] = _normalize_set_value(item, active)
-            return result if string_names else dict(sorted(result.items()))
         if isinstance(value, (list, tuple)):
             return [
-                item if item is None or isinstance(item, (str, int, float))
+                item if isinstance(item, _JSON_SCALAR_TYPES)
                 else _normalize_set_value(item, active)
                 for item in cast(Sequence[Any], value)
             ]
diff --git a/tests/durabletask/test_serialization.py b/tests/durabletask/test_serialization.py
index 4ae1cc68..2e70d2eb 100644
--- a/tests/durabletask/test_serialization.py
+++ b/tests/durabletask/test_serialization.py
@@ -764,6 +764,50 @@ def test_set_serialization_orders_normalized_unicode_names():
     assert to_json({SetItem(value)}) == '[{"\\ue000": 2, "\\ud83d\\ude00": 1}]'
 
 
+@pytest.mark.parametrize("value", [
+    {"\u2603" + str(i): i for i in range(10000)},
+    [{"a": i, "b": i + 1, "c": i + 2} for i in range(1000)],
+])
+def test_set_serialization_dictionary_heavy_payloads(value: Any):
+    class CountingItem(SetItem):
+        calls = 0
+
+        def to_json(self) -> Any:
+            self.calls += 1
+            return self.value
+
+    item = CountingItem(value)
+    assert to_json({item}) == json.dumps([value], sort_keys=True)
+    assert item.calls == 1
+
+
+@pytest.mark.parametrize("path", ["dict", "list", "hook"])
+def test_set_serialization_tracks_dictionary_cycles_through_children(path: str):
+    value: dict[str, Any] = {"a": 1}
+    if path == "dict":
+        value["z"] = value
+    elif path == "list":
+        value["z"] = [value]
+    else:
+        value["z"] = SetItem(value)
+    with pytest.raises(TypeError, match="Circular reference detected"):
+        to_json({SetItem(value)})
+
+
+def test_set_serialization_normalizes_dictionary_children_once_per_occurrence():
+    class CountingItem(SetItem):
+        calls = 0
+
+        def to_json(self) -> Any:
+            self.calls += 1
+            return self.value
+
+    child = CountingItem({"b": 2, "a": 1})
+    value = {"a": 0, "b": child, "c": child}
+    assert to_json({SetItem(value)}) == '[{"a": 0, "b": {"a": 1, "b": 2}, "c": {"a": 1, "b": 2}}]'
+    assert child.calls == 2
+
+
 @pytest.mark.parametrize("shape", ["list", "dict", "hook", "set"])
 def test_set_serialization_rejects_cycles(shape: str):
     class Item: