diff --git a/CHANGELOG.md b/CHANGELOG.md
index 70a2b505..4d4e2072 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,26 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+FIXED
+
+- Type discovery now recognizes `dict` / `Mapping` and `tuple` containers wrapping
+reconstructable types (such as dataclasses and `from_json()`-capable types) via
+`JsonDataConverter.can_reconstruct()`. Inbound activity and orchestrator inputs as
+well as inferred activity return types annotated with these containers are now
+automatically reconstructed into typed model instances without requiring an explicit
+`return_type` override.
+
+> [!WARNING]
+> **Replay-breaking bug fix:** Container annotations involving `dict`, `Mapping`,
+> or `tuple` wrapping reconstructable types, including nested annotations, now
+> reconstruct orchestration inputs and inferred activity results by default.
+> Existing instances that relied on raw dictionaries or lists can change behavior
+> or fail replay with `NonDeterminismError`.
+>
+> Drain affected instances on the previous SDK, or keep their task hub on the
+> previous SDK and use a separate task hub for new instances. Pin or lock the core
+> `durabletask` dependency, including when using the Azure Managed provider.
+
 ## v1.13.1
 
 ADDED
diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md
index 5c2be962..ac1dbe88 100644
--- a/durabletask-azuremanaged/CHANGELOG.md
+++ b/durabletask-azuremanaged/CHANGELOG.md
@@ -7,6 +7,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+FIXED
+
+- With the corresponding core `durabletask` SDK update, type discovery now recognizes
+`dict`, `Mapping`, and `tuple` container annotations wrapping reconstructable types,
+automatically reconstructing recorded inputs and inferred activity results into typed model
+instances. See the [core upgrade warning](../CHANGELOG.md#unreleased) for replay-breaking
+guidance and dependency pinning recommendations.
+
 ## v1.13.1
 
 ADDED
diff --git a/durabletask/serialization.py b/durabletask/serialization.py
index 45d045a0..7f10cae8 100644
--- a/durabletask/serialization.py
+++ b/durabletask/serialization.py
@@ -267,15 +267,14 @@ def _is_supported_set_element(converter: DataConverter, elem_type: Any) -> bool:
 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. 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
+    Recognizes dataclasses and ``from_json()``-capable types, plus ``Optional``,
+    ``list`` / ``Sequence``, ``dict`` / ``Mapping``, and ``tuple`` hints wrapping
+    them; builtins and unknown 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.
+    still participates in the element-type checks of wrapped hints.
     """
     origin = typing.get_origin(target_type)
     if origin is not None:
@@ -292,6 +291,14 @@ def _can_reconstruct(converter: DataConverter, target_type: Any) -> bool:
             if not args:
                 return True
             return _is_supported_set_element(converter, args[0])
+        if origin in (dict, Mapping):
+            return len(args) == 2 and converter.can_reconstruct(args[1])
+        if origin is tuple:
+            if not args:
+                return False
+            if len(args) == 2 and args[1] is Ellipsis:
+                return converter.can_reconstruct(args[0])
+            return any(converter.can_reconstruct(a) for a in args)
         return False
     if not isinstance(target_type, type):
         return False
diff --git a/tests/azure-functions-durable/test_converters.py b/tests/azure-functions-durable/test_converters.py
index e01c8034..2f9da6f5 100644
--- a/tests/azure-functions-durable/test_converters.py
+++ b/tests/azure-functions-durable/test_converters.py
@@ -361,3 +361,54 @@ def test_durable_client_decode_preserves_host_configuration():
     value = DurableClientConverter.decode(
         meta.Datum(type="string", value=config), trigger_metadata=None)
     assert value == config
+
+
+# ---------------------------------------------------------------------------
+# FunctionsDataConverter reconstruction and discovery
+# ---------------------------------------------------------------------------
+
+def test_functions_data_converter_can_reconstruct_and_discovery_unaffected():
+    """Azure Functions overrides can_reconstruct() to return True uniformly.
+
+    The core SDK's expanded dict/tuple traversal in JsonDataConverter does not
+    alter Azure Functions discovery behavior, because FunctionsDataConverter
+    already returns True for all types (delegating decoding to df_loads).
+    """
+    from collections.abc import Mapping
+    from dataclasses import dataclass
+    from durabletask.internal import type_discovery
+
+    @dataclass
+    class Order:
+        item: str
+        quantity: int
+
+    conv = FunctionsDataConverter()
+
+    # Always returns True for affected container types
+    assert conv.can_reconstruct(dict[str, Order]) is True
+    assert conv.can_reconstruct(Mapping[str, Order]) is True
+    assert conv.can_reconstruct(tuple[Order, ...]) is True
+    assert conv.can_reconstruct(tuple[Order, int]) is True
+
+    # Always returns True even for builtins and unknown types (distinct from core)
+    assert conv.can_reconstruct(dict[str, int]) is True
+    assert conv.can_reconstruct(tuple[int, str]) is True
+    assert conv.can_reconstruct(int) is True
+    assert conv.can_reconstruct(dict) is True
+    assert conv.can_reconstruct(tuple) is True
+
+    # Inbound and outbound type discovery with FunctionsDataConverter hands declared types through
+    def sample_func(ctx, orders: dict[str, Order]) -> tuple[Order, ...]:
+        pass
+
+    assert type_discovery.orchestrator_input_type(sample_func, conv) == dict[str, Order]
+    assert type_discovery.activity_input_type(sample_func, conv) == dict[str, Order]
+    assert type_discovery.activity_output_type(sample_func, conv) == tuple[Order, ...]
+
+    def builtin_func(ctx, data: dict[str, int]) -> tuple[int, ...]:
+        pass
+
+    # Builtins are also passed through to df_loads under FunctionsDataConverter
+    assert type_discovery.orchestrator_input_type(builtin_func, conv) == dict[str, int]
+    assert type_discovery.activity_output_type(builtin_func, conv) == tuple[int, ...]
diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py
index ed286c79..f8e6ea92 100644
--- a/tests/durabletask/test_orchestration_executor.py
+++ b/tests/durabletask/test_orchestration_executor.py
@@ -864,6 +864,278 @@ def orchestrator(ctx: task.OrchestrationContext, args: StartArgs):
     assert captured["count"] == 5
 
 
+def test_old_history_replay_orchestrator_dict_and_tuple_input_reconstruction():
+    """Demonstrates that replaying an old history reconstructs dict and tuple inputs to typed dataclasses."""
+    from dataclasses import dataclass
+
+    @dataclass
+    class Order:
+        item: str
+        quantity: int
+
+    captured_dict: dict = {}
+    captured_tuple: dict = {}
+
+    def dict_orchestrator(ctx: task.OrchestrationContext, orders: dict[str, Order]):
+        captured_dict["orders"] = orders
+        return sum(o.quantity for o in orders.values())
+
+    def tuple_orchestrator(ctx: task.OrchestrationContext, orders: tuple[Order, ...]):
+        captured_tuple["orders"] = orders
+        return len(orders)
+
+    registry = worker._Registry()
+    dict_name = registry.add_orchestrator(dict_orchestrator)
+    tuple_name = registry.add_orchestrator(tuple_orchestrator)
+
+    # Replay dict input from old history
+    encoded_dict_input = json.dumps({"o1": {"item": "widget", "quantity": 3}, "o2": {"item": "gadget", "quantity": 7}})
+    old_events_dict = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(dict_name, TEST_INSTANCE_ID, encoded_input=encoded_dict_input)]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result_dict = executor.execute(TEST_INSTANCE_ID, [], old_events_dict)
+
+    complete_dict = get_and_validate_complete_orchestration_action_list(1, result_dict.actions)
+    assert complete_dict.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert complete_dict.result.value == "10"
+    dict_orders = captured_dict["orders"]
+    assert isinstance(dict_orders, dict)
+    assert isinstance(dict_orders["o1"], Order)
+    assert dict_orders["o1"].item == "widget"
+    assert dict_orders["o1"].quantity == 3
+    assert isinstance(dict_orders["o2"], Order)
+
+    # Replay tuple input from old history
+    encoded_tuple_input = json.dumps([{"item": "widget", "quantity": 1}, {"item": "gadget", "quantity": 2}])
+    old_events_tuple = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(tuple_name, TEST_INSTANCE_ID, encoded_input=encoded_tuple_input)]
+
+    result_tuple = executor.execute(TEST_INSTANCE_ID, [], old_events_tuple)
+
+    complete_tuple = get_and_validate_complete_orchestration_action_list(1, result_tuple.actions)
+    assert complete_tuple.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert complete_tuple.result.value == "2"
+    tuple_orders = captured_tuple["orders"]
+    assert isinstance(tuple_orders, tuple)
+    assert len(tuple_orders) == 2
+    assert isinstance(tuple_orders[0], Order)
+    assert tuple_orders[0].item == "widget"
+    assert isinstance(tuple_orders[1], Order)
+
+
+def test_old_history_replay_orchestrator_dict_input_nondeterminism_when_branching_on_raw_dict():
+    """Demonstrates replay incompatibility when an orchestration branched on uncoerced input dict.
+
+    Under the previous SDK, an input annotated as dict[str, Order] was passed as a raw dict of dicts.
+    An orchestration that branched on `isinstance(orders["o1"], dict)` previously scheduled "HandleRawDict".
+    Under the new SDK, orders["o1"] is reconstructed as Order, taking the typed branch and failing replay
+    with NonDeterminismError against the committed history.
+    """
+    from dataclasses import dataclass
+
+    @dataclass
+    class Order:
+        item: str
+        quantity: int
+
+    def orchestrator(ctx: task.OrchestrationContext, orders: dict[str, Order]):
+        if isinstance(orders["o1"], dict):
+            yield ctx.call_activity("HandleRawDict")
+        else:
+            yield ctx.call_activity("HandleTypedOrder")
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    # History recorded under previous SDK behavior: scheduled "HandleRawDict"
+    encoded_input = json.dumps({"o1": {"item": "widget", "quantity": 3}})
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=encoded_input),
+        helpers.new_task_scheduled_event(1, "HandleRawDict")]
+    new_events = [helpers.new_task_completed_event(1, json.dumps("done"))]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    complete_action = get_and_validate_complete_orchestration_action_list(1, result.actions)
+    assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert complete_action.failureDetails.errorType == "durabletask.task.NonDeterminismError"
+    assert "HandleRawDict" in complete_action.failureDetails.errorMessage
+    assert "HandleTypedOrder" in complete_action.failureDetails.errorMessage
+
+
+def test_old_history_replay_orchestrator_tuple_input_nondeterminism_when_branching_on_raw_element():
+    """Demonstrates replay incompatibility when an orchestration branched on uncoerced tuple input."""
+    from dataclasses import dataclass
+
+    @dataclass
+    class Order:
+        item: str
+
+    def orchestrator(ctx: task.OrchestrationContext, orders: tuple[Order, ...]):
+        if isinstance(orders[0], dict):
+            yield ctx.call_activity("HandleRawTuple")
+        else:
+            yield ctx.call_activity("HandleTypedTuple")
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    encoded_input = json.dumps([{"item": "widget"}])
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=encoded_input),
+        helpers.new_task_scheduled_event(1, "HandleRawTuple")]
+    new_events = [helpers.new_task_completed_event(1, json.dumps("done"))]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    complete_action = get_and_validate_complete_orchestration_action_list(1, result.actions)
+    assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert complete_action.failureDetails.errorType == "durabletask.task.NonDeterminismError"
+    assert "HandleRawTuple" in complete_action.failureDetails.errorMessage
+    assert "HandleTypedTuple" in complete_action.failureDetails.errorMessage
+
+
+def test_old_history_replay_inferred_activity_result_reconstruction():
+    """Demonstrates that replaying an old history reconstructs inferred activity return types."""
+    from dataclasses import dataclass
+
+    @dataclass
+    class Order:
+        item: str
+        quantity: int
+
+    def fetch_orders(ctx: task.ActivityContext, _) -> dict[str, Order]:
+        pass
+
+    captured: dict = {}
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        orders = yield ctx.call_activity(fetch_orders)
+        captured["orders"] = orders
+        return orders["o1"].item
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+        helpers.new_task_scheduled_event(1, task.get_name(fetch_orders))]
+    new_events = [
+        helpers.new_task_completed_event(1, json.dumps({"o1": {"item": "widget", "quantity": 3}}))]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    complete_action = get_and_validate_complete_orchestration_action_list(1, result.actions)
+    assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert complete_action.result.value == json.dumps("widget")
+    orders = captured["orders"]
+    assert isinstance(orders, dict)
+    assert isinstance(orders["o1"], Order)
+    assert orders["o1"].item == "widget"
+    assert orders["o1"].quantity == 3
+
+
+def test_old_history_replay_inferred_activity_result_nondeterminism_when_branching_on_raw_result():
+    """Demonstrates replay incompatibility when an orchestration branched on uncoerced activity results.
+
+    Under the previous SDK, return annotation `-> dict[str, Order]` was not reconstructable, so
+    call_activity(fetch_orders) returned a raw dict of dicts. An orchestration branching on
+    `isinstance(orders["o1"], dict)` scheduled "HandleRawActivityDict".
+    Replaying this history under the new SDK reconstructs Order, selects "HandleTypedActivityResult",
+    and fails with NonDeterminismError against the committed history.
+    """
+    from dataclasses import dataclass
+
+    @dataclass
+    class Order:
+        item: str
+
+    def fetch_orders(ctx: task.ActivityContext, _) -> dict[str, Order]:
+        pass
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        orders = yield ctx.call_activity(fetch_orders)
+        if isinstance(orders["o1"], dict):
+            yield ctx.call_activity("HandleRawActivityDict")
+        else:
+            yield ctx.call_activity("HandleTypedActivityResult")
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+        helpers.new_task_scheduled_event(1, task.get_name(fetch_orders)),
+        helpers.new_task_completed_event(1, json.dumps({"o1": {"item": "widget"}})),
+        helpers.new_task_scheduled_event(2, "HandleRawActivityDict")]
+    new_events = [helpers.new_task_completed_event(2, json.dumps("done"))]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    complete_action = get_and_validate_complete_orchestration_action_list(1, result.actions)
+    assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert complete_action.failureDetails.errorType == "durabletask.task.NonDeterminismError"
+    assert "HandleRawActivityDict" in complete_action.failureDetails.errorMessage
+    assert "HandleTypedActivityResult" in complete_action.failureDetails.errorMessage
+
+
+def test_unaffected_annotations_and_explicit_return_type_remain_unchanged():
+    """Confirms that builtins (dict[str, int]) remain uncoerced and explicit return_type overrides annotation."""
+    from dataclasses import dataclass
+
+    @dataclass
+    class Order:
+        item: str
+
+    def fetch_orders(ctx: task.ActivityContext, _) -> dict[str, Order]:
+        pass
+
+    captured: dict = {}
+
+    def orchestrator(ctx: task.OrchestrationContext, counts: dict[str, int]):
+        # Unaffected builtin: dict[str, int] is not reconstructable and remains raw dict
+        captured["counts"] = counts
+        # Explicit return_type=dict overrides discovered dict[str, Order] annotation
+        orders = yield ctx.call_activity(fetch_orders, return_type=dict)
+        captured["orders"] = orders
+        return "done"
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=json.dumps({"widgets": 5})),
+        helpers.new_task_scheduled_event(1, task.get_name(fetch_orders))]
+    new_events = [
+        helpers.new_task_completed_event(1, json.dumps({"o1": {"item": "gadget"}}))]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    complete_action = get_and_validate_complete_orchestration_action_list(1, result.actions)
+    assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+
+    assert captured["counts"] == {"widgets": 5}
+    assert isinstance(captured["counts"], dict)
+
+    assert captured["orders"] == {"o1": {"item": "gadget"}}
+    assert isinstance(captured["orders"], dict)
+    assert isinstance(captured["orders"]["o1"], dict)
+    assert not isinstance(captured["orders"]["o1"], Order)
+
+
 def test_activity_task_failed():
     """Tests the failure of an activity task"""
     def dummy_activity(ctx, _):
diff --git a/tests/durabletask/test_type_discovery.py b/tests/durabletask/test_type_discovery.py
index 63f8cbc7..07fbe8f6 100644
--- a/tests/durabletask/test_type_discovery.py
+++ b/tests/durabletask/test_type_discovery.py
@@ -6,8 +6,9 @@
 import inspect
 import json
 import logging
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Optional
+from typing import Any, Optional, get_origin
 from unittest.mock import patch
 
 import pytest
@@ -72,6 +73,23 @@ def test_list_of_dataclass_is_reconstructable(self):
     def test_list_of_builtin_is_not_reconstructable(self):
         assert self.conv.can_reconstruct(list[int]) is False
 
+    def test_dict_and_mapping_of_dataclass_is_reconstructable(self):
+        assert self.conv.can_reconstruct(dict[str, Order]) is True
+        assert self.conv.can_reconstruct(Mapping[str, Order]) is True
+        assert self.conv.can_reconstruct(dict[str, int]) is False
+
+    def test_tuple_of_dataclass_is_reconstructable(self):
+        assert self.conv.can_reconstruct(tuple[Order, ...]) is True
+        assert self.conv.can_reconstruct(tuple[Order, int]) is True
+        assert self.conv.can_reconstruct(tuple[int, str]) is False
+        assert self.conv.can_reconstruct(tuple) is False
+
+    def test_nested_container_reconstructability(self):
+        assert self.conv.can_reconstruct(dict[str, list[Order]]) is True
+        assert self.conv.can_reconstruct(tuple[dict[str, Order], ...]) is True
+        assert self.conv.can_reconstruct(list[dict[str, Order]]) is True
+        assert self.conv.can_reconstruct(dict[str, list[int]]) is False
+
 
 class TestCustomConverterReconstructable:
     """A custom converter can extend reconstructability to its own types, and the
@@ -95,6 +113,48 @@ def can_reconstruct(self, target_type: Any) -> bool:
         assert conv.can_reconstruct(Optional[Widget]) is True
         # Builtins remain excluded.
         assert conv.can_reconstruct(int) is False
+        # The override is also consulted for elements inside dict and tuple hints.
+        assert conv.can_reconstruct(dict[str, Widget]) is True
+        assert conv.can_reconstruct(tuple[Widget, ...]) is True
+        assert conv.can_reconstruct(tuple[Widget, int]) is True
+
+    def test_override_can_enforce_custom_container_policy(self):
+        class SelectiveConverter(JsonDataConverter):
+            def can_reconstruct(self, target_type: Any) -> bool:
+                # Custom policy: reject all tuple hints even with supported elements
+                if get_origin(target_type) is tuple:
+                    return False
+                return super().can_reconstruct(target_type)
+
+        conv = SelectiveConverter()
+        assert conv.can_reconstruct(dict[str, Order]) is True
+        assert conv.can_reconstruct(tuple[Order, ...]) is False
+        assert conv.can_reconstruct(tuple[Order, int]) is False
+
+        def tuple_activity(ctx, items: tuple[Order, ...]) -> tuple[Order, ...]:
+            pass
+
+        assert type_discovery.activity_input_type(tuple_activity, conv) is None
+        assert type_discovery.activity_output_type(tuple_activity, conv) is None
+
+    def test_permissive_converter_discovery_policy_unaffected(self):
+        """A converter that overrides can_reconstruct to return True (like Azure Functions)
+        remains unaffected and continues to surface all container annotations."""
+        class PermissiveConverter(JsonDataConverter):
+            def can_reconstruct(self, target_type: Any) -> bool:
+                return True
+
+        conv = PermissiveConverter()
+        assert conv.can_reconstruct(dict[str, Order]) is True
+        assert conv.can_reconstruct(tuple[Order, ...]) is True
+        assert conv.can_reconstruct(dict[str, int]) is True
+        assert conv.can_reconstruct(tuple[int, str]) is True
+
+        def sample_act(ctx, d: dict[str, int]) -> tuple[int, ...]:
+            pass
+
+        assert type_discovery.activity_input_type(sample_act, conv) == dict[str, int]
+        assert type_discovery.activity_output_type(sample_act, conv) == tuple[int, ...]
 
     def test_discovery_uses_supplied_converter(self):
         class Widget:
@@ -414,6 +474,47 @@ def handle(ctx: task.ActivityContext, money: Money):
     assert json.loads(result) == 50
 
 
+def test_activity_input_coerced_for_dict_of_dataclass():
+    seen: dict[str, Any] = {}
+
+    def handle(ctx: task.ActivityContext, orders: dict[str, Order]):
+        seen["orders"] = orders
+        return sum(o.quantity for o in orders.values())
+
+    executor, name = _activity_executor(handle)
+    payload = json.dumps({"o1": {"item": "widget", "quantity": 3}, "o2": {"item": "gadget", "quantity": 7}})
+    result = executor.execute("orch1", name, 1, payload)
+
+    orders = seen["orders"]
+    assert isinstance(orders, dict)
+    assert isinstance(orders["o1"], Order)
+    assert isinstance(orders["o2"], Order)
+    assert orders["o1"].item == "widget"
+    assert orders["o2"].quantity == 7
+    assert json.loads(result) == 10
+
+
+def test_activity_input_coerced_for_tuple_of_dataclass():
+    seen: dict[str, Any] = {}
+
+    def handle(ctx: task.ActivityContext, orders: tuple[Order, ...]):
+        seen["orders"] = orders
+        return len(orders)
+
+    executor, name = _activity_executor(handle)
+    payload = json.dumps([{"item": "widget", "quantity": 3}, {"item": "gadget", "quantity": 7}])
+    result = executor.execute("orch1", name, 1, payload)
+
+    orders = seen["orders"]
+    assert isinstance(orders, tuple)
+    assert len(orders) == 2
+    assert isinstance(orders[0], Order)
+    assert isinstance(orders[1], Order)
+    assert orders[0].item == "widget"
+    assert orders[1].quantity == 7
+    assert json.loads(result) == 2
+
+
 def test_activity_builtin_input_unchanged():
     seen: dict[str, Any] = {}