Serialization context (#1102)

# Problem 

We must make it possible for users to create custom data converters, the methods of which have access to relevant context. In other words, when implementing any of the above interfaces, the user must be able to implement methods that have access to a context object.

# Solution

1. At every point where the data converter is used, we construct a context-aware data converter by calling `data_converter._with_context(context)`.
2. Users take advantage of this by implementing `with_context` methods on any of the following components:
    - payload converter
    - failure converter
    - payload codec
3. In addition, users may implement `with_context` on an `EncodingPayloadConverter`. These are used by the SDK’s default payload converter.
4. In addition, *we* provide a `with_context` method on `AsyncActivityHandle` that *they* can call their own activity context.
5. Add CommandAwarePayloadVisitor that sets command context when visiting payloads; use this for workflow activation decoding and encoding of the activation completion.

Author: dandavison

.github/workflows/ci.yml

@@ -148,6 +148,8 @@ jobs:
         with:
           submodules: recursive
       - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: "clippy"
       - uses: Swatinem/rust-cache@v2
         with:
           workspaces: temporalio/bridge -> target

scripts/gen_payload_visitor.py

@@ -1,7 +1,7 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import Optional, Tuple
+from typing import Optional
 
 from google.protobuf.descriptor import Descriptor, FieldDescriptor
 
@@ -89,6 +89,7 @@ def generate(self, roots: list[Descriptor]) -> str:
 
 from temporalio.api.common.v1.message_pb2 import Payload
 
+
 class VisitorFunctions(abc.ABC):
     \"\"\"Set of functions which can be called by the visitor. 
     Allows handling payloads as a sequence.

temporalio/activity.py

@@ -470,6 +470,7 @@ class _CompleteAsyncError(BaseException):
 def payload_converter() -> temporalio.converter.PayloadConverter:
     """Get the payload converter for the current activity.
 
+    The returned converter has :py:class:`temporalio.converter.ActivitySerializationContext` set.
     This is often used for dynamic activities to convert payloads.
     """
     return _Context.current().payload_converter

temporalio/bridge/worker.py

@@ -7,11 +7,9 @@
 
 from dataclasses import dataclass
 from typing import (
-    TYPE_CHECKING,
     Awaitable,
     Callable,
     List,
-    Mapping,
     MutableSequence,
     Optional,
     Sequence,
@@ -20,7 +18,6 @@
     Union,
 )
 
-import google.protobuf.internal.containers
 from typing_extensions import TypeAlias
 
 import temporalio.api.common.v1
@@ -35,12 +32,13 @@
 import temporalio.bridge.temporal_sdk_bridge
 import temporalio.converter
 import temporalio.exceptions
-from temporalio.api.common.v1.message_pb2 import Payload, Payloads
-from temporalio.bridge._visitor import PayloadVisitor, VisitorFunctions
+from temporalio.api.common.v1.message_pb2 import Payload
+from temporalio.bridge._visitor import VisitorFunctions
 from temporalio.bridge.temporal_sdk_bridge import (
     CustomSlotSupplier as BridgeCustomSlotSupplier,
 )
 from temporalio.bridge.temporal_sdk_bridge import PollShutdownError  # type: ignore
+from temporalio.worker._command_aware_visitor import CommandAwarePayloadVisitor
 
 
 @dataclass
@@ -299,22 +297,22 @@ async def visit_payloads(self, payloads: MutableSequence[Payload]) -> None:
 
 
 async def decode_activation(
-    act: temporalio.bridge.proto.workflow_activation.WorkflowActivation,
+    activation: temporalio.bridge.proto.workflow_activation.WorkflowActivation,
     codec: temporalio.converter.PayloadCodec,
     decode_headers: bool,
 ) -> None:
-    """Decode the given activation with the codec."""
-    await PayloadVisitor(
+    """Decode all payloads in the activation."""
+    await CommandAwarePayloadVisitor(
         skip_search_attributes=True, skip_headers=not decode_headers
-    ).visit(_Visitor(codec.decode), act)
+    ).visit(_Visitor(codec.decode), activation)
 
 
 async def encode_completion(
-    comp: temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion,
+    completion: temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion,
     codec: temporalio.converter.PayloadCodec,
     encode_headers: bool,
 ) -> None:
-    """Recursively encode the given completion with the codec."""
-    await PayloadVisitor(
+    """Encode all payloads in the completion."""
+    await CommandAwarePayloadVisitor(
         skip_search_attributes=True, skip_headers=not encode_headers
-    ).visit(_Visitor(codec.encode), comp)
+    ).visit(_Visitor(codec.encode), completion)