diff --git a/CHANGELOG.md b/CHANGELOG.md
index b5f0a94ec..c09e64315 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -82,6 +82,7 @@ to include examples, links to docs, or any other relevant information.
   context is now detached exactly when its token is still valid in the current
   `contextvars.Context`, which it stays when a workflow resumes on another pool thread.
 
+- Clarify activity and child workflow return-value conversion errors without hiding their original causes.
 - Decode enums that mix in `str` or `int` (for example `class Color(str, Enum)`) to the enum
   instead of a list of characters or an error, and decode `dict` keys typed as an `int` enum.
 
diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py
index 240385690..d66826e20 100644
--- a/temporalio/worker/_activity.py
+++ b/temporalio/worker/_activity.py
@@ -362,7 +362,17 @@ async def _handle_start_activity_task(
             result = await self._execute_activity(
                 start, running_activity, task_token, data_converter
             )
-            [payload] = await data_converter.encode([result])
+            try:
+                [payload] = await data_converter.encode([result])
+            except concurrent.futures.BrokenExecutor:
+                # The handler below fails the whole worker on a broken executor,
+                # so it has to see the original type rather than a re-label.
+                raise
+            except Exception as err:
+                raise temporalio.exceptions.ApplicationError(
+                    f"Failed to encode return value of activity {start.activity_type}",
+                    type=type(err).__name__,
+                ) from err
             completion.result.completed.result.CopyFrom(payload)
         except BaseException as err:
             try:
diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py
index 0fefc8fa0..f6fea870a 100644
--- a/temporalio/worker/_workflow_instance.py
+++ b/temporalio/worker/_workflow_instance.py
@@ -904,9 +904,10 @@ def _apply_resolve_activity(
             if job.result.completed.HasField("result"):
                 ret_types = [handle._input.ret_type] if handle._input.ret_type else None
                 ret_vals = self._convert_payloads(
-                    [job.result.completed.result],
-                    ret_types,
-                    payload_converter,
+                    payloads=[job.result.completed.result],
+                    types=ret_types,
+                    payload_converter=payload_converter,
+                    failure_message=f"Failed to decode return value of activity {handle._input.activity}",
                 )
                 ret = ret_vals[0]
             handle._resolve_success(ret)
@@ -942,9 +943,10 @@ def _apply_resolve_child_workflow_execution(
             if job.result.completed.HasField("result"):
                 ret_types = [handle._input.ret_type] if handle._input.ret_type else None
                 ret_vals = self._convert_payloads(
-                    [job.result.completed.result],
-                    ret_types,
-                    handle._payload_converter,
+                    payloads=[job.result.completed.result],
+                    types=ret_types,
+                    payload_converter=handle._payload_converter,
+                    failure_message=f"Failed to decode return value of child workflow {handle._input.workflow}",
                 )
                 ret = ret_vals[0]
             handle._resolve_success(ret)
@@ -1157,9 +1159,20 @@ def _apply_initialize_workflow(
         async def run_workflow(input: ExecuteWorkflowInput) -> None:
             try:
                 result = await self._inbound.execute_workflow(input)
-                result_payloads = self._workflow_context_payload_converter.to_payloads(
-                    [result]
-                )
+                try:
+                    result_payloads = (
+                        self._workflow_context_payload_converter.to_payloads([result])
+                    )
+                except Exception as err:
+                    error = RuntimeError(
+                        f"Failed to encode return value of workflow {self._info.workflow_type}"
+                    )
+                    # Wrapping must not change whether the workflow or its task fails.
+                    if self.workflow_is_failure_exception(
+                        err
+                    ) or self.workflow_is_failure_exception(error):
+                        raise
+                    raise error from err
                 if len(result_payloads) != 1:
                     raise ValueError(
                         f"Expected 1 result payload, got {len(result_payloads)}"
@@ -2407,6 +2420,8 @@ def _convert_payloads(
         payloads: Sequence[temporalio.api.common.v1.Payload],
         types: list[type] | None,
         payload_converter: temporalio.converter.PayloadConverter,
+        *,
+        failure_message: str = "Failed decoding arguments",
     ) -> list[Any]:
         if not payloads:
             return []
@@ -2421,7 +2436,7 @@ def _convert_payloads(
         except Exception as err:
             if self.workflow_is_failure_exception(err):
                 raise
-            raise RuntimeError("Failed decoding arguments") from err
+            raise RuntimeError(failure_message) from err
 
     def _payload_converter_with_context(
         self,
diff --git a/tests/worker/test_activity.py b/tests/worker/test_activity.py
index 64691a93f..a068ac176 100644
--- a/tests/worker/test_activity.py
+++ b/tests/worker/test_activity.py
@@ -30,6 +30,7 @@
     WorkflowHandle,
 )
 from temporalio.common import RawValue, RetryPolicy
+from temporalio.converter import DataConverter, DefaultPayloadConverter
 from temporalio.exceptions import (
     ActivityError,
     ApplicationError,
@@ -1242,6 +1243,80 @@ async def test_sync_activity_process_executor_crash(
         assert isinstance(worker_err.value.__cause__, BrokenProcessPool)
 
 
+@dataclass
+class UnserializableResult:
+    pass
+
+
+def _converter_raising_on_result(
+    error: BaseException,
+) -> type[DefaultPayloadConverter]:
+    class RaiseOnResultConverter(DefaultPayloadConverter):
+        def to_payloads(
+            self, values: Sequence[Any]
+        ) -> list[temporalio.api.common.v1.Payload]:
+            if any(isinstance(value, UnserializableResult) for value in values):
+                raise error
+            return super().to_payloads(values)
+
+    return RaiseOnResultConverter
+
+
+def _client_failing_result_encoding(client: Client, error: BaseException) -> Client:
+    return Client(
+        client.service_client,
+        namespace=client.namespace,
+        data_converter=DataConverter(
+            payload_converter_class=_converter_raising_on_result(error)
+        ),
+    )
+
+
+async def test_result_encoding_broken_executor_fails_worker(
+    client: Client, worker: ExternalWorker
+):
+    """A broken executor during encoding stays worker-fatal, as it is during run."""
+
+    @activity.defn
+    async def unserializable_result() -> UnserializableResult:
+        return UnserializableResult()
+
+    act_task_queue = str(uuid.uuid4())
+    act_client = _client_failing_result_encoding(
+        client, concurrent.futures.BrokenExecutor("pool is broken")
+    )
+    act_worker = Worker(
+        act_client, task_queue=act_task_queue, activities=[unserializable_result]
+    )
+    act_worker_task = asyncio.create_task(act_worker.run())
+    try:
+        with pytest.raises(WorkflowFailureError) as err:
+            await client.execute_workflow(
+                "kitchen_sink",
+                KSWorkflowParams(
+                    actions=[
+                        KSAction(
+                            execute_activity=KSExecuteActivityAction(
+                                name="unserializable_result", task_queue=act_task_queue
+                            )
+                        )
+                    ]
+                ),
+                id=str(uuid.uuid4()),
+                task_queue=worker.task_queue,
+            )
+        assert isinstance(err.value.cause, ActivityError)
+        assert isinstance(err.value.cause.cause, ApplicationError)
+        assert err.value.cause.cause.type == "BrokenExecutor"
+
+        with pytest.raises(RuntimeError) as worker_err:
+            await asyncio.wait_for(act_worker_task, 10)
+        assert str(worker_err.value) == "Activity worker failed"
+        assert isinstance(worker_err.value.__cause__, concurrent.futures.BrokenExecutor)
+    finally:
+        await act_worker.shutdown()
+
+
 class AsyncActivityWrapper:
     def __init__(self) -> None:
         self._info: activity.Info | None = None
diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py
index 909202fc4..dd1aebcae 100644
--- a/tests/worker/test_workflow.py
+++ b/tests/worker/test_workflow.py
@@ -4824,6 +4824,178 @@ async def test_workflow_activity_outbound_conversion_failure(client: Client):
         assert "Intentional outbound converter failure" in str(err.value.cause)
 
 
+@activity.defn
+async def result_conversion_activity(encode: bool) -> Any:
+    return object() if encode else "not an int"
+
+
+@workflow.defn
+class ResultConversionChildWorkflow:
+    @workflow.run
+    async def run(self, encode: bool) -> Any:
+        return object() if encode else "not an int"
+
+
+@workflow.defn
+class ResultConversionWorkflow:
+    @workflow.run
+    async def run(self, source: str, encode: bool) -> None:
+        if source == "child_workflow":
+            await workflow.execute_child_workflow(
+                workflow="ResultConversionChildWorkflow",
+                arg=encode,
+                id=f"{workflow.info().workflow_id}-child",
+                result_type=int,
+            )
+        else:
+            execute_activity = (
+                workflow.execute_local_activity
+                if source == "local_activity"
+                else workflow.execute_activity
+            )
+            await execute_activity(
+                activity="result_conversion_activity",
+                arg=encode,
+                result_type=int,
+                start_to_close_timeout=timedelta(seconds=10),
+                retry_policy=RetryPolicy(maximum_attempts=1),
+            )
+
+
+@pytest.mark.parametrize(
+    "source,target",
+    [
+        ("activity", "activity result_conversion_activity"),
+        ("local_activity", "activity result_conversion_activity"),
+        ("child_workflow", "child workflow ResultConversionChildWorkflow"),
+    ],
+)
+async def test_result_decoding_failure(
+    client: Client, source: str, target: str
+) -> None:
+    async with new_worker(
+        client,
+        ResultConversionWorkflow,
+        ResultConversionChildWorkflow,
+        activities=[result_conversion_activity],
+    ) as worker:
+        handle = await client.start_workflow(
+            workflow=ResultConversionWorkflow.run,
+            args=[source, False],
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        try:
+            await assert_task_fail_eventually(handle)
+            history = await handle.fetch_history()
+            failure = next(
+                event.workflow_task_failed_event_attributes.failure
+                for event in history.events
+                if event.HasField("workflow_task_failed_event_attributes")
+            )
+            assert failure.message == f"Failed to decode return value of {target}"
+            assert failure.application_failure_info.type == "RuntimeError"
+            assert failure.cause.application_failure_info.type == "TypeError"
+            assert (
+                failure.cause.message
+                == "Expected value to be int|float, was <class 'str'>"
+            )
+        finally:
+            await handle.terminate()
+
+
+@pytest.mark.parametrize("source", ["activity", "local_activity", "child_workflow"])
+async def test_result_encoding_failure(client: Client, source: str) -> None:
+    async with new_worker(
+        client,
+        ResultConversionWorkflow,
+        ResultConversionChildWorkflow,
+        activities=[result_conversion_activity],
+    ) as worker:
+        handle = await client.start_workflow(
+            workflow=ResultConversionWorkflow.run,
+            args=[source, True],
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        failure: Failure
+        if source == "child_workflow":
+            try:
+                child_handle = await assert_workflow_exists_eventually(
+                    client=client,
+                    workflow=ResultConversionChildWorkflow.run,
+                    workflow_id=f"{handle.id}-child",
+                )
+                await assert_task_fail_eventually(child_handle)
+                history = await child_handle.fetch_history()
+                failure = next(
+                    event.workflow_task_failed_event_attributes.failure
+                    for event in history.events
+                    if event.HasField("workflow_task_failed_event_attributes")
+                )
+                assert (
+                    failure.message
+                    == "Failed to encode return value of workflow ResultConversionChildWorkflow"
+                )
+            finally:
+                await handle.terminate()
+        else:
+            with pytest.raises(WorkflowFailureError) as err:
+                await handle.result()
+            cause: BaseException | None = err.value.cause
+            if source == "activity":
+                assert isinstance(cause, ActivityError)
+                cause = cause.cause
+            assert isinstance(cause, ApplicationError)
+            assert cause.failure
+            failure = cause.failure
+            assert (
+                failure.message
+                == "Failed to encode return value of activity result_conversion_activity"
+            )
+            assert failure.application_failure_info.type == "TypeError"
+        assert failure.cause.application_failure_info.type == "TypeError"
+        assert failure.cause.message == "Object of type object is not JSON serializable"
+
+
+@pytest.mark.parametrize("failure_type", [TypeError, RuntimeError])
+async def test_result_encoding_failure_exception_types(
+    client: Client, failure_type: type[Exception]
+) -> None:
+    async with new_worker(
+        client,
+        ResultConversionChildWorkflow,
+        workflow_failure_exception_types=[failure_type],
+    ) as worker:
+        handle = await client.start_workflow(
+            workflow=ResultConversionChildWorkflow.run,
+            arg=True,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        if failure_type is TypeError:
+            with pytest.raises(WorkflowFailureError) as err:
+                await handle.result()
+            assert isinstance(err.value.cause, ApplicationError)
+            assert err.value.cause.type == "TypeError"
+            assert (
+                err.value.cause.message
+                == "Object of type object is not JSON serializable"
+            )
+        else:
+            try:
+                await assert_task_fail_eventually(handle)
+                history = await handle.fetch_history()
+                failure = next(
+                    event.workflow_task_failed_event_attributes.failure
+                    for event in history.events
+                    if event.HasField("workflow_task_failed_event_attributes")
+                )
+                assert failure.application_failure_info.type == "TypeError"
+            finally:
+                await handle.terminate()
+
+
 @dataclass
 class ManualResultType:
     some_string: str