wire async activity completion context

Author: dandavison

temporalio/client.py

@@ -62,7 +62,11 @@
 import temporalio.service
 import temporalio.workflow
 from temporalio.activity import ActivityCancellationDetails
-from temporalio.converter import WorkflowSerializationContext
+from temporalio.converter import (
+    ActivitySerializationContext,
+    DataConverter,
+    WorkflowSerializationContext,
+)
 from temporalio.service import (
     HttpConnectProxyConfig,
     KeepAliveConfig,
@@ -6391,10 +6395,11 @@ async def _start_workflow_update_with_start(
     async def heartbeat_async_activity(
         self, input: HeartbeatAsyncActivityInput
     ) -> None:
+        data_converter = self._async_activity_data_converter(input.id_or_token)
         details = (
             None
             if not input.details
-            else await self._client.data_converter.encode_wrapper(input.details)
+            else await data_converter.encode_wrapper(input.details)
         )
         if isinstance(input.id_or_token, AsyncActivityIDReference):
             resp_by_id = await self._client.workflow_service.record_activity_task_heartbeat_by_id(
@@ -6445,10 +6450,11 @@ async def heartbeat_async_activity(
                 )
 
     async def complete_async_activity(self, input: CompleteAsyncActivityInput) -> None:
+        data_converter = self._async_activity_data_converter(input.id_or_token)
         result = (
             None
             if input.result is temporalio.common._arg_unset
-            else await self._client.data_converter.encode_wrapper([input.result])
+            else await data_converter.encode_wrapper([input.result])
         )
         if isinstance(input.id_or_token, AsyncActivityIDReference):
             await self._client.workflow_service.respond_activity_task_completed_by_id(
@@ -6478,14 +6484,14 @@ async def complete_async_activity(self, input: CompleteAsyncActivityInput) -> No
             )
 
     async def fail_async_activity(self, input: FailAsyncActivityInput) -> None:
+        data_converter = self._async_activity_data_converter(input.id_or_token)
+
         failure = temporalio.api.failure.v1.Failure()
-        await self._client.data_converter.encode_failure(input.error, failure)
+        await data_converter.encode_failure(input.error, failure)
         last_heartbeat_details = (
             None
             if not input.last_heartbeat_details
-            else await self._client.data_converter.encode_wrapper(
-                input.last_heartbeat_details
-            )
+            else await data_converter.encode_wrapper(input.last_heartbeat_details)
         )
         if isinstance(input.id_or_token, AsyncActivityIDReference):
             await self._client.workflow_service.respond_activity_task_failed_by_id(
@@ -6519,10 +6525,11 @@ async def fail_async_activity(self, input: FailAsyncActivityInput) -> None:
     async def report_cancellation_async_activity(
         self, input: ReportCancellationAsyncActivityInput
     ) -> None:
+        data_converter = self._async_activity_data_converter(input.id_or_token)
         details = (
             None
             if not input.details
-            else await self._client.data_converter.encode_wrapper(input.details)
+            else await data_converter.encode_wrapper(input.details)
         )
         if isinstance(input.id_or_token, AsyncActivityIDReference):
             await self._client.workflow_service.respond_activity_task_canceled_by_id(
@@ -6551,6 +6558,23 @@ async def report_cancellation_async_activity(
                 timeout=input.rpc_timeout,
             )
 
+    def _async_activity_data_converter(
+        self, id_or_token: Union[AsyncActivityIDReference, bytes]
+    ) -> DataConverter:
+        context = ActivitySerializationContext(
+            namespace=self._client.namespace,
+            workflow_id=(
+                id_or_token.workflow_id
+                if isinstance(id_or_token, AsyncActivityIDReference)
+                else ""
+            ),
+            workflow_type="",
+            activity_type="",
+            activity_task_queue="",
+            is_local=False,
+        )
+        return self._client.data_converter._with_context(context)
+
     ### Schedule calls
 
     async def create_schedule(self, input: CreateScheduleInput) -> ScheduleHandle:

temporalio/converter.py

@@ -28,7 +28,6 @@
     Mapping,
     NewType,
     Optional,
-    Self,
     Sequence,
     Tuple,
     Type,
@@ -44,6 +43,7 @@
 import google.protobuf.symbol_database
 import nexusrpc
 import typing_extensions
+from typing_extensions import Self
 
 import temporalio.api.common.v1
 import temporalio.api.enums.v1

temporalio/worker/_workflow_instance.py

@@ -3187,7 +3187,7 @@ def _resolve_failure(self, err: BaseException) -> None:
             self._result_fut.set_result(None)
 
     def _apply_schedule_command(self) -> None:
-        payload = self._payload_converter.to_payload(self._input.input)
+        payload = self._payload_converter.to_payload(self._input.input)  # type: ignore TODO
         command = self._instance._add_command()
         v = command.schedule_nexus_operation
         v.seq = self._seq