From 8d1d6c1aa0767a0d0736a110ad23678315358a11 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Wed, 23 Sep 2026 21:38:21 -0700 Subject: [PATCH 1/4] Clarify activity and child workflow return-value errors Signed-off-by: 1fanwang <1fannnw@gmail.com> --- CHANGELOG.md | 1 + temporalio/worker/_activity.py | 13 +- temporalio/worker/_workflow_instance.py | 35 +++-- tests/worker/test_workflow.py | 172 ++++++++++++++++++++++++ 4 files changed, 210 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ee28a3bbf3..2f4407b868 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,7 @@ to include examples, links to docs, or any other relevant information. (e.g., adding `delete` / `adelete`). - `temporalio.contrib.deepagents.TemporalBackend` now forwards the per-command `timeout` of deepagents' `execute` tool for a wrapped sandbox backend such as `LocalShellBackend`. +- Clarify activity and child workflow return-value conversion errors without hiding their original causes. - `GoogleAdkPlugin` now passes the optional `anthropic`, `litellm`, and `openai` SDKs through the workflow sandbox. - `GoogleAdkPlugin` now passes OpenTelemetry modules through the workflow sandbox so ADK 2.9 diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index f72df9cf4b..922e5697a3 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -351,7 +351,18 @@ 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 ( + temporalio.exceptions.FailureError, + concurrent.futures.BrokenExecutor, + ): + 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 5dcc4aa966..2ae50afbd2 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -907,9 +907,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) @@ -945,9 +946,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) @@ -1162,9 +1164,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)}" @@ -2416,6 +2429,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 [] @@ -2430,7 +2445,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_workflow.py b/tests/worker/test_workflow.py index 08d2d42a6e..4865f42cee 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -4808,6 +4808,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 " + ) + 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 From 1a6fe6d81adcee105bd8f26e690d914946450550 Mon Sep 17 00:00:00 2001 From: Stefan Wang <1fannnw@gmail.com> Date: Mon, 28 Sep 2026 12:40:35 -0700 Subject: [PATCH 2/4] Justify the return-value encoding error exclusions The activity return-value message is only safe for errors the surrounding handler does not dispatch on by class, so state that and cover it. Without the guard, an encoder's own FailureError loses its type and message, and a broken executor during encoding stops failing the worker, which is what makes a dead pool unrecoverable. Signed-off-by: Stefan Wang <1fannnw@gmail.com> --- temporalio/worker/_activity.py | 18 +++-- tests/worker/test_activity.py | 117 ++++++++++++++++++++++++++++++++- 2 files changed, 129 insertions(+), 6 deletions(-) diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index 922e5697a3..b951ce54ec 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -353,12 +353,20 @@ async def _handle_start_activity_task( ) try: [payload] = await data_converter.encode([result]) - except ( - temporalio.exceptions.FailureError, - concurrent.futures.BrokenExecutor, - ): - raise except Exception as err: + # The handler below re-types anything that is not already a + # Temporal failure, so re-label only what it dispatches on by + # class: it fails the whole worker on a broken executor, and it + # reports a FailureError as itself instead of an + # ApplicationError. Wrapping either would change the outcome. + if isinstance( + err, + ( + temporalio.exceptions.FailureError, + concurrent.futures.BrokenExecutor, + ), + ): + raise raise temporalio.exceptions.ApplicationError( f"Failed to encode return value of activity {start.activity_type}", type=type(err).__name__, diff --git a/tests/worker/test_activity.py b/tests/worker/test_activity.py index c2add47699..83e8acc15a 100644 --- a/tests/worker/test_activity.py +++ b/tests/worker/test_activity.py @@ -30,7 +30,7 @@ WorkflowHandle, ) from temporalio.common import RawValue, RetryPolicy -from temporalio.converter import DefaultPayloadConverter +from temporalio.converter import DataConverter, DefaultPayloadConverter from temporalio.exceptions import ( ActivityError, ApplicationError, @@ -1243,6 +1243,121 @@ 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_failure_keeps_failure_type( + client: Client, worker: ExternalWorker +): + """An encoder error the worker treats by type must not be re-labeled.""" + + @activity.defn + async def unserializable_result() -> UnserializableResult: + return UnserializableResult() + + act_task_queue = str(uuid.uuid4()) + act_client = _client_failing_result_encoding( + client, ApplicationError("converter failed", type="ConverterError") + ) + async with Worker( + act_client, task_queue=act_task_queue, activities=[unserializable_result] + ): + 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) + cause = err.value.cause.cause + # An ApplicationError is a FailureError, so the encoder's own type and message + # have to survive. Re-labeling would bury the cause under the new message and + # report the failure as an unnamed "ApplicationError". + assert isinstance(cause, ApplicationError) + assert cause.type == "ConverterError" + assert cause.message == "converter failed" + + +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 From 7d3fbef4775cba8df0fde8f4861f3ec97dd76710 Mon Sep 17 00:00:00 2001 From: Stefan Wang <1fannnw@gmail.com> Date: Mon, 28 Sep 2026 13:24:05 -0700 Subject: [PATCH 3/4] Narrow the return-value encoding exclusion to BrokenExecutor Nothing in-tree raises a FailureError from encode, and the only cost of dropping it is losing an encoder's own type and message, which the cause chain already carries. BrokenExecutor stays because the handler fails the worker on it and that check is by type. Signed-off-by: Stefan Wang <1fannnw@gmail.com> --- temporalio/worker/_activity.py | 17 ++++---------- tests/worker/test_activity.py | 41 ---------------------------------- 2 files changed, 4 insertions(+), 54 deletions(-) diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index b951ce54ec..0b0f067c87 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -353,20 +353,11 @@ async def _handle_start_activity_task( ) 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: - # The handler below re-types anything that is not already a - # Temporal failure, so re-label only what it dispatches on by - # class: it fails the whole worker on a broken executor, and it - # reports a FailureError as itself instead of an - # ApplicationError. Wrapping either would change the outcome. - if isinstance( - err, - ( - temporalio.exceptions.FailureError, - concurrent.futures.BrokenExecutor, - ), - ): - raise raise temporalio.exceptions.ApplicationError( f"Failed to encode return value of activity {start.activity_type}", type=type(err).__name__, diff --git a/tests/worker/test_activity.py b/tests/worker/test_activity.py index 83e8acc15a..f3dcf6a9d2 100644 --- a/tests/worker/test_activity.py +++ b/tests/worker/test_activity.py @@ -1272,47 +1272,6 @@ def _client_failing_result_encoding(client: Client, error: BaseException) -> Cli ) -async def test_result_encoding_failure_keeps_failure_type( - client: Client, worker: ExternalWorker -): - """An encoder error the worker treats by type must not be re-labeled.""" - - @activity.defn - async def unserializable_result() -> UnserializableResult: - return UnserializableResult() - - act_task_queue = str(uuid.uuid4()) - act_client = _client_failing_result_encoding( - client, ApplicationError("converter failed", type="ConverterError") - ) - async with Worker( - act_client, task_queue=act_task_queue, activities=[unserializable_result] - ): - 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) - cause = err.value.cause.cause - # An ApplicationError is a FailureError, so the encoder's own type and message - # have to survive. Re-labeling would bury the cause under the new message and - # report the failure as an unnamed "ApplicationError". - assert isinstance(cause, ApplicationError) - assert cause.type == "ConverterError" - assert cause.message == "converter failed" - - async def test_result_encoding_broken_executor_fails_worker( client: Client, worker: ExternalWorker ): From 901e67dabe2cb9f99eb131b392ad01d6bdbd3934 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 1 Oct 2026 14:44:06 -0700 Subject: [PATCH 4/4] Move changelog entry under Unreleased Signed-off-by: 1fanwang <1fannnw@gmail.com> --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f52e7a15ee..8bbc78a872 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,8 @@ to include examples, links to docs, or any other relevant information. OpenTelemetry's threading instrumentation (enabled by strands, among others) is active; a 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. ### Security ## [1.34.0] - 2026-09-30 @@ -109,7 +111,6 @@ to include examples, links to docs, or any other relevant information. (e.g., adding `delete` / `adelete`). - `temporalio.contrib.deepagents.TemporalBackend` now forwards the per-command `timeout` of deepagents' `execute` tool for a wrapped sandbox backend such as `LocalShellBackend`. -- Clarify activity and child workflow return-value conversion errors without hiding their original causes. - `GoogleAdkPlugin` now passes the optional `anthropic`, `litellm`, and `openai` SDKs through the workflow sandbox. - `GoogleAdkPlugin` now passes OpenTelemetry modules through the workflow sandbox so ADK 2.9