Merge remote-tracking branch 'origin/fix/google-adk-agents-mcp-connection-leak'

Author: wankhede04

CHANGELOG.md

@@ -20,6 +20,11 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- Added experimental SDK payload converter support for values and type hints
+  decorated with `@transfer_type_convertible(...)` using a `TransferTypeConverter` class.
+  This lets types with transfer type converters delegate their wire representation to the
+  configured payload converter, preserving SDK behavior such as serialization
+  contexts.
 - Added `TLSConfig.verification_server_name` to verify the server certificate against a fixed name
   instead of the connection's server name. Unlike `domain`, it does not change the TLS SNI or
   HTTP/2 authority values, which keep following the connected host, so it can be used when the
@@ -37,6 +42,13 @@ to include examples, links to docs, or any other relevant information.
 
 ### Breaking Changes
 
+- Custom workflow runners that construct `WorkflowInstanceDetails` must now pass
+  `payload_converter_factory` instead of `payload_converter_class`. The factory
+  returns the already wrapped payload converter that workflow instances should
+  use.
+- System Nexus payload converter helpers added for generated bindings are now
+  private implementation details, and the remaining public `temporalio.nexus.system`
+  APIs are marked experimental and subject to change.
 - Payload size limits have moved from `DataConverter` to `Client.connect`. Pass
   `payload_limits=PayloadLimitsConfig(...)` (now exported from
   `temporalio.client`) instead of setting `payload_limits` on `DataConverter`.

pyproject.toml

@@ -261,7 +261,7 @@ manifest-path = "temporalio/bridge/Cargo.toml"
 module-name = "temporalio.bridge.temporal_sdk_bridge"
 python-packages = ["temporalio"]
 include = ["LICENSE"]
-exclude = ["temporalio/bridge/target/**/*"]
+exclude = ["temporalio/bridge/target/**/*", "temporalio/bridge/sdk-core/.git"]
 
 [tool.uv]
 # Prevent uv commands from building the package by default

scripts/gen_payload_visitor.py

@@ -191,7 +191,7 @@ async def _visit_nexus_operation_input_payload(
         endpoint: str,
         payload: Payload,
     ) -> None:
-        new_payload = await temporalio.nexus.system.maybe_visit_payload(
+        new_payload = await temporalio.nexus.system._maybe_visit_payload(
             endpoint,
             payload,
             fs,

temporalio/activity.py

@@ -29,6 +29,9 @@
 import temporalio.bridge.proto.activity_task
 import temporalio.common
 import temporalio.converter
+from temporalio.converter._payload_converter import (
+    _TemporalTransferTypePayloadConverter,
+)
 
 from .types import CallableType
 
@@ -238,9 +241,13 @@ def payload_converter(self) -> temporalio.converter.PayloadConverter:
                 self.payload_converter_class_or_instance,
                 temporalio.converter.PayloadConverter,
             ):
-                self._payload_converter = self.payload_converter_class_or_instance
+                self._payload_converter = _TemporalTransferTypePayloadConverter.wrap(
+                    self.payload_converter_class_or_instance
+                )
             else:
-                self._payload_converter = self.payload_converter_class_or_instance()
+                self._payload_converter = _TemporalTransferTypePayloadConverter.wrap(
+                    self.payload_converter_class_or_instance()
+                )
         return self._payload_converter
 
     @property

temporalio/bridge/_visitor.py

@@ -61,7 +61,7 @@ async def _visit_nexus_operation_input_payload(
         endpoint: str,
         payload: Payload,
     ) -> None:
-        new_payload = await temporalio.nexus.system.maybe_visit_payload(
+        new_payload = await temporalio.nexus.system._maybe_visit_payload(
             endpoint,
             payload,
             fs,

temporalio/converter/__init__.py

@@ -33,6 +33,8 @@
     JSONTypeConverter,
     JSONTypeConverterUnhandled,
     PayloadConverter,
+    TransferTypeConverter,
+    transfer_type_convertible,
     value_to_type,
 )
 from temporalio.converter._search_attributes import (
@@ -64,6 +66,7 @@
     "BinaryPlainPayloadConverter",
     "BinaryProtoPayloadConverter",
     "CompositePayloadConverter",
+    "TransferTypeConverter",
     "DataConverter",
     "DefaultFailureConverter",
     "DefaultFailureConverterWithEncodedAttributes",
@@ -79,6 +82,7 @@
     "SerializationContext",
     "WithSerializationContext",
     "WorkflowSerializationContext",
+    "transfer_type_convertible",
     "decode_search_attributes",
     "decode_typed_search_attributes",
     "default",

temporalio/converter/_data_converter.py

@@ -28,6 +28,7 @@
 )
 from temporalio.converter._payload_converter import (
     PayloadConverter,
+    _TemporalTransferTypePayloadConverter,
 )
 from temporalio.converter._serialization_context import (
     SerializationContext,
@@ -90,9 +91,15 @@ class DataConverter(WithSerializationContext):
     """Singleton default data converter."""
 
     def __post_init__(self) -> None:  # noqa: D105
-        object.__setattr__(self, "payload_converter", self.payload_converter_class())
+        object.__setattr__(self, "payload_converter", self._new_payload_converter())
         object.__setattr__(self, "failure_converter", self.failure_converter_class())
 
+    def _new_payload_converter(self) -> PayloadConverter:
+        """Create a payload converter instance with SDK transfer type hooks enabled."""
+        return _TemporalTransferTypePayloadConverter.wrap(
+            self.payload_converter_class()
+        )
+
     async def encode(
         self, values: Sequence[Any]
     ) -> list[temporalio.api.common.v1.Payload]:

temporalio/converter/_payload_converter.py

@@ -21,6 +21,7 @@
 from typing import (
     Any,
     ClassVar,
+    Generic,
     Literal,
     NewType,
     TypeVar,
@@ -51,6 +52,75 @@
 )
 
 _sym_db = google.protobuf.symbol_database.Default()
+ValueT = TypeVar("ValueT")
+TransferTypeT = TypeVar("TransferTypeT")
+_TRANSFER_TYPE_CONVERTER_ATTR = "__temporal_transfer_type_converter"
+
+
+class TransferTypeConverter(Generic[ValueT, TransferTypeT], ABC):
+    """Converter between a user-facing value and a transfer type value.
+
+    .. warning::
+        This API is experimental and subject to change.
+    """
+
+    transfer_type: type[TransferTypeT] | None = None
+    """Optional type hint for the transfer type to use when decoding payloads.
+
+    .. warning::
+        This API is experimental and subject to change.
+    """
+
+    @abstractmethod
+    def to_transfer_type(self, value: ValueT) -> TransferTypeT:
+        """Convert a user-facing value to its transfer type value.
+
+        .. warning::
+            This API is experimental and subject to change.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def from_transfer_type(self, value: TransferTypeT) -> ValueT:
+        """Convert a transfer type value to its user-facing value.
+
+        .. warning::
+            This API is experimental and subject to change.
+        """
+        raise NotImplementedError
+
+
+class _TransferTypeConvertibleDecorator(Generic[ValueT, TransferTypeT]):
+    def __init__(
+        self, converter_type: type[TransferTypeConverter[ValueT, TransferTypeT]]
+    ) -> None:
+        self._converter_type = converter_type
+
+    def __call__(self, cls: type[ValueT]) -> type[ValueT]:
+        if hasattr(cls, _TRANSFER_TYPE_CONVERTER_ATTR):
+            raise TypeError("class already has a transfer type converter")
+        setattr(cls, _TRANSFER_TYPE_CONVERTER_ATTR, self._converter_type())
+        return cls
+
+
+def transfer_type_convertible(
+    converter_type: type[TransferTypeConverter[ValueT, TransferTypeT]],
+) -> _TransferTypeConvertibleDecorator[ValueT, TransferTypeT]:
+    """Decorate a class with a transfer type converter class.
+
+    .. warning::
+        This API is experimental and subject to change.
+    """
+    return _TransferTypeConvertibleDecorator(converter_type)
+
+
+def _get_transfer_type_converter(
+    value_type: object,
+) -> TransferTypeConverter[Any, Any] | None:
+    converter = getattr(value_type, _TRANSFER_TYPE_CONVERTER_ATTR, None)
+    if isinstance(converter, TransferTypeConverter):
+        return converter
+    return None
 
 
 class PayloadConverter(ABC):
@@ -514,6 +584,74 @@ def from_payload(
             raise RuntimeError("Failed parsing") from err
 
 
+class _TemporalTransferTypePayloadConverter(PayloadConverter, WithSerializationContext):
+    """Payload converter wrapper for registered Temporal transfer type converters.
+
+    Values with a registered transfer type converter are first converted to their
+    transfer type value, then encoded by the wrapped payload converter. When
+    decoding to a type with a registered transfer type converter, the wrapped
+    converter first decodes the payload to the transfer type value and this wrapper
+    constructs the requested user-facing type from it.
+    """
+
+    _inner_payload_converter: PayloadConverter
+
+    def __init__(self, inner_payload_converter: PayloadConverter) -> None:
+        """Create a Temporal transfer type payload converter."""
+        self._inner_payload_converter = inner_payload_converter
+
+    @staticmethod
+    def wrap(payload_converter: PayloadConverter) -> PayloadConverter:
+        """Wrap a payload converter unless it is already wrapped."""
+        if isinstance(payload_converter, _TemporalTransferTypePayloadConverter):
+            return payload_converter
+        return _TemporalTransferTypePayloadConverter(payload_converter)
+
+    def to_payloads(
+        self, values: Sequence[Any]
+    ) -> list[temporalio.api.common.v1.Payload]:
+        """See base class."""
+        transfer_type_values: list[Any] = []
+        for value in values:
+            converter = _get_transfer_type_converter(type(value))
+            if converter is not None:
+                value = converter.to_transfer_type(value)
+            transfer_type_values.append(value)
+        return self._inner_payload_converter.to_payloads(transfer_type_values)
+
+    def from_payloads(
+        self,
+        payloads: Sequence[temporalio.api.common.v1.Payload],
+        type_hints: list[type] | None = None,
+    ) -> list[Any]:
+        """See base class."""
+        if type_hints is None:
+            return self._inner_payload_converter.from_payloads(payloads, None)
+        converters = [
+            _get_transfer_type_converter(type_hint) for type_hint in type_hints
+        ]
+        inner_type_hints = [
+            converter.transfer_type if converter is not None else type_hint
+            for converter, type_hint in zip(converters, type_hints)
+        ]
+        values = self._inner_payload_converter.from_payloads(
+            payloads, typing.cast("list[type]", inner_type_hints)
+        )
+        return [
+            converter.from_transfer_type(value) if converter is not None else value
+            for value, converter in zip(values, converters)
+        ]
+
+    def with_context(self, context: SerializationContext) -> Self:
+        """Return a new instance with context set on the inner converter."""
+        if not isinstance(self._inner_payload_converter, WithSerializationContext):
+            return self
+        inner_payload_converter = self._inner_payload_converter.with_context(context)
+        if inner_payload_converter is self._inner_payload_converter:
+            return self
+        return type(self)(inner_payload_converter)
+
+
 class AdvancedJSONEncoder(json.JSONEncoder):
     """Advanced JSON encoder.