@@ -14,9 +14,11 @@ payloads using a shared payload store, including Azure Blob Storage. Configurati
 applies to synchronous and asynchronous clients, orchestration and activity
 inputs and outputs, custom status, external events, entity state and results,
 and registered blueprints. History APIs also hydrate entity operation inputs
-and results. Registered orchestration, entity, and activity handlers await
-asynchronous payload storage without occupying synchronous execution threads.
-User orchestrators remain generators, synchronous functions retain synchronous
+and results. Registered orchestration and entity handlers await asynchronous
+payload storage without occupying synchronous execution threads. Replay does
+not download unused nested inputs from historical entity requests.
+User orchestrators remain generators, and activities retain their synchronous
+or asynchronous calling convention. Synchronous functions retain synchronous
 client bindings, and invocation logging context is preserved.
 Worker payload-size and storage errors retain their original error details.
 
 
@@ -97,23 +97,31 @@ agree on the store and reference encoding; Functions references are JSON strings
 > host abandons and redelivers a work item after a storage failure. A transient
 > storage error can therefore become a terminal orchestration failure.
 
-Registered orchestration, entity, and activity handlers await the store's async
-methods before and after execution, including for synchronous user functions.
-Orchestrators remain synchronous generators; entities and synchronous activities
-run on execution threads with their invocation logging context preserved.
-Payload downloads and uploads do not occupy those threads, and serialization
-does not access storage during replay. Custom stores must implement genuinely
-nonblocking async methods to benefit from this behavior.
-
-The SDK reuses the Functions runtime's thread pool when the runtime exposes it;
-otherwise it uses a process-wide SDK pool. Both honor
-`PYTHON_THREADPOOL_THREAD_COUNT` for synchronous execution. Synchronous user
-functions still receive the synchronous durable client, and synchronous client
-APIs still use synchronous storage. Both client history APIs hydrate entity
-operation inputs and results, including values nested in the host's entity
-protocol envelopes. Direct `Orchestrator.handle()` and `Orchestrator.create()`
+Registered orchestration and entity handlers await the store's async methods
+before and after execution. Orchestrators remain synchronous generators, and
+orchestration replay and entity code run on execution threads with their
+invocation logging context preserved. Their payload downloads and uploads do
+not occupy those threads, and serialization does not access storage during
+replay. Custom stores must implement genuinely nonblocking async methods to
+benefit from this behavior.
+
+For orchestration and entity execution, the SDK reuses the Functions runtime's
+thread pool when the runtime exposes it; otherwise it uses a process-wide SDK
+pool. Both honor `PYTHON_THREADPOOL_THREAD_COUNT`.
+
+Activities retain their synchronous or asynchronous calling convention.
+Synchronous activities use synchronous storage inside the host-managed execution
+thread and remain directly callable without `await`; async activities await
+async storage. Binding converters perform no storage I/O. Synchronous functions
+still receive the synchronous durable client, and synchronous client APIs use
+synchronous storage. Direct `Orchestrator.handle()` and `Orchestrator.create()`
 adapters also remain synchronous.
 
+Both client history APIs hydrate entity operation inputs and results, including
+values nested in the host's entity protocol envelopes. During orchestration
+replay, nested entity results are hydrated, but historical nested request inputs
+are not downloaded because replay only needs their correlation metadata.
+
 > [!WARNING]
 > With payload storage configured, whole payload strings recognized by the
 > store's `is_known_token()` are reserved references, not literal application
 
@@ -326,9 +326,8 @@ def activity_trigger(self, input_name: str,
         """
         @self._build_function
         def wrap(fb: FunctionBuilder) -> FunctionBuilder:
-            registered = fb._function._func  # pyright: ignore[reportPrivateUsage]
             fb.add_trigger(
-                trigger=ActivityTrigger(name=getattr(registered, "_df_trigger_name", input_name), activity=activity))
+                trigger=ActivityTrigger(name=input_name, activity=activity))
             return fb
 
         def decorator(user_fn: Callable[..., Any]) -> FunctionBuilder:
@@ -337,8 +336,7 @@ def decorator(user_fn: Callable[..., Any]) -> FunctionBuilder:
             # pass through unchanged.
             function = (user_fn._function._func  # pyright: ignore[reportPrivateUsage]
                         if isinstance(user_fn, FunctionBuilder) else user_fn)
-            registered = wrap_invocation(
-                wrap_activity_payloads(wrap_activity(function, input_name), input_name), input_name)
+            registered = wrap_activity_payloads(wrap_activity(function, input_name), input_name)
             if isinstance(user_fn, FunctionBuilder):
                 user_fn._function._func = registered  # pyright: ignore[reportPrivateUsage]
                 return wrap(user_fn)
@@ -425,7 +423,6 @@ def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder:
                         if isinstance(user_fn, FunctionBuilder) else user_fn)
             signature = inspect.signature(function)
             is_async_function = inspect.iscoroutinefunction(function)
-            is_async_user = getattr(function, "_df_user_is_async", is_async_function)
 
             def bind_client(
                     args: tuple[Any, ...],
@@ -440,7 +437,7 @@ def bind_client(
                 if not isinstance(raw_client, str):
                     raise TypeError(
                         f"durable client binding '{client_name}' did not provide its configuration")
-                client = (DurableFunctionsClient(raw_client) if is_async_user
+                client = (DurableFunctionsClient(raw_client) if is_async_function
                           else SyncDurableFunctionsClient.get_cached(raw_client))
                 bound.arguments[client_name] = client
                 return bound, client
 
@@ -33,10 +33,11 @@
 from azure.functions._durable_functions import df_loads
 
 from ..converters import ActivityTriggerConverter
-from ..invocation import run_sync
 from ..payloads import (
     ActivityPayload,
+    deexternalize_payload,
     deexternalize_payload_async,
+    externalize_activity_output,
     externalize_activity_output_async,
     get_payload_store,
 )
@@ -177,30 +178,48 @@ def wrap_activity(fn: Callable[..., Any], input_name: str) -> Callable[..., Any]
 
 
 def wrap_activity_payloads(fn: Callable[..., Any], input_name: str) -> Callable[..., Any]:
-    """Await payload transport and offload synchronous user activities."""
+    """Run payload I/O inside the invocation, preserving sync/async dispatch."""
     signature = inspect.signature(fn)
-    is_async = inspect.iscoroutinefunction(fn)
 
     def decode(value: str) -> Any:
         try:
             return df_loads(value)
         except json.JSONDecodeError:
             return value
 
-    @wraps(fn)
-    async def wrapper(*args: Any, **kwargs: Any) -> Any:
-        enabled = get_payload_store() is not None
-        bound = signature.bind(*args, **kwargs)
-        value = bound.arguments.get(input_name)
-        if enabled and isinstance(value, ActivityPayload):
-            bound.arguments[input_name] = decode(await deexternalize_payload_async(value.value))
-        result = (await fn(*bound.args, **bound.kwargs) if is_async
-                  else await run_sync(fn, *bound.args, **bound.kwargs))
-        if not enabled or result is None:
-            return result
-        encoded = ActivityTriggerConverter.encode(result, expected_type=None).value
-        return ActivityPayload(await externalize_activity_output_async(encoded))
+    wrapper: Callable[..., Any]
+    if inspect.iscoroutinefunction(fn):
+        @wraps(fn)
+        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+            if get_payload_store() is None:
+                return await fn(*args, **kwargs)
+            bound = signature.bind(*args, **kwargs)
+            value = bound.arguments.get(input_name)
+            if isinstance(value, ActivityPayload):
+                bound.arguments[input_name] = decode(await deexternalize_payload_async(value.value))
+            result = await fn(*bound.args, **bound.kwargs)
+            if result is None:
+                return None
+            encoded = ActivityTriggerConverter.encode(result, expected_type=None).value
+            return ActivityPayload(await externalize_activity_output_async(encoded))
+
+        wrapper = async_wrapper
+    else:
+        @wraps(fn)
+        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
+            if get_payload_store() is None:
+                return fn(*args, **kwargs)
+            bound = signature.bind(*args, **kwargs)
+            value = bound.arguments.get(input_name)
+            if isinstance(value, ActivityPayload):
+                bound.arguments[input_name] = decode(deexternalize_payload(value.value))
+            result = fn(*bound.args, **bound.kwargs)
+            if result is None:
+                return None
+            encoded = ActivityTriggerConverter.encode(result, expected_type=None).value
+            return ActivityPayload(externalize_activity_output(encoded))
+
+        wrapper = sync_wrapper
 
     setattr(wrapper, "__signature__", signature)
-    setattr(wrapper, "_df_user_is_async", getattr(fn, "_df_user_is_async", is_async))
     return wrapper
@@ -137,7 +137,7 @@ def externalize_activity_output(value: str) -> str:
 
 
 def _entity_payload_fields(
-        events: Sequence[history.HistoryEvent], instance_id: str,
+        events: Sequence[history.HistoryEvent], instance_id: str, *, include_inputs: bool,
 ) -> Iterator[tuple[history.EventSentEvent | history.EventRaisedEvent, dict[str, Any], str]]:
     pending: set[str] = set()
     for event in events:
@@ -182,17 +182,18 @@ def _entity_payload_fields(
                 pending.add(request_id)
             elif envelope.get("signal") is not True:
                 continue
-        if isinstance(envelope.get(field), str):
+        if (include_inputs or field == "result") and isinstance(envelope.get(field), str):
             yield event, envelope, field
 
 
 def hydrate_entity_history(
         events: list[history.HistoryEvent], store: PayloadStore | None, instance_id: str,
+        *, include_inputs: bool = True,
 ) -> None:
     """Hydrate serialized entity protocol fields, never arbitrary object members."""
     if store is None:
         return
-    for event, envelope, field in _entity_payload_fields(events, instance_id):
+    for event, envelope, field in _entity_payload_fields(events, instance_id, include_inputs=include_inputs):
         request = ActivityRequest(input=StringValue(value=envelope[field]))
         deexternalize_payloads(request, store)
         if request.input.value != envelope[field]:
@@ -202,11 +203,12 @@ def hydrate_entity_history(
 
 async def hydrate_entity_history_async(
         events: list[history.HistoryEvent], store: PayloadStore | None, instance_id: str,
+        *, include_inputs: bool = True,
 ) -> None:
     """Hydrate entity history using the store's asynchronous download API."""
     if store is None:
         return
-    for event, envelope, field in _entity_payload_fields(events, instance_id):
+    for event, envelope, field in _entity_payload_fields(events, instance_id, include_inputs=include_inputs):
         request = ActivityRequest(input=StringValue(value=envelope[field]))
         await deexternalize_payloads_async(request, store)
         if request.input.value != envelope[field]:
@@ -224,19 +226,19 @@ def _entity_request_events(request: OrchestratorRequest) -> list[tuple[HistoryEv
 
 def _update_entity_request_events(events: list[tuple[HistoryEvent, history.HistoryEvent]]) -> None:
     for source, event in events:
-        if isinstance(event, history.EventSentEvent) and event.input is not None:
-            source.eventSent.input.value = event.input
-        elif isinstance(event, history.EventRaisedEvent) and event.input is not None:
+        if isinstance(event, history.EventRaisedEvent) and event.input is not None:
             source.eventRaised.input.value = event.input
 
 
 def hydrate_entity_request(request: OrchestratorRequest, store: PayloadStore) -> None:
+    """Hydrate entity replies for replay, leaving unused historical inputs alone."""
     events = _entity_request_events(request)
-    hydrate_entity_history([event for _, event in events], store, request.instanceId)
+    hydrate_entity_history([event for _, event in events], store, request.instanceId, include_inputs=False)
     _update_entity_request_events(events)
 
 
 async def hydrate_entity_request_async(request: OrchestratorRequest, store: PayloadStore) -> None:
+    """Await entity reply payloads before replay, without downloading historical inputs."""
     events = _entity_request_events(request)
-    await hydrate_entity_history_async([event for _, event in events], store, request.instanceId)
+    await hydrate_entity_history_async([event for _, event in events], store, request.instanceId, include_inputs=False)
     _update_entity_request_events(events)
@@ -177,8 +177,7 @@ async def native_activity(context, payload):
     store.upload_async.assert_awaited_once()
 
 
-@pytest.mark.asyncio
-async def test_sync_activity_awaits_storage_and_offloads_user_code(monkeypatch, payload_store_factory):
+def test_sync_activity_uses_sync_storage_on_the_calling_thread(monkeypatch, payload_store_factory):
     store = payload_store_factory()
     monkeypatch.setattr(payloads, "_payload_store", store)
     value = {"large": "x" * 200}
@@ -187,11 +186,11 @@ async def test_sync_activity_awaits_storage_and_offloads_user_code(monkeypatch,
     original_download = store.download
     original_upload = store.upload
 
-    async def download(reference):
+    def download(reference):
         thread_ids.append(threading.get_ident())
         return original_download(reference)
 
-    async def upload(data, *, instance_id=None):
+    def upload(data, *, instance_id=None):
         thread_ids.append(threading.get_ident())
         return original_upload(data, instance_id=instance_id)
 
@@ -200,18 +199,22 @@ def activity(payload):
         assert payload == value
         return payload
 
-    monkeypatch.setattr(store, "download_async", download)
-    monkeypatch.setattr(store, "upload_async", upload)
-    monkeypatch.setattr(store, "download", Mock(side_effect=AssertionError("sync download")))
-    monkeypatch.setattr(store, "upload", Mock(side_effect=AssertionError("sync upload")))
+    monkeypatch.setattr(store, "download", Mock(side_effect=download))
+    monkeypatch.setattr(store, "upload", Mock(side_effect=upload))
+    monkeypatch.setattr(store, "download_async", AsyncMock(side_effect=AssertionError("async download")))
+    monkeypatch.setattr(store, "upload_async", AsyncMock(side_effect=AssertionError("async upload")))
     wrapper = wrap_activity_payloads(activity, "payload")
-    assert inspect.iscoroutinefunction(wrapper)
+    assert not inspect.iscoroutinefunction(wrapper)
     decoded = ActivityTriggerConverter.decode(
         meta.Datum(type="string", value=token), trigger_metadata=None)
-    await wrapper(decoded)
-    assert len(thread_ids) == 3
-    assert thread_ids[0] == thread_ids[2] == threading.get_ident()
-    assert thread_ids[1] != threading.get_ident()
+    result = wrapper(decoded)
+    encoded = ActivityTriggerConverter.encode(result, expected_type=None)
+    assert json.loads(original_download(json.loads(encoded.value))) == value
+    assert thread_ids == [threading.get_ident()] * 3
+    store.download.assert_called_once_with(token)
+    store.upload.assert_called_once()
+    store.download_async.assert_not_called()
+    store.upload_async.assert_not_called()
 
 
 @pytest.mark.asyncio
@@ -223,8 +226,8 @@ async def test_activity_storage_errors_propagate_without_retry(
     monkeypatch.setattr(payloads, "_payload_store", store)
     token = store.upload(json.dumps("x" * 200).encode())
     error = OSError("payload storage unavailable")
-    failure = AsyncMock(side_effect=error)
-    monkeypatch.setattr(store, operation + "_async", failure)
+    failure = AsyncMock(side_effect=error) if use_async else Mock(side_effect=error)
+    monkeypatch.setattr(store, operation + ("_async" if use_async else ""), failure)
     called = []
 
     def activity(payload):
@@ -238,10 +241,16 @@ async def async_activity(payload):
     decoded = ActivityTriggerConverter.decode(
         meta.Datum(type="string", value=token), trigger_metadata=None)
     with pytest.raises(OSError) as raised:
-        await wrapper(decoded)
+        if use_async:
+            await wrapper(decoded)
+        else:
+            wrapper(decoded)
     assert raised.value is error
     assert called == ([True] if operation == "upload" else [])
-    failure.assert_awaited_once()
+    if use_async:
+        failure.assert_awaited_once()
+    else:
+        failure.assert_called_once()
 
 
 @pytest.mark.parametrize("setting, expected", [(None, None), ("2", 2), ("invalid", None), ("0", None)])
@@ -289,42 +298,3 @@ async def handle(payload):
             assert await loop.run_in_executor(executor, host_invocation_id.get) is None
             assert await loop.run_in_executor(executor, getattr, storage, "invocation_id") == "previous"
             assert invocation._invocation_context.get() is None
-
-
-async def test_storage_downloads_do_not_wait_for_execution_thread(monkeypatch, payload_store_factory):
-    store = payload_store_factory()
-    monkeypatch.setattr(payloads, "_payload_store", store)
-    token = store.upload(b'"input"')
-    loop = asyncio.get_running_loop()
-    executing = asyncio.Event()
-    downloaded = asyncio.Event()
-    release = threading.Event()
-    downloads = []
-
-    async def download(reference):
-        downloads.append(reference)
-        if len(downloads) == 2:
-            downloaded.set()
-        return store._blobs[reference]
-
-    def activity(payload):
-        loop.call_soon_threadsafe(executing.set)
-        assert release.wait(timeout=5)
-        return payload
-
-    monkeypatch.setattr(store, "download_async", download)
-    monkeypatch.setattr(store, "download", Mock(side_effect=AssertionError("sync download")))
-    wrapper = wrap_activity_payloads(activity, "payload")
-    with ThreadPoolExecutor(max_workers=1) as executor:
-        monkeypatch.setattr(invocation, "_executor", lambda: executor)
-        first = asyncio.create_task(wrapper(payloads.ActivityPayload(token)))
-        second = None
-        try:
-            await asyncio.wait_for(executing.wait(), timeout=5)
-            second = asyncio.create_task(wrapper(payloads.ActivityPayload(token)))
-            await asyncio.wait_for(downloaded.wait(), timeout=5)
-            assert not first.done()
-            assert not second.done()
-        finally:
-            release.set()
-            await asyncio.gather(first, *([second] if second is not None else []))