[Bug] Failures in activites are not decoded in workflow when custom codec is used

[romank0]

What are you really trying to do?

I use a custom codec to encrypt payload data.

Describe the bug

If the activity raises an exception with details failure is properly encoded and stored. But in the worker when the workflow is activated the following error happens and workflow cannot proceed and eventually time outs:

Failed activation on workflow MyWorkflow with ID 9c556743-9ebf-4c3b-9800-3c56f02248a7 and run ID 3aeb4af6-97b6-43df-9c3e-a8b0f4326cd6
Traceback (most recent call last):
  File "..venv/lib/python3.11/site-packages/temporalio/worker/_workflow_instance.py", line 413, in activate
    self._apply(job)
  File "..venv/lib/python3.11/site-packages/temporalio/worker/_workflow_instance.py", line 512, in _apply
    self._apply_resolve_activity(job.resolve_activity)
  File "..venv/lib/python3.11/site-packages/temporalio/worker/_workflow_instance.py", line 769, in _apply_resolve_activity
    self._failure_converter.from_failure(
  File "..venv/lib/python3.11/site-packages/temporalio/converter.py", line 1096, in from_failure
    err.__cause__ = self.from_failure(failure.cause, payload_converter)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "..venv/lib/python3.11/site-packages/temporalio/converter.py", line 992, in from_failure
    *payload_converter.from_payloads_wrapper(app_info.details),
     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "..venv/lib/python3.11/site-packages/temporalio/converter.py", line 139, in from_payloads_wrapper
    return self.from_payloads(payloads.payloads)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "..venv/lib/python3.11/site-packages/temporalio/converter.py", line 309, in from_payloads
    raise KeyError(f"Unknown payload encoding {encoding.decode()}")
KeyError: 'Unknown payload encoding binary/encrypted'

Minimal Reproduction

This test when added to tests/worker/tests_workflow.py fails because of the timeout:

@activity.defn
async def activity_that_fails_with_details() -> str:
    raise ApplicationError(
        "Activity failed intentionally",
        "detail1",
        {"error_code": "NOT_FOUND", "id": "test-123"},
        non_retryable=True,
    )


@workflow.defn
class WorkflowWithFailingActivityAndCodec:
    @workflow.run
    async def run(self) -> str:
        try:
            return await workflow.execute_activity(
                activity_that_fails_with_details,
                schedule_to_close_timeout=timedelta(seconds=3),
                retry_policy=RetryPolicy(maximum_attempts=1),
            )
        except ActivityError as err:
            return "Handled encrypted failure successfully"


async def test_activity_failure_with_encoded_payload_is_decoded_in_workflow(client: Client):
    config = client.config()
    config["data_converter"] = dataclasses.replace(
        temporalio.converter.default(), payload_codec=EncryptionCodec()
    )
    client = Client(**config)

    async with new_worker(
        client,
        WorkflowWithFailingActivityAndCodec,
        activities=[activity_that_fails_with_details],
    ) as worker:
        result = await client.execute_workflow(
            WorkflowWithFailingActivityAndCodec.run,
            id=f"workflow-{uuid.uuid4()}",
            task_queue=worker.task_queue,
            run_timeout=timedelta(seconds=5),
        )
        assert result == "Handled encrypted failure successfully"

Environment/Versions

• OS and processor: fails both on M1 Mac and Linux, docker and native.
• Temporal Version: 2.31.2, python sdk 1.18.0, 1.18.1 and master

Additional context

No.