From 4ea369ad18ce37963263f5dece52dd3abd7af732 Mon Sep 17 00:00:00 2001 From: Gregory Michael Travis Date: Mon, 14 Sep 2026 19:30:04 -0400 Subject: [PATCH 1/9] Remove Experimental tag from Standalone Activities (#1863) * Remove Experimental tag from Standalone Activities. --- CHANGELOG.md | 3 ++ temporalio/client/_activity.py | 51 +++------------------- temporalio/client/_client.py | 27 ------------ temporalio/client/_exceptions.py | 6 +-- temporalio/client/_interceptor.py | 72 ++++++------------------------- temporalio/common.py | 6 --- 6 files changed, 21 insertions(+), 144 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f1a526e7e..33c575183 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,9 @@ to include examples, links to docs, or any other relevant information. ### Changed +- Standalone Activities are now generally available (GA). (Standalone Activities as Nexus operations + and Standalone Activities operator commands remain experimental. Operator commands are `pause`, + `unpause`, `updateOptions`, `restoreOriginal`.) - System Nexus Signal-with-Start Workflow operations now use the typed `WorkflowOutboundInterceptor.start_signal_with_start_workflow` interception point instead of the generic `WorkflowOutboundInterceptor.start_nexus_operation` method. diff --git a/temporalio/client/_activity.py b/temporalio/client/_activity.py index 9b325c51c..df7cd5c28 100644 --- a/temporalio/client/_activity.py +++ b/temporalio/client/_activity.py @@ -65,9 +65,6 @@ class ActivityExecutionAsyncIterator: """Asynchronous iterator for activity execution values. You should typically use ``async for`` on this iterator and not call any of its methods. - - .. warning:: - This API is experimental. """ def __init__( @@ -168,11 +165,7 @@ async def __anext__(self) -> ActivityExecution: @dataclass(frozen=True, eq=False, kw_only=True) class ActivityExecution: - """Info for an activity execution not started by a workflow, from list response. - - .. warning:: - This API is experimental. - """ + """Info for an activity execution not started by a workflow, from list response.""" activity_id: str """Activity ID.""" @@ -265,11 +258,7 @@ def _from_raw_info( @dataclass(frozen=True, eq=False, kw_only=True) class ActivityExecutionDescription(ActivityExecution): - """Detailed information about an activity execution not started by a workflow. - - .. warning:: - This API is experimental. - """ + """Detailed information about an activity execution not started by a workflow.""" attempt: int """Current attempt number.""" @@ -563,9 +552,6 @@ async def outcome_failure(self) -> BaseException | None: class ActivityExecutionStatus(IntEnum): """Status of an activity execution. - .. warning:: - This API is experimental. - See :py:class:`temporalio.api.enums.v1.ActivityExecutionStatus`. """ @@ -598,9 +584,6 @@ class ActivityExecutionStatus(IntEnum): class PendingActivityState(IntEnum): """Detailed state of an activity execution that is in ACTIVITY_EXECUTION_STATUS_RUNNING. - .. warning:: - This API is experimental. - See :py:class:`temporalio.api.enums.v1.PendingActivityState`. """ @@ -782,11 +765,7 @@ def _from_proto( @dataclass(frozen=True) class ActivityExecutionCount: - """Representation of a count from a count activities call. - - .. warning:: - This API is experimental. - """ + """Representation of a count from a count activities call.""" count: int """Total count matching the filter, if any.""" @@ -809,11 +788,7 @@ def _from_raw( @dataclass(frozen=True) class ActivityExecutionCountAggregationGroup: - """A single aggregation group from a count activities call. - - .. warning:: - This API is experimental. - """ + """A single aggregation group from a count activities call.""" count: int """Count for this group.""" @@ -985,11 +960,7 @@ def with_context(self, context: SerializationContext) -> Self: class ActivityHandle(Generic[ReturnType]): - """Handle representing an activity execution not started by a workflow. - - .. warning:: - This API is experimental. - """ + """Handle representing an activity execution not started by a workflow.""" def __init__( self, @@ -1040,9 +1011,6 @@ async def result( ) -> ReturnType: """Wait for result of the activity. - .. warning:: - This API is experimental. - The result may already be known if this method has been called before, in which case no network call is made. Otherwise the result will be polled for until it is available. @@ -1133,9 +1101,6 @@ async def cancel( ) -> None: """Request cancellation of the activity. - .. warning:: - This API is experimental. - Requesting cancellation of an activity does not automatically transition the activity to canceled status. If the activity is heartbeating, a :py:class:`exceptions.CancelledError` exception will be raised when receiving the heartbeat response; if the activity allows this @@ -1166,9 +1131,6 @@ async def terminate( ) -> None: """Terminate the activity execution immediately. - .. warning:: - This API is experimental. - Termination does not reach the worker and the activity code cannot react to it. A terminated activity may have a running attempt and will be requested to be canceled by the server when it heartbeats. @@ -1348,9 +1310,6 @@ async def describe( ) -> ActivityExecutionDescription: """Describe the activity execution. - .. warning:: - This API is experimental. - Args: include_input: Include activity input in the response if available. include_outcome: Include activity outcome in the response if available. diff --git a/temporalio/client/_client.py b/temporalio/client/_client.py index 3c89c10ff..b5db0c4b7 100644 --- a/temporalio/client/_client.py +++ b/temporalio/client/_client.py @@ -1487,9 +1487,6 @@ async def start_activity( ) -> ActivityHandle[ReturnType]: """Start an activity and return its handle. - .. warning:: - This API is experimental. - Args: activity: String name or callable activity function to execute. arg: Single argument to the activity. @@ -1741,9 +1738,6 @@ async def execute_activity( ) -> ReturnType: """Start an activity, wait for it to complete, and return its result. - .. warning:: - This API is experimental. - This is a convenience method that combines :py:meth:`start_activity` and :py:meth:`ActivityHandle.result`. @@ -1943,9 +1937,6 @@ async def start_activity_class( ) -> ActivityHandle[Any]: """Start an activity from a callable class. - .. warning:: - This API is experimental. - See :py:meth:`start_activity` for parameter and return details. """ return await self.start_activity( @@ -2137,9 +2128,6 @@ async def execute_activity_class( ) -> Any: """Start an activity from a callable class and wait for completion. - .. warning:: - This API is experimental. - This is a shortcut for ``await`` :py:meth:`start_activity_class`. """ return await self.execute_activity( @@ -2286,9 +2274,6 @@ async def start_activity_method( ) -> ActivityHandle[Any]: """Start an activity from a method. - .. warning:: - This API is experimental. - See :py:meth:`start_activity` for parameter and return details. """ return await self.start_activity( @@ -2435,9 +2420,6 @@ async def execute_activity_method( ) -> Any: """Start an activity from a method and wait for completion. - .. warning:: - This API is experimental. - This is a shortcut for ``await`` :py:meth:`start_activity_method`. """ return await self.execute_activity( @@ -2474,9 +2456,6 @@ def list_activities( ) -> ActivityExecutionAsyncIterator: """List activities not started by a workflow. - .. warning:: - This API is experimental. - This does not make a request until the first iteration is attempted. Therefore any errors will not occur until then. @@ -2517,9 +2496,6 @@ async def count_activities( ) -> ActivityExecutionCount: """Count activities not started by a workflow. - .. warning:: - This API is experimental. - Args: query: A Temporal visibility filter for activities. rpc_metadata: Headers used on the RPC call. Keys here override @@ -2563,9 +2539,6 @@ def get_activity_handle( The activity must not have been started by a workflow. - .. warning:: - This API is experimental. - To get a handle to an activity execution that you control for manual completion and heartbeating, see :py:meth:`Client.get_async_activity_handle`. diff --git a/temporalio/client/_exceptions.py b/temporalio/client/_exceptions.py index fdcd263e6..2d91372f1 100644 --- a/temporalio/client/_exceptions.py +++ b/temporalio/client/_exceptions.py @@ -104,11 +104,7 @@ def __init__(self) -> None: class ActivityFailureError(temporalio.exceptions.TemporalError): - """Error that occurs when an activity is unsuccessful. - - .. warning:: - This API is experimental. - """ + """Error that occurs when an activity is unsuccessful.""" def __init__(self, *, cause: BaseException) -> None: """Create activity failure error.""" diff --git a/temporalio/client/_interceptor.py b/temporalio/client/_interceptor.py index b9d82d6ea..fedec2cf6 100644 --- a/temporalio/client/_interceptor.py +++ b/temporalio/client/_interceptor.py @@ -202,11 +202,7 @@ class TerminateWorkflowInput: @dataclass class StartActivityInput: - """Input for :py:meth:`OutboundInterceptor.start_activity`. - - .. warning:: - This API is experimental. - """ + """Input for :py:meth:`OutboundInterceptor.start_activity`.""" activity_type: str args: Sequence[Any] @@ -231,11 +227,7 @@ class StartActivityInput: @dataclass class CancelActivityInput: - """Input for :py:meth:`OutboundInterceptor.cancel_activity`. - - .. warning:: - This API is experimental. - """ + """Input for :py:meth:`OutboundInterceptor.cancel_activity`.""" activity_id: str activity_run_id: str | None @@ -246,11 +238,7 @@ class CancelActivityInput: @dataclass class TerminateActivityInput: - """Input for :py:meth:`OutboundInterceptor.terminate_activity`. - - .. warning:: - This API is experimental. - """ + """Input for :py:meth:`OutboundInterceptor.terminate_activity`.""" activity_id: str activity_run_id: str | None @@ -308,11 +296,7 @@ class UpdateActivityOptionsInput: @dataclass class DescribeActivityInput: - """Input for :py:meth:`OutboundInterceptor.describe_activity`. - - .. warning:: - This API is experimental. - """ + """Input for :py:meth:`OutboundInterceptor.describe_activity`.""" activity_id: str activity_run_id: str | None @@ -326,11 +310,7 @@ class DescribeActivityInput: @dataclass class ListActivitiesInput: - """Input for :py:meth:`OutboundInterceptor.list_activities`. - - .. warning:: - This API is experimental. - """ + """Input for :py:meth:`OutboundInterceptor.list_activities`.""" query: str | None page_size: int @@ -342,11 +322,7 @@ class ListActivitiesInput: @dataclass class CountActivitiesInput: - """Input for :py:meth:`OutboundInterceptor.count_activities`. - - .. warning:: - This API is experimental. - """ + """Input for :py:meth:`OutboundInterceptor.count_activities`.""" query: str | None rpc_metadata: Mapping[str, str | bytes] @@ -807,27 +783,15 @@ async def terminate_workflow(self, input: TerminateWorkflowInput) -> None: ### Activity calls async def start_activity(self, input: StartActivityInput) -> ActivityHandle[Any]: - """Called for every :py:meth:`Client.start_activity` call. - - .. warning:: - This API is experimental. - """ + """Called for every :py:meth:`Client.start_activity` call.""" return await self.next.start_activity(input) async def cancel_activity(self, input: CancelActivityInput) -> None: - """Called for every :py:meth:`ActivityHandle.cancel` call. - - .. warning:: - This API is experimental. - """ + """Called for every :py:meth:`ActivityHandle.cancel` call.""" await self.next.cancel_activity(input) async def terminate_activity(self, input: TerminateActivityInput) -> None: - """Called for every :py:meth:`ActivityHandle.terminate` call. - - .. warning:: - This API is experimental. - """ + """Called for every :py:meth:`ActivityHandle.terminate` call.""" await self.next.terminate_activity(input) async def pause_activity(self, input: PauseActivityInput) -> None: @@ -860,31 +824,19 @@ async def update_activity_options( async def describe_activity( self, input: DescribeActivityInput ) -> ActivityExecutionDescription: - """Called for every :py:meth:`ActivityHandle.describe` call. - - .. warning:: - This API is experimental. - """ + """Called for every :py:meth:`ActivityHandle.describe` call.""" return await self.next.describe_activity(input) def list_activities( self, input: ListActivitiesInput ) -> ActivityExecutionAsyncIterator: - """Called for every :py:meth:`Client.list_activities` call. - - .. warning:: - This API is experimental. - """ + """Called for every :py:meth:`Client.list_activities` call.""" return self.next.list_activities(input) async def count_activities( self, input: CountActivitiesInput ) -> ActivityExecutionCount: - """Called for every :py:meth:`Client.count_activities` call. - - .. warning:: - This API is experimental. - """ + """Called for every :py:meth:`Client.count_activities` call.""" return await self.next.count_activities(input) async def start_workflow_update( diff --git a/temporalio/common.py b/temporalio/common.py index ad75b56b9..04081bf19 100644 --- a/temporalio/common.py +++ b/temporalio/common.py @@ -151,9 +151,6 @@ class WorkflowIDConflictPolicy(IntEnum): class ActivityIDReusePolicy(IntEnum): """How already-closed activity IDs are handled on start. - .. warning:: - This API is experimental. - See :py:class:`temporalio.api.enums.v1.ActivityIdReusePolicy`. """ @@ -174,9 +171,6 @@ class ActivityIDReusePolicy(IntEnum): class ActivityIDConflictPolicy(IntEnum): """How already-running activity IDs are handled on start. - .. warning:: - This API is experimental. - See :py:class:`temporalio.api.enums.v1.ActivityIdConflictPolicy`. """ From 263f7822f181a83426c67e8d9020d594b719f321 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 14 Sep 2026 16:35:51 -0700 Subject: [PATCH 2/9] Prepare release 1.33.0 (#1864) --- CHANGELOG.md | 18 +++++++++++--- pyproject.toml | 2 +- temporalio/service.py | 2 +- uv.lock | 58 +++++++++++++++++++++---------------------- 4 files changed, 45 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33c575183..bf2cab36e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,20 @@ to include examples, links to docs, or any other relevant information. ### Added +### Changed + +### Deprecated + +### :boom: Breaking Changes + +### Fixed + +### Security + +## [1.33.0] - 2026-09-14 + +### Added + #### Standalone Activity operator commands - `ActivityHandle` now supports operator commands for standalone activities: `pause`, @@ -40,8 +54,6 @@ to include examples, links to docs, or any other relevant information. `WorkflowOutboundInterceptor.start_system_nexus_operation` after their typed interception point. They continue not to invoke `WorkflowOutboundInterceptor.start_nexus_operation`. -### Deprecated - ### :boom: Breaking Changes - Experimental external storage: `ExternalStorage.driver_selector` is now called with a @@ -90,8 +102,6 @@ to include examples, links to docs, or any other relevant information. or callbacks to attach. - The workflow sandbox now passes `pydantic_core` through by default, alongside `pydantic`. -### Security - ## [1.32.0] - 2026-08-24 ### Added diff --git a/pyproject.toml b/pyproject.toml index 27114dc67..fc2d3d17c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "temporalio" -version = "1.32.0" +version = "1.33.0" description = "Temporal.io Python SDK" authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }] requires-python = ">=3.10" diff --git a/temporalio/service.py b/temporalio/service.py index 6a3a053dd..c0807f666 100644 --- a/temporalio/service.py +++ b/temporalio/service.py @@ -24,7 +24,7 @@ import temporalio.runtime from temporalio.bridge.client import RPCError as BridgeRPCError -__version__ = "1.32.0" +__version__ = "1.33.0" ServiceRequest = TypeVar("ServiceRequest", bound=google.protobuf.message.Message) ServiceResponse = TypeVar("ServiceResponse", bound=google.protobuf.message.Message) diff --git a/uv.lock b/uv.lock index 767c4e587..8dce7d158 100644 --- a/uv.lock +++ b/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. +exclude-newer = "2026-08-31T19:12:49.465398Z" exclude-newer-span = "P2W" [[package]] @@ -257,14 +257,14 @@ name = "anthropic" version = "0.117.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anyio" }, - { name = "distro" }, - { name = "docstring-parser" }, - { name = "httpx" }, - { name = "jiter" }, - { name = "pydantic" }, - { name = "sniffio" }, - { name = "typing-extensions" }, + { name = "anyio", marker = "python_full_version >= '3.11'" }, + { name = "distro", marker = "python_full_version >= '3.11'" }, + { name = "docstring-parser", marker = "python_full_version >= '3.11'" }, + { name = "httpx", marker = "python_full_version >= '3.11'" }, + { name = "jiter", marker = "python_full_version >= '3.11'" }, + { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "sniffio", marker = "python_full_version >= '3.11'" }, + { name = "typing-extensions", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/41/0d/8f71d535edb0d438f023bd825fb65f67c14fa88a2bd6b75f292a58a63de4/anthropic-0.117.0.tar.gz", hash = "sha256:98107f2b76439641e0ae2a1754087534b8f178dbab99d6eb1bc4b7bc8c744496", size = 989933, upload-time = "2026-07-16T19:36:13.07Z" } wheels = [ @@ -942,12 +942,12 @@ name = "deepagents" version = "0.6.12" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain" }, - { name = "langchain-anthropic" }, - { name = "langchain-core" }, - { name = "langchain-google-genai" }, - { name = "langsmith" }, - { name = "wcmatch" }, + { name = "langchain", marker = "python_full_version >= '3.11'" }, + { name = "langchain-anthropic", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core", marker = "python_full_version >= '3.11'" }, + { name = "langchain-google-genai", marker = "python_full_version >= '3.11'" }, + { name = "langsmith", marker = "python_full_version >= '3.11'" }, + { name = "wcmatch", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e0/db/a6acdc72a9e90c3f07ed10de35c951734a02d4facb693bb59684ad368801/deepagents-0.6.12.tar.gz", hash = "sha256:1f281c0bc5a63132f62e2ee345c1dc593b23188da6e23016401f6879fbe54b5f", size = 211364, upload-time = "2026-06-25T17:26:52.775Z" } wheels = [ @@ -1022,7 +1022,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1971,9 +1971,9 @@ name = "langchain" version = "1.3.14" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain-core" }, - { name = "langgraph" }, - { name = "pydantic" }, + { name = "langchain-core", marker = "python_full_version >= '3.11'" }, + { name = "langgraph", marker = "python_full_version >= '3.11'" }, + { name = "pydantic", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/29/68/a6dbad9c22df4087a0f9e79ddd46226c442b30128bfeee538d5889492a73/langchain-1.3.14.tar.gz", hash = "sha256:1b6696c72ba3bbbce54d745e0180742c9f6ece8bbc59ed5a46c3e20b9a435929", size = 645181, upload-time = "2026-07-16T13:28:18.29Z" } wheels = [ @@ -1985,9 +1985,9 @@ name = "langchain-anthropic" version = "1.4.8" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anthropic" }, - { name = "langchain-core" }, - { name = "pydantic" }, + { name = "anthropic", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core", marker = "python_full_version >= '3.11'" }, + { name = "pydantic", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/98/22/40ab129b08329ca295b391aa1d48267692b42594757084c6918e22b655ac/langchain_anthropic-1.4.8.tar.gz", hash = "sha256:c76891b2044d56105ff13c106ed12650637b53bd598a4bdf15b4796eefa2a4ec", size = 708524, upload-time = "2026-06-26T21:28:46.916Z" } wheels = [ @@ -2019,10 +2019,10 @@ name = "langchain-google-genai" version = "4.2.7" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "filetype" }, - { name = "google-genai" }, - { name = "langchain-core" }, - { name = "pydantic" }, + { name = "filetype", marker = "python_full_version >= '3.11'" }, + { name = "google-genai", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core", marker = "python_full_version >= '3.11'" }, + { name = "pydantic", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1d/0c/bc60dabc362ca7c6ffe8c4bcc2f724c7e566b43eb230cee51419f88f784c/langchain_google_genai-4.2.7.tar.gz", hash = "sha256:03b1463ffe4d42435f43c7870467f2215f684bb46400d2543435d10157c80ac7", size = 281605, upload-time = "2026-07-06T13:51:58.724Z" } wheels = [ @@ -2838,7 +2838,7 @@ wheels = [ [package.optional-dependencies] litellm = [ - { name = "litellm" }, + { name = "litellm", marker = "python_full_version < '3.14'" }, ] [[package]] @@ -4696,7 +4696,7 @@ wheels = [ [[package]] name = "temporalio" -version = "1.32.0" +version = "1.33.0" source = { virtual = "." } dependencies = [ { name = "nexus-rpc" }, @@ -5377,7 +5377,7 @@ name = "wcmatch" version = "11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "bracex" }, + { name = "bracex", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/16/25/1da725838132221e33568973da484ff43813662ccc06ebf7f6e3abddfcd5/wcmatch-11.0.tar.gz", hash = "sha256:55d95c2447789712774b198ceec72939e88b5618f1f8f0a9b605bf7740b63b96", size = 141360, upload-time = "2026-07-10T05:50:24.183Z" } wheels = [ From a848fc7d2b4af924e75d4693f3f87572e83b135d Mon Sep 17 00:00:00 2001 From: sdk-sentinel-bot Date: Mon, 14 Sep 2026 19:37:04 -0400 Subject: [PATCH 3/9] Retry worker deployment test RPCs (#1857) --- tests/worker/test_worker.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 57614c21e..7aa666cd4 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -1281,7 +1281,8 @@ async def mk_call() -> DescribeWorkerDeploymentResponse: DescribeWorkerDeploymentRequest( namespace=client.namespace, deployment_name=version.deployment_name, - ) + ), + retry=True, ) except RPCError: # Expected @@ -1304,7 +1305,8 @@ async def set_current_deployment_version( deployment_name=version.deployment_name, version=version.to_canonical_string(), conflict_token=conflict_token, - ) + ), + retry=True, ) @@ -1321,7 +1323,8 @@ async def set_ramping_version( version=version.to_canonical_string(), conflict_token=conflict_token, percentage=percentage, - ) + ), + retry=True, ) return response @@ -1341,7 +1344,8 @@ async def check() -> bool: DescribeWorkerDeploymentRequest( namespace=client.namespace, deployment_name=deployment_name, - ) + ), + retry=True, ) routing_config = resp.worker_deployment_info.routing_config if ( From 98aa05d191f1860988dccb5ea90b73d25cc3c2c9 Mon Sep 17 00:00:00 2001 From: Frenchwood <46058503+JoshuaFrenchwood@users.noreply.github.com> Date: Mon, 14 Sep 2026 18:45:25 -0500 Subject: [PATCH 4/9] NexusSerializationContext for data/failure converters (#1828) * NexusSerializationContext for data/failure converters * Updating the nexus context for summary * Moving decoding logic to the nexus operation handle * Adding GetNexusOperationResultOutput for the interceptor * Refine Nexus serialization context propagation --- CHANGELOG.md | 6 + temporalio/client/_impl.py | 40 +- temporalio/client/_interceptor.py | 7 +- temporalio/client/_nexus.py | 7 +- temporalio/converter/__init__.py | 2 + .../converter/_serialization_context.py | 36 ++ temporalio/worker/_command_aware_visitor.py | 13 + temporalio/worker/_nexus.py | 55 ++- temporalio/worker/_workflow_instance.py | 25 +- tests/nexus/test_link_propagation.py | 6 + tests/nexus/test_standalone_operations.py | 20 +- tests/test_serialization_context.py | 438 ++++++++++++++++-- 12 files changed, 586 insertions(+), 69 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf2cab36e..bef2a3eed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,10 @@ to include examples, links to docs, or any other relevant information. - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`. - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome. - New properties and methods in ActivityExecution and ActivityExecutionDescription. +- Added experimental `temporalio.converter.NexusSerializationContext` support for Nexus callers + and handlers. Callers use it for inputs, results, and failures; handlers use it for inputs, + synchronous results, and failures. Asynchronous handler results and detached standalone handles + are not yet supported. Standalone `USE_EXISTING` handles use their start request's context. ### Changed @@ -53,6 +57,8 @@ to include examples, links to docs, or any other relevant information. - System Nexus Signal-with-Start Workflow operations now invoke `WorkflowOutboundInterceptor.start_system_nexus_operation` after their typed interception point. They continue not to invoke `WorkflowOutboundInterceptor.start_nexus_operation`. +- The experimental `GetNexusOperationResultInput` now includes the Nexus endpoint, service, and + operation. ### :boom: Breaking Changes diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index b5f6ab677..848975dc9 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -1549,6 +1549,12 @@ async def start_nexus_operation( self, input: StartNexusOperationInput ) -> NexusOperationHandle[Any]: """Start a nexus operation and return a handle to it.""" + nexus_context = temporalio.converter.NexusSerializationContext( + endpoint=input.endpoint, + service=input.service, + operation=input.operation, + ) + data_converter = self._client.data_converter.with_context(nexus_context) req = temporalio.api.workflowservice.v1.StartNexusOperationExecutionRequest( namespace=self._client.namespace, identity=self._client.identity, @@ -1575,7 +1581,7 @@ async def start_nexus_operation( req.start_to_close_timeout.FromTimedelta(input.start_to_close_timeout) # Set input payload - encoded = await self._client.data_converter.encode([input.arg]) + encoded = await data_converter.encode([input.arg]) if encoded: req.input.CopyFrom(encoded[0]) @@ -1620,6 +1626,7 @@ async def start_nexus_operation( result_type=input.result_type, endpoint=input.endpoint, service=input.service, + operation=input.operation, ) async def describe_nexus_operation( @@ -1637,15 +1644,31 @@ async def describe_nexus_operation( metadata=input.rpc_metadata, timeout=input.rpc_timeout, ) + data_converter = self._client.data_converter.with_context( + temporalio.converter.NexusSerializationContext( + endpoint=resp.info.endpoint, + service=resp.info.service, + operation=resp.info.operation, + ) + ) return await NexusOperationExecutionDescription._from_execution_info( info=resp.info, - data_converter=self._client.data_converter, + data_converter=data_converter, ) async def get_nexus_operation_result( self, input: GetNexusOperationResultInput ) -> Any: """Poll for nexus operation result until it's available.""" + data_converter = self._client.data_converter + if input.endpoint and input.service and input.operation: + data_converter = data_converter.with_context( + temporalio.converter.NexusSerializationContext( + endpoint=input.endpoint, + service=input.service, + operation=input.operation, + ) + ) req = temporalio.api.workflowservice.v1.PollNexusOperationExecutionRequest( namespace=self._client.namespace, operation_id=input.operation_id, @@ -1667,21 +1690,14 @@ async def get_nexus_operation_result( match res.WhichOneof("outcome"): case "result": type_hints = [input.result_type] if input.result_type else None - [result] = await self._client.data_converter.decode( - [res.result], type_hints - ) + [result] = await data_converter.decode([res.result], type_hints) return result - case "failure": raise NexusOperationFailureError( - cause=await self._client.data_converter.decode_failure( - res.failure - ) + cause=await data_converter.decode_failure(res.failure) ) - case None: - # poll again - pass + continue except RPCError as err: match err.status: case RPCStatusCode.DEADLINE_EXCEEDED: diff --git a/temporalio/client/_interceptor.py b/temporalio/client/_interceptor.py index fedec2cf6..68077ebc9 100644 --- a/temporalio/client/_interceptor.py +++ b/temporalio/client/_interceptor.py @@ -18,9 +18,7 @@ import temporalio.api.common.v1 import temporalio.api.workflowservice.v1 import temporalio.common -from temporalio.converter import ( - DataConverter, -) +from temporalio.converter import DataConverter if TYPE_CHECKING: from ._activity import ( @@ -633,6 +631,9 @@ class GetNexusOperationResultInput: operation_id: str run_id: str | None + endpoint: str + service: str + operation: str rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None result_type: type[Any] | None diff --git a/temporalio/client/_nexus.py b/temporalio/client/_nexus.py index 7eea155a9..1f0a7338e 100644 --- a/temporalio/client/_nexus.py +++ b/temporalio/client/_nexus.py @@ -1066,6 +1066,7 @@ def __init__( result_type: type | None = None, endpoint: str = "", service: str = "", + operation: str = "", ) -> None: """Create nexus operation handle.""" self._client = client @@ -1074,6 +1075,7 @@ def __init__( self._result_type = result_type self._endpoint = endpoint self._service = service + self._operation = operation # the default value is `_arg_unset` because ReturnType could be None self._known_outcome: ReturnType | NexusOperationFailureError | object = ( temporalio.common._arg_unset @@ -1136,9 +1138,12 @@ async def result( GetNexusOperationResultInput( operation_id=self._operation_id, run_id=self._run_id, - result_type=self._result_type, + endpoint=self._endpoint, + service=self._service, + operation=self._operation, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout, + result_type=self._result_type, ) ) ) diff --git a/temporalio/converter/__init__.py b/temporalio/converter/__init__.py index 99e55a775..324b477f2 100644 --- a/temporalio/converter/__init__.py +++ b/temporalio/converter/__init__.py @@ -50,6 +50,7 @@ ) from temporalio.converter._serialization_context import ( ActivitySerializationContext, + NexusSerializationContext, SerializationContext, WithSerializationContext, WorkflowSerializationContext, @@ -82,6 +83,7 @@ "JSONProtoPayloadConverter", "JSONTypeConverter", "JSONTypeConverterUnhandled", + "NexusSerializationContext", "PayloadCodec", "PayloadConverter", "SerializationContext", diff --git a/temporalio/converter/_serialization_context.py b/temporalio/converter/_serialization_context.py index 73a4a7104..8046a814c 100644 --- a/temporalio/converter/_serialization_context.py +++ b/temporalio/converter/_serialization_context.py @@ -28,6 +28,10 @@ class SerializationContext(ABC): context type is :py:class:`ActivitySerializationContext` and the workflow ID is that of the currently-executing workflow. ActivitySerializationContext is also set on data converter operations in the activity context. + + When operating on a Nexus operation payload, the context type is + :py:class:`NexusSerializationContext` and identifies the Nexus endpoint, service, and + resolved operation name. """ pass @@ -94,6 +98,38 @@ class ActivitySerializationContext(SerializationContext): """Whether the activity is a local activity started from a workflow.""" +@dataclass(frozen=True) +class NexusSerializationContext(SerializationContext): + """Serialization context for Nexus operation payloads. + + Callers receive this context when encoding inputs and decoding results or failures. The context + is not propagated to a handler that completes an asynchronous operation. Handlers receive it + when decoding inputs, encoding synchronous results, and encoding failures produced while + handling a Nexus task. + + A standalone operation handle retains the context used to start the operation and uses it to + decode the result, including when the start request returns an existing operation. A handle + created with :py:meth:`temporalio.client.Client.get_nexus_operation_handle` has no endpoint, + service, or operation information and therefore decodes without Nexus context. + + A failure encoded by a handler is later decoded by a caller. Because some operation paths may + lack this context, contextual encodings must be self-describing and decoders must continue to + accept payloads encoded without context. + + .. warning:: + This API is experimental and unstable. + """ + + endpoint: str + """Nexus endpoint name.""" + + service: str + """Nexus service name.""" + + operation: str + """Nexus operation name.""" + + class WithSerializationContext(ABC): """Interface for classes that can use serialization context. diff --git a/temporalio/worker/_command_aware_visitor.py b/temporalio/worker/_command_aware_visitor.py index 500fc4db5..7c03c2cd4 100644 --- a/temporalio/worker/_command_aware_visitor.py +++ b/temporalio/worker/_command_aware_visitor.py @@ -24,6 +24,7 @@ ScheduleNexusOperation, SignalExternalWorkflowExecution, StartChildWorkflowExecution, + WorkflowCommand, ) @@ -115,6 +116,18 @@ async def _visit_coresdk_workflow_commands_ScheduleNexusOperation( with current_command(CommandType.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, o.seq): await super()._visit_coresdk_workflow_commands_ScheduleNexusOperation(fs, o) + async def _visit_coresdk_workflow_commands_WorkflowCommand( + self, fs: VisitorFunctions, o: WorkflowCommand + ) -> None: + if o.HasField("schedule_nexus_operation"): + with current_command( + CommandType.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, + o.schedule_nexus_operation.seq, + ): + await super()._visit_coresdk_workflow_commands_WorkflowCommand(fs, o) + else: + await super()._visit_coresdk_workflow_commands_WorkflowCommand(fs, o) + # Workflow activation jobs with payloads async def _visit_coresdk_workflow_activation_ResolveActivity( self, fs: VisitorFunctions, o: ResolveActivity diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index 90ba40382..a2f4b8ca7 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -230,18 +230,31 @@ async def _complete_task( await asyncio.shield(self._bridge_worker().complete_nexus_task(completion)) async def _encode_completion( - self, completion: temporalio.bridge.proto.nexus.NexusTaskCompletion + self, + completion: temporalio.bridge.proto.nexus.NexusTaskCompletion, + data_converter: temporalio.converter.DataConverter, ) -> None: """Apply the payload codec then external storage to the completion's payloads.""" - dc = self._data_converter await PayloadVisitor(skip_search_attributes=True, skip_headers=True).visit( - _PayloadTransformVisitor(dc._encode_payload_sequence), completion + _PayloadTransformVisitor(data_converter._encode_payload_sequence), + completion, ) await PayloadVisitor(skip_search_attributes=True).visit( - _PayloadTransformVisitor(dc._external_store_payload_sequence), + _PayloadTransformVisitor(data_converter._external_store_payload_sequence), completion, ) + def _data_converter_for_nexus_task( + self, endpoint: str, service: str, operation: str + ) -> temporalio.converter.DataConverter: + return self._data_converter.with_context( + temporalio.converter.NexusSerializationContext( + endpoint=endpoint, + service=service, + operation=operation, + ) + ) + # TODO(nexus-preview): stack trace pruning. See sdk-typescript NexusHandler.execute # "Any call up to this function and including this one will be trimmed out of stack traces."" @@ -272,6 +285,9 @@ async def _handle_cancel_operation_task( task_cancellation=task_cancellation, request_deadline=request_deadline, ) + data_converter = self._data_converter_for_nexus_task( + endpoint, request.service, request.operation + ) temporalio.nexus._operation_context._TemporalCancelOperationContext( info=lambda: Info( endpoint=endpoint, @@ -293,7 +309,7 @@ async def _handle_cancel_operation_task( ), ) # No-op but keeps the cancel covered if it ever carries a payload. - await self._encode_completion(completion) + await self._encode_completion(completion, data_converter) except asyncio.CancelledError: completion = temporalio.bridge.proto.nexus.NexusTaskCompletion( task_token=task_token, @@ -305,12 +321,12 @@ async def _handle_cancel_operation_task( completion = temporalio.bridge.proto.nexus.NexusTaskCompletion( task_token=task_token, ) - self._data_converter.failure_converter.to_failure( + data_converter.failure_converter.to_failure( handler_error, - self._data_converter.payload_converter, + data_converter.payload_converter, completion.failure, ) - await self._encode_completion(completion) + await self._encode_completion(completion, data_converter) await self._complete_task(completion) except Exception: logger.exception("Failed to send Nexus task completion") @@ -336,6 +352,9 @@ async def _handle_start_operation_task( Attempt to execute the user start_operation method and invoke the data converter on the result. Handle errors and send the task completion. """ + data_converter = self._data_converter_for_nexus_task( + endpoint, start_request.service, start_request.operation + ) try: try: start_response = await self._start_operation( @@ -344,6 +363,7 @@ async def _handle_start_operation_task( task_cancellation, request_deadline, endpoint, + data_converter, ) completion = temporalio.bridge.proto.nexus.NexusTaskCompletion( task_token=task_token, @@ -351,7 +371,7 @@ async def _handle_start_operation_task( start_operation=start_response ), ) - await self._encode_completion(completion) + await self._encode_completion(completion, data_converter) except asyncio.CancelledError: completion = temporalio.bridge.proto.nexus.NexusTaskCompletion( task_token=task_token, @@ -363,15 +383,15 @@ async def _handle_start_operation_task( task_token=task_token, ) handler_error = _exception_to_handler_error(err) - self._data_converter.failure_converter.to_failure( + data_converter.failure_converter.to_failure( handler_error, - self._data_converter.payload_converter, + data_converter.payload_converter, completion.failure, ) if isinstance(err, concurrent.futures.BrokenExecutor): self._fail_worker_exception_queue.put_nowait(err) - await self._encode_completion(completion) + await self._encode_completion(completion, data_converter) await self._complete_task(completion) except Exception: @@ -391,6 +411,7 @@ async def _start_operation( cancellation: nexusrpc.handler.OperationTaskCancellation, request_deadline: datetime | None, endpoint: str, + data_converter: temporalio.converter.DataConverter, ) -> temporalio.api.nexus.v1.StartOperationResponse: """Invoke the Nexus handler's start_operation method and construct the StartOperationResponse. @@ -430,7 +451,7 @@ async def _start_operation( ).set() input = LazyValue( serializer=_NexusPayloadSerializer( - data_converter=self._data_converter, + data_converter=data_converter, payload=start_request.payload, ), headers={}, @@ -450,9 +471,7 @@ async def _start_operation( ) ) elif isinstance(result, nexusrpc.handler.StartOperationResultSync): - [payload] = self._data_converter.payload_converter.to_payloads( - [result.value] - ) + [payload] = data_converter.payload_converter.to_payloads([result.value]) return temporalio.api.nexus.v1.StartOperationResponse( sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync( payload=payload, @@ -481,9 +500,9 @@ async def _start_operation( ) from err.__cause__ except FailureError as new_err: response = temporalio.api.nexus.v1.StartOperationResponse() - self._data_converter.failure_converter.to_failure( + data_converter.failure_converter.to_failure( new_err, - self._data_converter.payload_converter, + data_converter.payload_converter, response.failure, ) return response diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 80d77f103..89625d64c 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2194,14 +2194,26 @@ async def operation_handle_fn() -> OutputT: user_payload_converter, user_failure_converter, ) + failure_converter = user_failure_converter else: - payload_converter = self._context_free_payload_converter + serialization_context = temporalio.converter.NexusSerializationContext( + endpoint=input.endpoint, + service=input.service, + operation=input.operation_name, + ) + payload_converter = self._payload_converter_with_context( + serialization_context + ) + failure_converter = self._failure_converter_with_context( + serialization_context + ) handle = _NexusOperationHandle( self, self._next_seq("nexus_operation"), input, operation_handle_fn(), payload_converter, + failure_converter, ) handle._apply_schedule_command() self._pending_nexus_operations[handle._seq] = handle @@ -2454,9 +2466,11 @@ def get_serialization_context( nexus_operation._input.operation_name, nexus_operation._input.input, ) - # Other Nexus operations have no context because the caller workflow context is - # unavailable on the handler side for decryption. - return None + return temporalio.converter.NexusSerializationContext( + endpoint=nexus_operation._input.endpoint, + service=nexus_operation._input.service, + operation=nexus_operation._input.operation_name, + ) else: # Use payload codec with workflow context for all other payloads @@ -3648,6 +3662,7 @@ def __init__( input: StartNexusOperationInput[Any, OutputT], fn: Coroutine[Any, Any, OutputT], payload_converter: temporalio.converter.PayloadConverter, + failure_converter: temporalio.converter.FailureConverter, ): self._instance = instance self._seq = seq @@ -3656,7 +3671,7 @@ def __init__( self._start_fut: asyncio.Future[str | None] = instance.create_future() self._result_fut: asyncio.Future[OutputT | None] = instance.create_future() self._payload_converter = payload_converter - self._failure_converter = self._instance._context_free_failure_converter + self._failure_converter = failure_converter @property def operation_token(self) -> str | None: diff --git a/tests/nexus/test_link_propagation.py b/tests/nexus/test_link_propagation.py index 554620b95..f656f57c1 100644 --- a/tests/nexus/test_link_propagation.py +++ b/tests/nexus/test_link_propagation.py @@ -884,6 +884,9 @@ async def test_sync_response_includes_signal_backlinks() -> None: cancellation=_NexusTaskCancellation(), request_deadline=None, endpoint="endpoint", + data_converter=worker._data_converter_for_nexus_task( + "endpoint", "_BacklinkStashingService", "sync_op" + ), ) assert response.HasField("sync_success") assert len(response.sync_success.links) == 1 @@ -898,6 +901,9 @@ async def test_async_response_includes_signal_backlinks() -> None: cancellation=_NexusTaskCancellation(), request_deadline=None, endpoint="endpoint", + data_converter=worker._data_converter_for_nexus_task( + "endpoint", "_BacklinkStashingService", "async_op" + ), ) assert response.HasField("async_success") assert response.async_success.operation_token == "op-token" diff --git a/tests/nexus/test_standalone_operations.py b/tests/nexus/test_standalone_operations.py index 26a8316b4..f12a7fd85 100644 --- a/tests/nexus/test_standalone_operations.py +++ b/tests/nexus/test_standalone_operations.py @@ -869,7 +869,9 @@ async def get_nexus_operation_result( self, input: GetNexusOperationResultInput ) -> Any: self._parent.result_calls.append(input) - return await super().get_nexus_operation_result(input) + result = await super().get_nexus_operation_result(input) + self._parent.result_outputs.append(result) + return result async def cancel_nexus_operation(self, input: CancelNexusOperationInput) -> None: self._parent.cancel_calls.append(input) @@ -898,6 +900,7 @@ def __init__(self) -> None: self.start_calls: list[StartNexusOperationInput] = [] self.describe_calls: list[DescribeNexusOperationInput] = [] self.result_calls: list[GetNexusOperationResultInput] = [] + self.result_outputs: list[Any] = [] self.cancel_calls: list[CancelNexusOperationInput] = [] self.terminate_calls: list[TerminateNexusOperationInput] = [] self.list_calls: list[ListNexusOperationsInput] = [] @@ -981,6 +984,21 @@ async def test_interceptor_receives_inputs(client: Client, env: WorkflowEnvironm assert isinstance(result_input, GetNexusOperationResultInput) assert result_input.operation_id == op_id assert result_input.result_type == EchoOutput + assert result_input.endpoint == endpoint_name + assert result_input.service == "StandaloneTestService" + assert result_input.operation == "blocking_async" + + # Interceptors receive successfully decoded results. + value = f"interceptor-success-{uuid.uuid4()}" + handle = await nexus_client.start_operation( + StandaloneTestService.echo_sync, + EchoInput(value=value), + id=str(uuid.uuid4()), + schedule_to_close_timeout=timedelta(seconds=30), + ) + result = await handle.result() + assert result == EchoOutput(value=value) + assert interceptor.result_outputs == [EchoOutput(value=value)] # Start another so we can terminate it previous_start_count = len(interceptor.start_calls) diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 8d65d5f1f..793663ca9 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -28,6 +28,10 @@ from temporalio.client import ( AsyncActivityHandle, Client, + GetNexusOperationResultInput, + Interceptor, + NexusOperationFailureError, + OutboundInterceptor, WorkflowFailureError, WorkflowUpdateFailedError, ) @@ -41,17 +45,17 @@ DefaultPayloadConverter, EncodingPayloadConverter, JSONPlainPayloadConverter, + NexusSerializationContext, PayloadCodec, PayloadConverter, SerializationContext, WithSerializationContext, WorkflowSerializationContext, ) -from temporalio.exceptions import ApplicationError +from temporalio.exceptions import ApplicationError, NexusOperationError from temporalio.testing import WorkflowEnvironment -from temporalio.worker import Worker +from temporalio.worker import Replayer, Worker from temporalio.worker._workflow_instance import UnsandboxedWorkflowRunner -from tests.helpers.nexus import make_nexus_endpoint_name @dataclass @@ -1688,25 +1692,88 @@ async def test_decode_context_matches_encode_context( # Test nexus payload codec -class AssertNexusLacksContextPayloadCodec(PayloadCodec, WithSerializationContext): - def __init__(self): - self.context = None +class _NexusResultDecodingInterceptor(Interceptor): + def __init__(self) -> None: + super().__init__() + self.decoded_results: list[Any] = [] + + def intercept_client(self, next: OutboundInterceptor) -> OutboundInterceptor: + return _NexusResultDecodingOutboundInterceptor(next, self) + + +class _NexusResultDecodingOutboundInterceptor(OutboundInterceptor): + def __init__( + self, next: OutboundInterceptor, parent: _NexusResultDecodingInterceptor + ) -> None: + super().__init__(next) + self._parent = parent + + async def get_nexus_operation_result( + self, input: GetNexusOperationResultInput + ) -> Any: + result = await super().get_nexus_operation_result(input) + self._parent.decoded_results.append(result) + return result + + +class NexusContextMarkerPayloadCodec(PayloadCodec, WithSerializationContext): + MARKER_KEY = "nexus-context-marker" + + def __init__( + self, + markers: dict[NexusSerializationContext, bytes], + context: SerializationContext | None = None, + ): + self.markers = markers + self.context = context def with_context( self, context: SerializationContext - ) -> AssertNexusLacksContextPayloadCodec: - codec = AssertNexusLacksContextPayloadCodec() - codec.context = context - return codec + ) -> NexusContextMarkerPayloadCodec: + return NexusContextMarkerPayloadCodec(self.markers, context) + + def _marker(self) -> bytes | None: + if not isinstance(self.context, NexusSerializationContext): + return None + try: + return self.markers[self.context] + except KeyError: + raise AssertionError( + f"No Nexus payload codec configured for {self.context!r}" + ) from None - async def _assert_context_iff_not_nexus( + async def encode( self, payloads: Sequence[temporalio.api.common.v1.Payload] ) -> list[temporalio.api.common.v1.Payload]: - [payload] = payloads - assert bool(self.context) == (payload.data.decode() != '"nexus-data"') - return list(payloads) + marker = self._marker() + if marker is None: + return list(payloads) + encoded = [] + for payload in payloads: + marked_payload = temporalio.api.common.v1.Payload() + marked_payload.CopyFrom(payload) + marked_payload.metadata[self.MARKER_KEY] = marker + encoded.append(marked_payload) + return encoded - encode = decode = _assert_context_iff_not_nexus + async def decode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + marker = self._marker() + if marker is None: + return list(payloads) + decoded = [] + for payload in payloads: + actual_marker = payload.metadata.get(self.MARKER_KEY) + if actual_marker is None: + decoded.append(payload) + continue + assert actual_marker == marker + decoded_payload = temporalio.api.common.v1.Payload() + decoded_payload.CopyFrom(payload) + del decoded_payload.metadata[self.MARKER_KEY] + decoded.append(decoded_payload) + return decoded @nexusrpc.handler.service_handler @@ -1717,52 +1784,365 @@ async def operation( ) -> str: return data + @nexusrpc.handler.sync_operation + async def fail(self, _: nexusrpc.handler.StartOperationContext, data: str) -> str: + raise ApplicationError(data, non_retryable=True) + @workflow.defn class NexusOperationTestWorkflow: @workflow.run - async def run(self, _data: str) -> None: + async def run(self, red_endpoint_name: str, blue_endpoint_name: str) -> list[str]: + red_handle, blue_handle = await asyncio.gather( + workflow.create_nexus_client( + service=NexusOperationTestServiceHandler, + endpoint=red_endpoint_name, + ).start_operation( + NexusOperationTestServiceHandler.operation, + input="nexus-data", + summary="nexus-summary", + ), + workflow.create_nexus_client( + service=NexusOperationTestServiceHandler, + endpoint=blue_endpoint_name, + ).start_operation( + NexusOperationTestServiceHandler.operation, + input="nexus-data", + summary="nexus-summary", + ), + ) + return list(await asyncio.gather(red_handle, blue_handle)) + + +@workflow.defn +class NexusOperationFailureTestWorkflow: + @workflow.run + async def run(self, endpoint_name: str) -> None: nexus_client = workflow.create_nexus_client( service=NexusOperationTestServiceHandler, - endpoint=make_nexus_endpoint_name(workflow.info().task_queue), - ) - await nexus_client.start_operation( - NexusOperationTestServiceHandler.operation, input="nexus-data" + endpoint=endpoint_name, ) + try: + await nexus_client.start_operation( + NexusOperationTestServiceHandler.fail, input="nexus-failure" + ) + except NexusOperationError: + return + raise AssertionError("Nexus operation should have failed") + + +nexus_failure_context_traces: list[tuple[str, NexusSerializationContext]] = [] + + +class NexusFailureConverterWithContext( + DefaultFailureConverter, WithSerializationContext +): + def __init__(self, context: SerializationContext | None = None): + super().__init__() + self.context = context + + def with_context( + self, context: SerializationContext + ) -> NexusFailureConverterWithContext: + return NexusFailureConverterWithContext(context) + + def to_failure( + self, + exception: BaseException, + payload_converter: PayloadConverter, + failure: temporalio.api.failure.v1.Failure, + ) -> None: + if isinstance(self.context, NexusSerializationContext): + nexus_failure_context_traces.append(("to_failure", self.context)) + super().to_failure(exception, payload_converter, failure) + + def from_failure( + self, + failure: temporalio.api.failure.v1.Failure, + payload_converter: PayloadConverter, + ) -> BaseException: + if isinstance(self.context, NexusSerializationContext): + nexus_failure_context_traces.append(("from_failure", self.context)) + return super().from_failure(failure, payload_converter) @pytest.mark.requires_local_server -async def test_nexus_payload_codec_operations_lack_context( +async def test_workflow_nexus_payload_codec_receives_context( env: WorkflowEnvironment, ): - """ - encode() and decode() on nexus payloads should not have any context set. - """ + """Nexus payload codecs get context for workflow inputs, summaries, and results.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with the Java test server") + task_queue = "workflow-nexus-context-codec-task-queue" + red_endpoint_name = "workflow-red-nexus-endpoint" + blue_endpoint_name = "workflow-blue-nexus-endpoint" + red_context = NexusSerializationContext( + endpoint=red_endpoint_name, + service="NexusOperationTestServiceHandler", + operation="operation", + ) + blue_context = NexusSerializationContext( + endpoint=blue_endpoint_name, + service="NexusOperationTestServiceHandler", + operation="operation", + ) + payload_codec = NexusContextMarkerPayloadCodec( + {red_context: b"red", blue_context: b"blue"} + ) config = env.client.config() config["data_converter"] = dataclasses.replace( DataConverter.default, - payload_codec=AssertNexusLacksContextPayloadCodec(), + payload_codec=payload_codec, ) client = Client(**config) async with Worker( client, - task_queue=str(uuid.uuid4()), + task_queue=task_queue, workflows=[NexusOperationTestWorkflow], nexus_service_handlers=[NexusOperationTestServiceHandler()], ) as worker: - endpoint_name = make_nexus_endpoint_name(worker.task_queue) + await env.create_nexus_endpoint(red_endpoint_name, worker.task_queue) + await env.create_nexus_endpoint(blue_endpoint_name, worker.task_queue) + handle = await client.start_workflow( + NexusOperationTestWorkflow.run, + args=[red_endpoint_name, blue_endpoint_name], + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + ) + assert await handle.result() == ["nexus-data", "nexus-data"] + + history = await handle.fetch_history() + scheduled_endpoints: dict[int, str] = {} + encoded_summaries: dict[str, temporalio.api.common.v1.Payload] = {} + encoded_results: dict[str, temporalio.api.common.v1.Payload] = {} + for event in history.events: + if event.HasField("nexus_operation_scheduled_event_attributes"): + scheduled_attrs = event.nexus_operation_scheduled_event_attributes + assert scheduled_attrs.service == "NexusOperationTestServiceHandler" + assert scheduled_attrs.operation == "operation" + scheduled_endpoints[event.event_id] = scheduled_attrs.endpoint + assert event.HasField("user_metadata") + assert event.user_metadata.HasField("summary") + encoded_summaries[scheduled_attrs.endpoint] = ( + event.user_metadata.summary + ) + elif event.HasField("nexus_operation_completed_event_attributes"): + completed_attrs = event.nexus_operation_completed_event_attributes + endpoint = scheduled_endpoints[completed_attrs.scheduled_event_id] + encoded_results[endpoint] = completed_attrs.result + assert set(scheduled_endpoints.values()) == { + red_endpoint_name, + blue_endpoint_name, + } + assert { + endpoint: payload.metadata[NexusContextMarkerPayloadCodec.MARKER_KEY] + for endpoint, payload in encoded_summaries.items() + } == { + red_endpoint_name: b"red", + blue_endpoint_name: b"blue", + } + assert { + endpoint: payload.metadata[NexusContextMarkerPayloadCodec.MARKER_KEY] + for endpoint, payload in encoded_results.items() + } == { + red_endpoint_name: b"red", + blue_endpoint_name: b"blue", + } + + scheduled_contexts: dict[int, NexusSerializationContext] = {} + for event in history.events: + if event.HasField("nexus_operation_scheduled_event_attributes"): + scheduled_attrs = event.nexus_operation_scheduled_event_attributes + context = NexusSerializationContext( + endpoint=scheduled_attrs.endpoint, + service=scheduled_attrs.service, + operation=scheduled_attrs.operation, + ) + scheduled_contexts[event.event_id] = context + [decoded] = await payload_codec.with_context(context).decode( + [scheduled_attrs.input] + ) + scheduled_attrs.input.CopyFrom(decoded) + [decoded] = await payload_codec.with_context(context).decode( + [event.user_metadata.summary] + ) + event.user_metadata.summary.CopyFrom(decoded) + elif event.HasField("nexus_operation_completed_event_attributes"): + completed_attrs = event.nexus_operation_completed_event_attributes + context = scheduled_contexts[completed_attrs.scheduled_event_id] + [decoded] = await payload_codec.with_context(context).decode( + [completed_attrs.result] + ) + completed_attrs.result.CopyFrom(decoded) + await Replayer( + workflows=[NexusOperationTestWorkflow], + data_converter=config["data_converter"], + ).replay_workflow(history) + + +@pytest.mark.requires_local_server +async def test_standalone_nexus_payload_codec_receives_context( + env: WorkflowEnvironment, +): + """Nexus payload codecs get context for standalone inputs and results.""" + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + task_queue = "standalone-nexus-context-codec-task-queue" + red_endpoint_name = "standalone-red-nexus-endpoint" + blue_endpoint_name = "standalone-blue-nexus-endpoint" + red_context = NexusSerializationContext( + endpoint=red_endpoint_name, + service="NexusOperationTestServiceHandler", + operation="operation", + ) + blue_context = NexusSerializationContext( + endpoint=blue_endpoint_name, + service="NexusOperationTestServiceHandler", + operation="operation", + ) + config = env.client.config() + config["data_converter"] = dataclasses.replace( + DataConverter.default, + payload_codec=NexusContextMarkerPayloadCodec( + {red_context: b"red", blue_context: b"blue"} + ), + ) + result_interceptor = _NexusResultDecodingInterceptor() + config["interceptors"] = list(config.get("interceptors") or []) + [ + result_interceptor + ] + client = Client(**config) + + async with Worker( + client, + task_queue=task_queue, + nexus_service_handlers=[NexusOperationTestServiceHandler()], + ) as worker: + await env.create_nexus_endpoint(red_endpoint_name, worker.task_queue) + await env.create_nexus_endpoint(blue_endpoint_name, worker.task_queue) + red_standalone_result, blue_standalone_result = await asyncio.gather( + client.create_nexus_client( + service=NexusOperationTestServiceHandler, + endpoint=red_endpoint_name, + ).execute_operation( + NexusOperationTestServiceHandler.operation, + "standalone-red", + id=str(uuid.uuid4()), + schedule_to_close_timeout=timedelta(seconds=10), + ), + client.create_nexus_client( + service=NexusOperationTestServiceHandler, + endpoint=blue_endpoint_name, + ).execute_operation( + NexusOperationTestServiceHandler.operation, + "standalone-blue", + id=str(uuid.uuid4()), + schedule_to_close_timeout=timedelta(seconds=10), + ), + ) + assert red_standalone_result == "standalone-red" + assert blue_standalone_result == "standalone-blue" + assert set(result_interceptor.decoded_results) == { + "standalone-red", + "standalone-blue", + } + + +@pytest.mark.requires_local_server +async def test_workflow_nexus_failure_converter_has_context( + env: WorkflowEnvironment, +): + """Workflow Nexus callers and handlers use context for failures.""" + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + nexus_failure_context_traces.clear() + task_queue = "workflow-nexus-failure-context-task-queue" + endpoint_name = "workflow-failure-nexus-endpoint" + expected_context = NexusSerializationContext( + endpoint=endpoint_name, + service="NexusOperationTestServiceHandler", + operation="fail", + ) + config = env.client.config() + config["data_converter"] = dataclasses.replace( + DataConverter.default, + failure_converter_class=NexusFailureConverterWithContext, + ) + client = Client(**config) + + async with Worker( + client, + task_queue=task_queue, + workflows=[NexusOperationFailureTestWorkflow], + nexus_service_handlers=[NexusOperationTestServiceHandler()], + workflow_runner=UnsandboxedWorkflowRunner(), + ) as worker: await env.create_nexus_endpoint(endpoint_name, worker.task_queue) await client.execute_workflow( - NexusOperationTestWorkflow.run, - "workflow-data", + NexusOperationFailureTestWorkflow.run, + endpoint_name, id=str(uuid.uuid4()), task_queue=worker.task_queue, ) + assert ("to_failure", expected_context) in nexus_failure_context_traces + assert ("from_failure", expected_context) in nexus_failure_context_traces + + +@pytest.mark.requires_local_server +async def test_standalone_nexus_failure_converter_has_context( + env: WorkflowEnvironment, +): + """Standalone Nexus callers and handlers use context for failures.""" + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + nexus_failure_context_traces.clear() + task_queue = "standalone-nexus-failure-context-task-queue" + endpoint_name = "standalone-failure-nexus-endpoint" + expected_context = NexusSerializationContext( + endpoint=endpoint_name, + service="NexusOperationTestServiceHandler", + operation="fail", + ) + config = env.client.config() + config["data_converter"] = dataclasses.replace( + DataConverter.default, + failure_converter_class=NexusFailureConverterWithContext, + ) + client = Client(**config) + + async with Worker( + client, + task_queue=task_queue, + nexus_service_handlers=[NexusOperationTestServiceHandler()], + ) as worker: + await env.create_nexus_endpoint(endpoint_name, worker.task_queue) + nexus_client = client.create_nexus_client( + service=NexusOperationTestServiceHandler, + endpoint=endpoint_name, + ) + operation_handle = await nexus_client.start_operation( + NexusOperationTestServiceHandler.fail, + "nexus-failure", + id=str(uuid.uuid4()), + schedule_to_close_timeout=timedelta(seconds=10), + ) + with pytest.raises(NexusOperationFailureError): + await operation_handle.result() + + assert ("to_failure", expected_context) in nexus_failure_context_traces + assert ("from_failure", expected_context) in nexus_failure_context_traces + + nexus_failure_context_traces.clear() + description = await operation_handle.describe() + assert description.last_attempt_failure is not None + assert ("from_failure", expected_context) in nexus_failure_context_traces + # Test pydantic converter with context From b5a2bef1522da1825d931821a8b99e6492e2eac2 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 14 Sep 2026 17:34:56 -0700 Subject: [PATCH 5/9] Support google-genai 2.21 file downloads (#1865) --- CHANGELOG.md | 3 + pyproject.toml | 2 +- .../contrib/google_genai/_gemini_activity.py | 6 +- temporalio/contrib/google_genai/_models.py | 1 + .../google_genai/_temporal_async_client.py | 8 ++- .../contrib/google_genai/_temporal_files.py | 64 ++++++++++++++++--- tests/contrib/google_genai/test_gemini.py | 28 ++++++++ uv.lock | 8 +-- 8 files changed, 104 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bef2a3eed..87dafcbe2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -77,6 +77,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- `temporalio.contrib.google_genai` now requires `google-genai` 2.21.0 or later + and supports its file download API, including video inputs and download + destinations. - `temporalio.contrib.deepagents` no longer dedups repeated identical tool, model, and backend-op calls: each dispatch runs its own Activity, and the continue-as-new result cache is retired for new executions (a continued run diff --git a/pyproject.toml b/pyproject.toml index fc2d3d17c..b7077785f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,7 +55,7 @@ cloud-run-worker-otel = [ "protobuf<7", ] aioboto3 = ["aioboto3>=10.4.0", "types-aioboto3[s3]>=10.4.0"] -google-genai = ["google-genai>=2.10.0,<3.0.0"] +google-genai = ["google-genai>=2.21.0,<3.0.0"] strands-agents = ["strands-agents>=1.39.0"] [project.urls] diff --git a/temporalio/contrib/google_genai/_gemini_activity.py b/temporalio/contrib/google_genai/_gemini_activity.py index 496bbf1ab..dde83bfff 100644 --- a/temporalio/contrib/google_genai/_gemini_activity.py +++ b/temporalio/contrib/google_genai/_gemini_activity.py @@ -198,10 +198,12 @@ async def gemini_files_upload( @activity.defn async def gemini_files_download( req: _GeminiDownloadFileRequest, - ) -> bytes: + ) -> bytes | None: """Download a file using the real genai.Client on the worker.""" return await self._client.aio.files.download( - file=req.file, config=req.config + file=req.file, + destination=req.destination, + config=req.config, ) @activity.defn diff --git a/temporalio/contrib/google_genai/_models.py b/temporalio/contrib/google_genai/_models.py index 2f70d9b4d..21caa8d39 100644 --- a/temporalio/contrib/google_genai/_models.py +++ b/temporalio/contrib/google_genai/_models.py @@ -100,6 +100,7 @@ class _GeminiDownloadFileRequest(BaseModel): """Serializable activity input for a file download.""" file: str + destination: str | None = None config: types.DownloadFileConfig | None = None diff --git a/temporalio/contrib/google_genai/_temporal_async_client.py b/temporalio/contrib/google_genai/_temporal_async_client.py index c77cd5cee..c9abfd988 100644 --- a/temporalio/contrib/google_genai/_temporal_async_client.py +++ b/temporalio/contrib/google_genai/_temporal_async_client.py @@ -255,13 +255,19 @@ def __init__( # Closure-wrap bound-method tools so google-genai's internal # config deep-copy (>= 2.8.0) can't clone the workflow instance. self._models = _TemporalAsyncModels(api_client) - self._files = TemporalAsyncFiles(api_client, activity_config) + self._temporal_files = TemporalAsyncFiles(api_client, activity_config) + self._files = self._temporal_files self._file_search_stores = TemporalAsyncFileSearchStores( api_client, activity_config ) self._temporal_interactions = TemporalAsyncInteractions(activity_config) self._temporal_agents = TemporalAsyncAgents(activity_config) + @property + def files(self) -> TemporalAsyncFiles: + """Temporal-aware files resource; operations run as activities.""" + return self._temporal_files + @property def interactions( # type: ignore[override] self, diff --git a/temporalio/contrib/google_genai/_temporal_files.py b/temporalio/contrib/google_genai/_temporal_files.py index f785c00cb..c2f3a3f53 100644 --- a/temporalio/contrib/google_genai/_temporal_files.py +++ b/temporalio/contrib/google_genai/_temporal_files.py @@ -11,7 +11,7 @@ import io import os from datetime import timedelta -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, overload if TYPE_CHECKING: import google.auth.credentials @@ -100,12 +100,31 @@ async def upload( **act_config, ) + @overload async def download( self, *, - file: str | types.File, + file: str | types.File | types.Video | types.GeneratedVideo, + destination: None = None, config: types.DownloadFileConfigOrDict | None = None, - ) -> bytes: + ) -> bytes: ... + + @overload + async def download( + self, + *, + file: str | types.File | types.Video | types.GeneratedVideo, + destination: str | os.PathLike[str] | io.IOBase, + config: types.DownloadFileConfigOrDict | None = None, + ) -> None: ... + + async def download( + self, + *, + file: str | types.File | types.Video | types.GeneratedVideo, + destination: str | os.PathLike[str] | io.IOBase | None = None, + config: types.DownloadFileConfigOrDict | None = None, + ) -> bytes | None: """Download a file via a Temporal activity.""" act_config: ActivityConfig = {**self._activity_config} if "summary" not in act_config: @@ -119,19 +138,48 @@ async def download( download_config = config _validate_http_options(download_config.http_options) - if isinstance(file, types.File): - if not file.name: - raise ValueError("File object must have a name to download.") + if isinstance(file, types.GeneratedVideo): + file_video = file.video + file_name = file_video.uri if file_video is not None else None + elif isinstance(file, types.Video): + file_name = file.uri + elif isinstance(file, types.File): file_name = file.name else: file_name = file + if not file_name: + raise ValueError("File name is required.") + + if isinstance(destination, io.IOBase): + destination_stream = destination + destination_path = None + elif destination is not None: + destination_stream = None + destination_path = os.fspath(destination) + else: + destination_stream = None + destination_path = None - return await temporal_workflow.execute_activity( + data = await temporal_workflow.execute_activity( "gemini_files_download", - _GeminiDownloadFileRequest(file=file_name, config=download_config), + _GeminiDownloadFileRequest( + file=file_name, + destination=destination_path, + config=download_config, + ), result_type=bytes, **act_config, ) + if destination_stream is not None: + destination_stream.write(data) + return None + if destination_path is not None: + return None + if isinstance(file, types.Video): + file.video_bytes = data + elif isinstance(file, types.GeneratedVideo) and file.video is not None: + file.video.video_bytes = data + return data async def register_files( self, diff --git a/tests/contrib/google_genai/test_gemini.py b/tests/contrib/google_genai/test_gemini.py index 60b717840..68e7dd0fc 100644 --- a/tests/contrib/google_genai/test_gemini.py +++ b/tests/contrib/google_genai/test_gemini.py @@ -742,6 +742,16 @@ async def run(self, file_name: str) -> bytes: return await client.files.download(file=file_name) +@workflow.defn +class FileDownloadToPathWorkflow: + """Workflow that downloads a file to a path on the activity worker.""" + + @workflow.run + async def run(self, file_name: str, destination: str) -> None: + client = TemporalAsyncClient() + await client.files.download(file=file_name, destination=destination) + + @workflow.defn class FileSearchStoreUploadWorkflow: """Workflow that uploads to a file search store.""" @@ -1242,6 +1252,24 @@ async def test_file_download(client: Client): assert result == b"fake file content" +async def test_file_download_to_path(client: Client): + """Download destinations are passed to the activity worker.""" + new_client, api_tracker = apply_plugin(client, []) + + async with new_worker(new_client, FileDownloadToPathWorkflow) as worker: + await new_client.execute_workflow( + FileDownloadToPathWorkflow.run, + args=["files/some-file", "/tmp/downloaded-file"], + id=f"gemini-file-download-to-path-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + + assert len(api_tracker.file_download_requests) == 1 + request = api_tracker.file_download_requests[0] + assert request.file == "files/some-file" + assert request.destination == "/tmp/downloaded-file" + + # =========================================================================== # File search store upload tests # =========================================================================== diff --git a/uv.lock b/uv.lock index 8dce7d158..ab1c17005 100644 --- a/uv.lock +++ b/uv.lock @@ -1354,7 +1354,7 @@ requests = [ [[package]] name = "google-genai" -version = "2.11.0" +version = "2.21.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, @@ -1368,9 +1368,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a6/01/e7b5f3aac89200c78318ed7643401e7f5ed3131b0cd353c07483606b1e61/google_genai-2.11.0.tar.gz", hash = "sha256:4c5e524d24b145c96be327f9a7f8f04b0fe4efee0533877795e9848afed01749", size = 622366, upload-time = "2026-07-09T17:49:43.862Z" } +sdist = { url = "https://files.pythonhosted.org/packages/60/a7/a45f64f22ab9302b55fcbeb32acb6f313690a7748629b01e451aad1817a3/google_genai-2.21.0.tar.gz", hash = "sha256:0ecc11c6a5b9f5e3cc58e77ae5fead00c6719f8a1b2b654b803f514a9a6b64c0", size = 677301, upload-time = "2026-08-31T21:49:14.508Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/93/ef/d296c23390160a8b0b1dafb36dd3cb36a39ed40c81cd27e04e6233334186/google_genai-2.11.0-py3-none-any.whl", hash = "sha256:5bc8186100e1d34d691fbe0cba392b7e04e98d286ca952323a6672d054accf95", size = 984162, upload-time = "2026-07-09T17:49:42.15Z" }, + { url = "https://files.pythonhosted.org/packages/fb/d7/c2419dd5fedd5803ce09810e4e39561a0a13a960b3f48d2581c12e42af3e/google_genai-2.21.0-py3-none-any.whl", hash = "sha256:36b575034be46a03acd603a852e22a6359f2cdd6b26bb1d65d9b7e0cc7ab3648", size = 1080223, upload-time = "2026-08-31T21:49:12.699Z" }, ] [[package]] @@ -4813,7 +4813,7 @@ requires-dist = [ { name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, { name = "deepagents", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=0.6.12,<0.7" }, { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.2.0,<3" }, - { name = "google-genai", marker = "extra == 'google-genai'", specifier = ">=2.10.0,<3.0.0" }, + { name = "google-genai", marker = "extra == 'google-genai'", specifier = ">=2.21.0,<3.0.0" }, { name = "grpcio", marker = "extra == 'grpc'", specifier = ">=1.48.2,<2" }, { name = "langchain", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=1.3.11,<2" }, { name = "langchain-core", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=1.4.8,<2" }, From f74399de1c267c22c69f7713c66ea4a332465ad8 Mon Sep 17 00:00:00 2001 From: Sean Bollin Date: Tue, 15 Sep 2026 15:35:58 -0700 Subject: [PATCH 6/9] Add Google Cloud Run CloudRunIDPlugin (#1776) * Add Google Cloud Run worker identity/deployment helper Adds an experimental Google Cloud Run helper, mirroring the existing AWS Lambda module's worker-ID behavior. Because Cloud Run runs a long-lived container (unlike Lambda's per-invocation model), this is a metadata helper rather than a worker wrapper: it reads the Cloud Run instance metadata -- the instance id from the metadata server, plus the worker pool/service name and revision from CLOUD_RUN_WORKER_POOL / CLOUD_RUN_REVISION (worker pools) or K_SERVICE / K_REVISION (services) -- and derives a worker identity and a WorkerDeploymentVersion to apply to a normal long-lived worker. Covers both Cloud Run worker pools and services. Co-Authored-By: Claude Opus 4.8 * Set worker versioning behavior to PINNED in the Cloud Run worker apply The worker-side apply helper enabled versioning and set the deployment version but left the default versioning behavior unset, so a versioned worker with a plain (un-annotated) workflow failed to register. Default it to PINNED; a per-workflow versioning behavior still takes precedence. Co-Authored-By: Claude Opus 4.8 * Add unit tests for the Google Cloud Run metadata helper Cover the temporalio.contrib.gcp.cloud_run helper: - Environment precedence for the deployment name (CLOUD_RUN_WORKER_POOL over K_SERVICE) and revision (CLOUD_RUN_REVISION over K_REVISION). - Worker identity formatting and its revision -> name -> instance-id fallbacks. - WorkerDeploymentVersion derivation and its ValueError when name/revision empty. - WorkerDeploymentConfig enabling worker versioning with PINNED default behavior. - The metadata HTTP fetch via a local in-process server: asserts the Metadata-Flavor: Google header is sent, the body is trimmed, and a clear RuntimeError is raised on non-200 and unreachable responses. Uses the helper's dependency-injection seams (getenv and metadata_url) so no real environment or network access is required. Co-Authored-By: Claude Opus 4.8 * Add CloudRunPlugin for Google Cloud Run worker defaults Re-architect the Cloud Run worker-ID helper into a plugin mirroring the SDK's OpenTelemetry Cloud Run plugin. CloudRunPlugin subclasses temporalio.plugin.SimplePlugin and is registered once on the client via Client.connect(plugins=[...]); it propagates to workers automatically. The plugin fetches Cloud Run instance metadata lazily at client connect and caches it, then sets the client identity (only when the caller did not provide one) and configures the worker with a PINNED WorkerDeploymentConfig derived from the Cloud Run revision. Connecting off Cloud Run fails fast with a clear error. The GoogleCloudRunMetadata dataclass and its worker_identity / worker_deployment_version / worker_deployment_config properties are kept for advanced and non-plugin use. Co-Authored-By: Claude Opus 4.8 * Rename CloudRunPlugin to WorkerIDPlugin and fix test type errors Cloud Run can host multiple plugins (a worker-ID plugin and an OpenTelemetry plugin both live in the same cloud_run area), so the worker-ID plugin needs a specific name rather than the generic CloudRunPlugin. - Rename class CloudRunPlugin -> WorkerIDPlugin and move _plugin.py -> _worker_id_plugin.py (cloud_run package and GoogleCloudRunMetadata unchanged). - Update the package __init__ export/__all__, quick-start, and README. - Rename test_plugin.py -> test_worker_id_plugin.py and fix the basedpyright reportInvalidCast errors by constructing WorkerConfig() instead of cast(WorkerConfig, {}); silence reportUnusedParameter on the unused connect() callbacks. Co-Authored-By: Claude Opus 4.8 * Move Cloud Run worker-ID plugin under cloud_run/worker_id/ Relocate the Google Cloud Run worker-ID plugin from directly inside temporalio/contrib/gcp/cloud_run/ into a new worker_id/ sub-package so it owns its own __init__.py and README.md. This avoids a hard collision with the separate OTel Cloud Run plugin, which also owns cloud_run/README.md and cloud_run/__init__.py; after the move the two plugins share only the minimal cloud_run/ and gcp/ namespace-marker __init__.py files. The public names are unchanged; only the import path gains .worker_id: from temporalio.contrib.gcp.cloud_run.worker_id import WorkerIDPlugin cloud_run/__init__.py is reduced to a minimal namespace-marker docstring with no worker-ID exports. Mirrors the Go (contrib/gcp/cloudrun/workerid) and .NET (CloudRun.WorkerId) layouts. Co-Authored-By: Claude Opus 4.8 * Add CHANGELOG entry for Cloud Run worker-ID plugin Satisfies the changelog checkpoint, which requires a user-facing change to add an entry under Unreleased in a CHANGELOG.md. Co-Authored-By: Claude Opus 4.8 * Align gcp namespace markers with the Cloud Run OTel PR Reduce temporalio/contrib/gcp/__init__.py and temporalio/contrib/gcp/cloud_run/__init__.py to the same generic namespace markers used by the Cloud Run OpenTelemetry PR, so both PRs add byte-identical files and merge into main without an add/add conflict. The worker_id plugin imports from cloud_run.worker_id, so this docstring-only change to the cloud_run root is safe. Co-Authored-By: Claude Opus 4.8 * Make Cloud Run WorkerID plugin identity-only Drop all Worker Deployment Versioning from the Cloud Run WorkerID plugin so it only sets the worker identity from Cloud Run instance metadata. - Remove the worker_deployment_version and worker_deployment_config (PINNED) properties from GoogleCloudRunMetadata, plus the now-unused WorkerDeploymentVersion, WorkerDeploymentConfig, and VersioningBehavior imports. - Remove WorkerIDPlugin.configure_worker (which set deployment_config); the plugin no longer overrides the worker configurator. The client hook still sets the identity when unset or equal to the SDK pid@host default, and the worker inherits it. - Update tests, docstrings, README, and the CHANGELOG entry to identity-only wording. Co-Authored-By: Claude Opus 4.8 * Remove deployment-name/build-ID wording from Cloud Run worker-ID docs The plugin sets only the worker identity, so describe the Cloud Run metadata as the worker pool/service name and revision rather than a Temporal deployment name and build ID. Co-Authored-By: Claude Opus 4.8 * Simplify worker identity docs and comments Trim negative-contrast framing and editorializing so the comments and docs describe only the worker identity functionality, per review feedback. Co-Authored-By: Claude Opus 4.8 * Fixing Claude Code prose to be more succinct * Put the experimental warning back to allow for future updates * Rename plugin to CloudRunIDPlugin The plugin sets the client identity (which in turn sets the worker identity), and the new name carries the Cloud Run context without relying on the full package path. Co-Authored-By: Claude Opus 4.8 * Rename package leaf to id and metadata accessor to Identity Renames the Cloud Run extension package from cloud_run.worker_id to cloud_run.id, and GoogleCloudRunMetadata.worker_identity to .identity. Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: Claude Opus 4.8 --- CHANGELOG.md | 1 + temporalio/contrib/gcp/cloud_run/id/README.md | 84 +++++++ .../contrib/gcp/cloud_run/id/__init__.py | 44 ++++ .../gcp/cloud_run/id/_cloud_run_id_plugin.py | 102 ++++++++ .../contrib/gcp/cloud_run/id/_metadata.py | 104 +++++++++ tests/contrib/gcp/cloud_run/id/__init__.py | 0 .../cloud_run/id/test_cloud_run_id_plugin.py | 127 ++++++++++ .../contrib/gcp/cloud_run/id/test_metadata.py | 220 ++++++++++++++++++ 8 files changed, 682 insertions(+) create mode 100644 temporalio/contrib/gcp/cloud_run/id/README.md create mode 100644 temporalio/contrib/gcp/cloud_run/id/__init__.py create mode 100644 temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py create mode 100644 temporalio/contrib/gcp/cloud_run/id/_metadata.py create mode 100644 tests/contrib/gcp/cloud_run/id/__init__.py create mode 100644 tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py create mode 100644 tests/contrib/gcp/cloud_run/id/test_metadata.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 87dafcbe2..172bc72d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ to include examples, links to docs, or any other relevant information. ### Added +- Added the `temporalio.contrib.gcp.cloud_run.id` module plugin to help set the worker identity on Cloud Run. ### Changed ### Deprecated diff --git a/temporalio/contrib/gcp/cloud_run/id/README.md b/temporalio/contrib/gcp/cloud_run/id/README.md new file mode 100644 index 000000000..87e0b0ff4 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/id/README.md @@ -0,0 +1,84 @@ +# id + +> ⚠️ **This package is currently at an experimental release stage.** ⚠️ + +A plugin for running [Temporal](https://temporal.io) workers on Google Cloud Run. `CloudRunIDPlugin` +reads Cloud Run instance metadata and sets the client identity. Both Cloud Run **worker pools** and +**services** are supported. + +Register the plugin once when connecting the client and it sets the client **identity** to a value +derived from the Cloud Run instance (unless you already passed an `identity`). + +## Quick start + +```python +import asyncio + +from temporalio.client import Client +from temporalio.contrib.gcp.cloud_run.id import CloudRunIDPlugin +from temporalio.worker import Worker + +from my_workflows import MyWorkflow +from my_activities import my_activity + + +async def main() -> None: + # Install the plugin on the client; it propagates to workers automatically. + client = await Client.connect( + "localhost:7233", + plugins=[CloudRunIDPlugin()], + ) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[MyWorkflow], + activities=[my_activity], + ) + await worker.run() + + +if __name__ == "__main__": + asyncio.run(main()) +``` + +## How it works + +Cloud Run exposes workload metadata through environment variables and a metadata server: + +- **Worker pools** get `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` (and no `K_*` variables). +- **Services** get `K_SERVICE`, `K_REVISION`, and `K_CONFIGURATION` (and no `CLOUD_RUN_*` variables). + +The unique instance id is not available as an environment variable on either; it is only exposed by +the +[Cloud Run metadata server](https://cloud.google.com/run/docs/container-contract#metadata-server) +at `http://metadata.google.internal/computeMetadata/v1/instance/id`, which requires the +`Metadata-Flavor: Google` request header. + +When the client connects, `CloudRunIDPlugin` resolves the worker pool name from +`CLOUD_RUN_WORKER_POOL` (falling back to the service name `K_SERVICE`) and the revision from +`CLOUD_RUN_REVISION` (falling back to +`K_REVISION`), then performs a single synchronous HTTP GET to the metadata server for the instance +id. From that metadata the plugin sets: + +- **Client identity** -- `@`, uniquely identifying this worker instance in + Temporal tooling. It falls back to `@`, then to just ``, when the + revision or name is unavailable. An `identity` you pass to `Client.connect` always wins. + +The metadata server is only reachable from within Cloud Run, so connecting elsewhere raises an +error. The plugin uses only the Python standard library and adds no new dependencies. + +## Advanced / non-plugin use + +For advanced scenarios or unit tests you can bypass the metadata server by passing a pre-built +metadata object, or steer the fetch with `getenv` / `metadata_url` / `timeout`: + +```python +from temporalio.contrib.gcp.cloud_run.id import CloudRunIDPlugin, get_google_cloud_run_metadata + +metadata = get_google_cloud_run_metadata() +plugin = CloudRunIDPlugin(metadata=metadata) + +# metadata.identity exposes the same value the plugin applies, for use +# without the plugin if needed. +``` diff --git a/temporalio/contrib/gcp/cloud_run/id/__init__.py b/temporalio/contrib/gcp/cloud_run/id/__init__.py new file mode 100644 index 000000000..d5f52b9e1 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/id/__init__.py @@ -0,0 +1,44 @@ +"""Run Temporal workers on Google Cloud Run. + +:py:class:`CloudRunIDPlugin` reads Cloud Run instance metadata (from a worker pool or a service) and +sets the client identity from the Cloud Run instance. + +Quick start:: + + import asyncio + + from temporalio.client import Client + from temporalio.contrib.gcp.cloud_run.id import CloudRunIDPlugin + from temporalio.worker import Worker + + async def main() -> None: + # Install the plugin on the client; it propagates to workers automatically. + client = await Client.connect( + "localhost:7233", + plugins=[CloudRunIDPlugin()], + ) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[MyWorkflow], + activities=[my_activity], + ) + await worker.run() + + asyncio.run(main()) +""" + +from temporalio.contrib.gcp.cloud_run.id._cloud_run_id_plugin import CloudRunIDPlugin +from temporalio.contrib.gcp.cloud_run.id._metadata import ( + CLOUD_RUN_METADATA_URL, + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) + +__all__ = [ + "CLOUD_RUN_METADATA_URL", + "GoogleCloudRunMetadata", + "CloudRunIDPlugin", + "get_google_cloud_run_metadata", +] diff --git a/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py b/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py new file mode 100644 index 000000000..e4216fa37 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py @@ -0,0 +1,102 @@ +"""Plugin setting a Temporal client's identity from Google Cloud Run instance metadata.""" + +from __future__ import annotations + +import os +import socket +from collections.abc import Awaitable, Callable + +import temporalio.plugin +from temporalio.contrib.gcp.cloud_run.id._metadata import ( + CLOUD_RUN_METADATA_URL, + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) +from temporalio.service import ConnectConfig, ServiceClient + + +class CloudRunIDPlugin(temporalio.plugin.SimplePlugin): + """Set a Temporal client's identity from Google Cloud Run instance metadata. + + Install this plugin once when connecting the client; the identity it sets + automatically propagates to workers created from that client. It sets the + client **identity** to a value derived from the Cloud Run instance, but only + when the caller did not already provide one. Both Cloud Run worker pools and + services are supported. + + The Cloud Run instance metadata is fetched once, lazily, when the client + connects. If it cannot be read (usually because the process is not running on + Cloud Run), connecting raises an error. + + Unit tests and advanced callers can bypass the metadata server by passing a + pre-built ``metadata`` object, or steer the fetch with ``getenv`` / + ``metadata_url`` / ``timeout``. + + .. warning:: + Google Cloud Run support is experimental and may change in future versions. + """ + + def __init__( + self, + *, + metadata: GoogleCloudRunMetadata | None = None, + timeout: float = 2.0, + metadata_url: str = CLOUD_RUN_METADATA_URL, + getenv: Callable[[str], str | None] = os.environ.get, + ) -> None: + """Create a Cloud Run plugin. + + Args: + metadata: Pre-fetched Cloud Run instance metadata. When supplied, the + plugin uses it directly and never contacts the metadata server. + Primarily for testing and advanced use. + timeout: Timeout, in seconds, for the request to the metadata server. + Ignored when ``metadata`` is supplied. + metadata_url: URL of the Cloud Run metadata server endpoint that + returns the instance id. Ignored when ``metadata`` is supplied. + getenv: Callable used to look up environment variables. Defaults to + ``os.environ.get`` and exists primarily for testing. Ignored when + ``metadata`` is supplied. + """ + super().__init__("CloudRunIDPlugin") + self._metadata = metadata + self._timeout = timeout + self._metadata_url = metadata_url + self._getenv = getenv + + async def connect_service_client( + self, + config: ConnectConfig, + next: Callable[[ConnectConfig], Awaitable[ServiceClient]], + ) -> ServiceClient: + """Fetch Cloud Run metadata and set the client identity before connecting. + + The identity is only set when the caller did not provide one, so an + explicit ``identity`` passed to :py:meth:`temporalio.client.Client.connect` + always wins. + """ + metadata = self._resolve_metadata() + if not config.identity or config.identity == _default_identity(): + config.identity = metadata.identity + return await super().connect_service_client(config, next) + + def _resolve_metadata(self) -> GoogleCloudRunMetadata: + """Return the cached Cloud Run metadata, fetching it once on first use.""" + if self._metadata is None: + self._metadata = get_google_cloud_run_metadata( + timeout=self._timeout, + metadata_url=self._metadata_url, + getenv=self._getenv, # type: ignore[arg-type] + ) + return self._metadata + + +def _default_identity() -> str: + """Recreate the identity ``ConnectConfig`` auto-generates when none is given. + + :py:class:`temporalio.service.ConnectConfig` fills an unset identity with + ``@`` in ``__post_init__``, so by the time this plugin runs the + identity is never literally empty. Matching that value lets the plugin tell an + auto-generated identity (safe to replace) from one the caller chose (kept). + """ + return f"{os.getpid()}@{socket.gethostname()}" diff --git a/temporalio/contrib/gcp/cloud_run/id/_metadata.py b/temporalio/contrib/gcp/cloud_run/id/_metadata.py new file mode 100644 index 000000000..df545bae6 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/id/_metadata.py @@ -0,0 +1,104 @@ +"""Read Google Cloud Run instance metadata for Temporal worker configuration. + +Helpers for deriving a worker identity from Cloud Run instance metadata. Both Cloud Run worker pools +and services are supported. +""" + +from __future__ import annotations + +import os +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass + +CLOUD_RUN_METADATA_URL = ( + "http://metadata.google.internal/computeMetadata/v1/instance/id" +) +"""Default Cloud Run metadata server endpoint returning the unique instance id.""" + + +@dataclass(frozen=True) +class GoogleCloudRunMetadata: + """Identifying metadata for the current Google Cloud Run instance. + + Both Cloud Run worker pools and services are supported. Worker pools expose + ``CLOUD_RUN_WORKER_POOL`` and ``CLOUD_RUN_REVISION``; services expose ``K_SERVICE`` and + ``K_REVISION``. + + Attributes: + instance_id: Unique id of this Cloud Run container instance, read from the Cloud Run + metadata server. + name: The Cloud Run worker pool name (``CLOUD_RUN_WORKER_POOL``) or, for a service, the + service name (``K_SERVICE``). May be empty when the process is not running on Cloud Run. + revision: Cloud Run revision name (``CLOUD_RUN_REVISION`` for worker pools or ``K_REVISION`` + for services). May be empty when the process is not running on Cloud Run. + """ + + instance_id: str + name: str + revision: str + + @property + def identity(self) -> str: + """Worker identity string uniquely identifying this Cloud Run instance. + + The format is ``@``. When the revision is empty the worker pool or + service name is used instead (``@``); when both are empty the instance id + is returned on its own. + """ + if self.revision: + return f"{self.instance_id}@{self.revision}" + if self.name: + return f"{self.instance_id}@{self.name}" + return self.instance_id + + +def get_google_cloud_run_metadata( + *, + timeout: float = 2.0, + metadata_url: str = CLOUD_RUN_METADATA_URL, + getenv: Callable[[str], str] = os.environ.get, # type: ignore[assignment] +) -> GoogleCloudRunMetadata: + """Read metadata identifying the current Google Cloud Run instance. + + Resolves the worker pool name from ``CLOUD_RUN_WORKER_POOL`` (Cloud Run worker pools), falling + back to the service name (``K_SERVICE``, Cloud Run services), and the revision from + ``CLOUD_RUN_REVISION`` falling back to ``K_REVISION``. The unique instance id is fetched from + the Cloud Run metadata server with a single synchronous HTTP GET. Intended to be called once at + worker startup. + + Args: + timeout: Timeout, in seconds, for the request to the metadata server. + metadata_url: URL of the Cloud Run metadata server endpoint that returns the instance id. + getenv: Callable used to look up environment variables. Defaults to ``os.environ.get`` and + exists primarily for testing. + + Returns: + A :py:class:`GoogleCloudRunMetadata` describing the current instance. + + Raises: + RuntimeError: If the metadata server cannot be reached, which usually means the process is + not running on a Cloud Run worker pool or service. + """ + name = getenv("CLOUD_RUN_WORKER_POOL") or getenv("K_SERVICE") or "" + revision = getenv("CLOUD_RUN_REVISION") or getenv("K_REVISION") or "" + + request = urllib.request.Request( + metadata_url, + headers={"Metadata-Flavor": "Google"}, + method="GET", + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + instance_id = response.read().decode("utf-8").strip() + except OSError as err: + raise RuntimeError( + f"Failed to reach the Cloud Run metadata server at {metadata_url!r}; " + "this process may not be running on a Cloud Run worker pool or service." + ) from err + + return GoogleCloudRunMetadata( + instance_id=instance_id, + name=name, + revision=revision, + ) diff --git a/tests/contrib/gcp/cloud_run/id/__init__.py b/tests/contrib/gcp/cloud_run/id/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py b/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py new file mode 100644 index 000000000..083ed05ad --- /dev/null +++ b/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py @@ -0,0 +1,127 @@ +"""Tests for the Google Cloud Run worker-ID plugin.""" + +from __future__ import annotations + +import os +import socket +from typing import cast +from unittest.mock import Mock + +import pytest + +from temporalio.contrib.gcp.cloud_run.id import ( + CloudRunIDPlugin, + GoogleCloudRunMetadata, +) +from temporalio.service import ConnectConfig, ServiceClient + + +def _metadata( + *, + instance_id: str = "instance-1", + name: str = "my-pool", + revision: str = "rev-1", +) -> GoogleCloudRunMetadata: + return GoogleCloudRunMetadata( + instance_id=instance_id, + name=name, + revision=revision, + ) + + +def _closed_port() -> int: + """Return a port number that nothing is listening on.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def _service_client() -> ServiceClient: + return cast(ServiceClient, Mock(spec=ServiceClient)) + + +# ---- Client identity ---- + + +class TestClientIdentity: + @pytest.mark.asyncio + async def test_sets_identity_when_unset(self) -> None: + plugin = CloudRunIDPlugin( + metadata=_metadata(instance_id="abc", revision="rev-1") + ) + # ConnectConfig auto-fills identity with @ when none is given. + config = ConnectConfig(target_host="localhost:7233") + assert config.identity == f"{os.getpid()}@{socket.gethostname()}" + service_client = _service_client() + + async def connect(input: ConnectConfig) -> ServiceClient: + assert input.identity == "abc@rev-1" + return service_client + + assert await plugin.connect_service_client(config, connect) is service_client + assert config.identity == "abc@rev-1" + + @pytest.mark.asyncio + async def test_preserves_caller_identity(self) -> None: + plugin = CloudRunIDPlugin( + metadata=_metadata(instance_id="abc", revision="rev-1") + ) + config = ConnectConfig(target_host="localhost:7233", identity="my-identity") + service_client = _service_client() + + async def connect(input: ConnectConfig) -> ServiceClient: + assert input.identity == "my-identity" + return service_client + + assert await plugin.connect_service_client(config, connect) is service_client + assert config.identity == "my-identity" + + +# ---- Metadata fetching / caching ---- + + +class TestMetadataFetch: + def test_construction_does_not_fetch(self) -> None: + # A bad metadata URL must not raise at construction -- the fetch is lazy. + CloudRunIDPlugin( + metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", + getenv={}.get, # type: ignore[arg-type] + ) + + @pytest.mark.asyncio + async def test_connect_fails_fast_off_platform(self) -> None: + plugin = CloudRunIDPlugin( + timeout=1.0, + metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", + getenv={}.get, # type: ignore[arg-type] + ) + config = ConnectConfig(target_host="localhost:7233") + + async def connect(_input: ConnectConfig) -> ServiceClient: + raise AssertionError("should not connect when metadata is unavailable") + + with pytest.raises(RuntimeError, match="metadata server"): + await plugin.connect_service_client(config, connect) + + @pytest.mark.asyncio + async def test_metadata_fetched_from_server_at_connect( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fetch = Mock(return_value=_metadata(instance_id="abc", revision="rev-1")) + monkeypatch.setattr( + "temporalio.contrib.gcp.cloud_run.id._cloud_run_id_plugin.get_google_cloud_run_metadata", + fetch, + ) + plugin = CloudRunIDPlugin() + config = ConnectConfig(target_host="localhost:7233") + + async def connect(_input: ConnectConfig) -> ServiceClient: + return _service_client() + + await plugin.connect_service_client(config, connect) + + # Fetched from the metadata server when the client connects. + fetch.assert_called_once() + assert config.identity == "abc@rev-1" diff --git a/tests/contrib/gcp/cloud_run/id/test_metadata.py b/tests/contrib/gcp/cloud_run/id/test_metadata.py new file mode 100644 index 000000000..75b11d65b --- /dev/null +++ b/tests/contrib/gcp/cloud_run/id/test_metadata.py @@ -0,0 +1,220 @@ +"""Tests for temporalio.contrib.gcp.cloud_run.id.""" + +from __future__ import annotations + +import socket +import threading +from collections.abc import Iterator +from email.message import Message +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any + +import pytest + +from temporalio.contrib.gcp.cloud_run.id import ( + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) + + +def _metadata( + *, + instance_id: str = "instance-1", + name: str = "", + revision: str = "", +) -> GoogleCloudRunMetadata: + return GoogleCloudRunMetadata( + instance_id=instance_id, + name=name, + revision=revision, + ) + + +# ---- Local metadata-server fixture ---- + + +class _MetadataServer(HTTPServer): + """In-process stand-in for the Cloud Run metadata server. + + Records the headers and path of the last request and serves a configurable + status and body so tests can assert on both the request and the response. + """ + + url: str = "" + response_status: int = 200 + response_body: str = "instance-1" + received_path: str | None = None + received_headers: Message[str, str] | None = None + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 (http.server naming) + server: _MetadataServer = self.server # type: ignore[assignment] + server.received_path = self.path + server.received_headers = self.headers + body = server.response_body.encode("utf-8") + self.send_response(server.response_status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + # Silence the default stderr request logging. The parameter is named + # ``format`` to match BaseHTTPRequestHandler.log_message. + pass + + +@pytest.fixture +def metadata_server() -> Iterator[_MetadataServer]: + server = _MetadataServer(("127.0.0.1", 0), _Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + server.url = f"http://127.0.0.1:{port}/computeMetadata/v1/instance/id" + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _closed_port() -> int: + """Return a port number that nothing is listening on.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +# ---- Environment precedence ---- + + +class TestEnvPrecedence: + def test_worker_pool_wins_over_service( + self, metadata_server: _MetadataServer + ) -> None: + env = {"CLOUD_RUN_WORKER_POOL": "my-pool", "K_SERVICE": "my-service"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.name == "my-pool" + + def test_service_used_when_pool_absent( + self, metadata_server: _MetadataServer + ) -> None: + env = {"K_SERVICE": "my-service"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.name == "my-service" + + def test_name_empty_when_neither_set( + self, metadata_server: _MetadataServer + ) -> None: + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + assert metadata.name == "" + + def test_cloud_run_revision_wins_over_k_revision( + self, metadata_server: _MetadataServer + ) -> None: + env = {"CLOUD_RUN_REVISION": "rev-cr", "K_REVISION": "rev-k"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.revision == "rev-cr" + + def test_k_revision_used_when_cloud_run_revision_absent( + self, metadata_server: _MetadataServer + ) -> None: + env = {"K_REVISION": "rev-k"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.revision == "rev-k" + + def test_revision_empty_when_neither_set( + self, metadata_server: _MetadataServer + ) -> None: + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + assert metadata.revision == "" + + +# ---- Worker identity ---- + + +class TestWorkerIdentity: + def test_identity_uses_revision(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="rev-1") + assert metadata.identity == "abc@rev-1" + + def test_identity_falls_back_to_name(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="") + assert metadata.identity == "abc@my-pool" + + def test_identity_falls_back_to_instance_id(self) -> None: + metadata = _metadata(instance_id="abc", name="", revision="") + assert metadata.identity == "abc" + + +# ---- HTTP fetch ---- + + +class TestHttpFetch: + def test_sends_metadata_flavor_header_and_trims_body( + self, metadata_server: _MetadataServer + ) -> None: + metadata_server.response_body = " instance-xyz\n" + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + assert metadata.instance_id == "instance-xyz" + assert metadata_server.received_headers is not None + assert metadata_server.received_headers.get("Metadata-Flavor") == "Google" + assert metadata_server.received_path == "/computeMetadata/v1/instance/id" + + def test_errors_on_non_200(self, metadata_server: _MetadataServer) -> None: + metadata_server.response_status = 500 + metadata_server.response_body = "boom" + with pytest.raises(RuntimeError, match="metadata server"): + get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + + def test_errors_when_unreachable(self) -> None: + url = f"http://127.0.0.1:{_closed_port()}/computeMetadata/v1/instance/id" + with pytest.raises(RuntimeError, match="metadata server"): + get_google_cloud_run_metadata( + metadata_url=url, + timeout=1.0, + getenv={}.get, # type: ignore[arg-type] + ) + + def test_end_to_end_from_env_and_server( + self, metadata_server: _MetadataServer + ) -> None: + metadata_server.response_body = "instance-42" + env = {"CLOUD_RUN_WORKER_POOL": "my-pool", "CLOUD_RUN_REVISION": "rev-7"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata == GoogleCloudRunMetadata( + instance_id="instance-42", + name="my-pool", + revision="rev-7", + ) + assert metadata.identity == "instance-42@rev-7" From fa5cf46ea04920be138bafca7e20ca6208822c30 Mon Sep 17 00:00:00 2001 From: Sean Bollin Date: Tue, 15 Sep 2026 17:24:16 -0700 Subject: [PATCH 7/9] Rename CloudRunIDPlugin to CloudRunIdPlugin (#1869) * Rename CloudRunIDPlugin to CloudRunIdPlugin Uses lowercase Id (CloudRunId) per repo naming convention, matching the .NET review feedback and applied across all SDKs. Co-Authored-By: Claude Opus 4.8 * Add CHANGELOG entry naming CloudRunIdPlugin Names the renamed client plugin in the existing Unreleased entry so the changelog reflects the final CloudRunIdPlugin name and satisfies the changelog checkpoint. Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: Claude Opus 4.8 --- CHANGELOG.md | 2 +- temporalio/contrib/gcp/cloud_run/id/README.md | 12 ++++++------ temporalio/contrib/gcp/cloud_run/id/__init__.py | 10 +++++----- .../contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py | 4 ++-- .../gcp/cloud_run/id/test_cloud_run_id_plugin.py | 12 ++++++------ 5 files changed, 20 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 172bc72d9..e8597d2c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,7 @@ to include examples, links to docs, or any other relevant information. ### Added -- Added the `temporalio.contrib.gcp.cloud_run.id` module plugin to help set the worker identity on Cloud Run. +- Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run. ### Changed ### Deprecated diff --git a/temporalio/contrib/gcp/cloud_run/id/README.md b/temporalio/contrib/gcp/cloud_run/id/README.md index 87e0b0ff4..cb56b0045 100644 --- a/temporalio/contrib/gcp/cloud_run/id/README.md +++ b/temporalio/contrib/gcp/cloud_run/id/README.md @@ -2,7 +2,7 @@ > ⚠️ **This package is currently at an experimental release stage.** ⚠️ -A plugin for running [Temporal](https://temporal.io) workers on Google Cloud Run. `CloudRunIDPlugin` +A plugin for running [Temporal](https://temporal.io) workers on Google Cloud Run. `CloudRunIdPlugin` reads Cloud Run instance metadata and sets the client identity. Both Cloud Run **worker pools** and **services** are supported. @@ -15,7 +15,7 @@ derived from the Cloud Run instance (unless you already passed an `identity`). import asyncio from temporalio.client import Client -from temporalio.contrib.gcp.cloud_run.id import CloudRunIDPlugin +from temporalio.contrib.gcp.cloud_run.id import CloudRunIdPlugin from temporalio.worker import Worker from my_workflows import MyWorkflow @@ -26,7 +26,7 @@ async def main() -> None: # Install the plugin on the client; it propagates to workers automatically. client = await Client.connect( "localhost:7233", - plugins=[CloudRunIDPlugin()], + plugins=[CloudRunIdPlugin()], ) worker = Worker( @@ -55,7 +55,7 @@ the at `http://metadata.google.internal/computeMetadata/v1/instance/id`, which requires the `Metadata-Flavor: Google` request header. -When the client connects, `CloudRunIDPlugin` resolves the worker pool name from +When the client connects, `CloudRunIdPlugin` resolves the worker pool name from `CLOUD_RUN_WORKER_POOL` (falling back to the service name `K_SERVICE`) and the revision from `CLOUD_RUN_REVISION` (falling back to `K_REVISION`), then performs a single synchronous HTTP GET to the metadata server for the instance @@ -74,10 +74,10 @@ For advanced scenarios or unit tests you can bypass the metadata server by passi metadata object, or steer the fetch with `getenv` / `metadata_url` / `timeout`: ```python -from temporalio.contrib.gcp.cloud_run.id import CloudRunIDPlugin, get_google_cloud_run_metadata +from temporalio.contrib.gcp.cloud_run.id import CloudRunIdPlugin, get_google_cloud_run_metadata metadata = get_google_cloud_run_metadata() -plugin = CloudRunIDPlugin(metadata=metadata) +plugin = CloudRunIdPlugin(metadata=metadata) # metadata.identity exposes the same value the plugin applies, for use # without the plugin if needed. diff --git a/temporalio/contrib/gcp/cloud_run/id/__init__.py b/temporalio/contrib/gcp/cloud_run/id/__init__.py index d5f52b9e1..40f349797 100644 --- a/temporalio/contrib/gcp/cloud_run/id/__init__.py +++ b/temporalio/contrib/gcp/cloud_run/id/__init__.py @@ -1,6 +1,6 @@ """Run Temporal workers on Google Cloud Run. -:py:class:`CloudRunIDPlugin` reads Cloud Run instance metadata (from a worker pool or a service) and +:py:class:`CloudRunIdPlugin` reads Cloud Run instance metadata (from a worker pool or a service) and sets the client identity from the Cloud Run instance. Quick start:: @@ -8,14 +8,14 @@ import asyncio from temporalio.client import Client - from temporalio.contrib.gcp.cloud_run.id import CloudRunIDPlugin + from temporalio.contrib.gcp.cloud_run.id import CloudRunIdPlugin from temporalio.worker import Worker async def main() -> None: # Install the plugin on the client; it propagates to workers automatically. client = await Client.connect( "localhost:7233", - plugins=[CloudRunIDPlugin()], + plugins=[CloudRunIdPlugin()], ) worker = Worker( @@ -29,7 +29,7 @@ async def main() -> None: asyncio.run(main()) """ -from temporalio.contrib.gcp.cloud_run.id._cloud_run_id_plugin import CloudRunIDPlugin +from temporalio.contrib.gcp.cloud_run.id._cloud_run_id_plugin import CloudRunIdPlugin from temporalio.contrib.gcp.cloud_run.id._metadata import ( CLOUD_RUN_METADATA_URL, GoogleCloudRunMetadata, @@ -39,6 +39,6 @@ async def main() -> None: __all__ = [ "CLOUD_RUN_METADATA_URL", "GoogleCloudRunMetadata", - "CloudRunIDPlugin", + "CloudRunIdPlugin", "get_google_cloud_run_metadata", ] diff --git a/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py b/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py index e4216fa37..ba11f43fe 100644 --- a/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py +++ b/temporalio/contrib/gcp/cloud_run/id/_cloud_run_id_plugin.py @@ -15,7 +15,7 @@ from temporalio.service import ConnectConfig, ServiceClient -class CloudRunIDPlugin(temporalio.plugin.SimplePlugin): +class CloudRunIdPlugin(temporalio.plugin.SimplePlugin): """Set a Temporal client's identity from Google Cloud Run instance metadata. Install this plugin once when connecting the client; the identity it sets @@ -58,7 +58,7 @@ def __init__( ``os.environ.get`` and exists primarily for testing. Ignored when ``metadata`` is supplied. """ - super().__init__("CloudRunIDPlugin") + super().__init__("CloudRunIdPlugin") self._metadata = metadata self._timeout = timeout self._metadata_url = metadata_url diff --git a/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py b/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py index 083ed05ad..52055ad58 100644 --- a/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py +++ b/tests/contrib/gcp/cloud_run/id/test_cloud_run_id_plugin.py @@ -10,7 +10,7 @@ import pytest from temporalio.contrib.gcp.cloud_run.id import ( - CloudRunIDPlugin, + CloudRunIdPlugin, GoogleCloudRunMetadata, ) from temporalio.service import ConnectConfig, ServiceClient @@ -48,7 +48,7 @@ def _service_client() -> ServiceClient: class TestClientIdentity: @pytest.mark.asyncio async def test_sets_identity_when_unset(self) -> None: - plugin = CloudRunIDPlugin( + plugin = CloudRunIdPlugin( metadata=_metadata(instance_id="abc", revision="rev-1") ) # ConnectConfig auto-fills identity with @ when none is given. @@ -65,7 +65,7 @@ async def connect(input: ConnectConfig) -> ServiceClient: @pytest.mark.asyncio async def test_preserves_caller_identity(self) -> None: - plugin = CloudRunIDPlugin( + plugin = CloudRunIdPlugin( metadata=_metadata(instance_id="abc", revision="rev-1") ) config = ConnectConfig(target_host="localhost:7233", identity="my-identity") @@ -85,14 +85,14 @@ async def connect(input: ConnectConfig) -> ServiceClient: class TestMetadataFetch: def test_construction_does_not_fetch(self) -> None: # A bad metadata URL must not raise at construction -- the fetch is lazy. - CloudRunIDPlugin( + CloudRunIdPlugin( metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", getenv={}.get, # type: ignore[arg-type] ) @pytest.mark.asyncio async def test_connect_fails_fast_off_platform(self) -> None: - plugin = CloudRunIDPlugin( + plugin = CloudRunIdPlugin( timeout=1.0, metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", getenv={}.get, # type: ignore[arg-type] @@ -114,7 +114,7 @@ async def test_metadata_fetched_from_server_at_connect( "temporalio.contrib.gcp.cloud_run.id._cloud_run_id_plugin.get_google_cloud_run_metadata", fetch, ) - plugin = CloudRunIDPlugin() + plugin = CloudRunIdPlugin() config = ConnectConfig(target_host="localhost:7233") async def connect(_input: ConnectConfig) -> ServiceClient: From 7e9ee8062fde175cb5702ac6049403770e89aaec Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 13:38:49 -0700 Subject: [PATCH 8/9] Unwrap current payload converters (#1867) --- CHANGELOG.md | 3 +++ temporalio/activity.py | 4 +++- temporalio/converter/_payload_converter.py | 7 +++++++ temporalio/nexus/system/__init__.py | 4 +++- temporalio/worker/_workflow_instance.py | 7 ++++++- tests/nexus/test_temporal_system_nexus.py | 13 +++++++------ tests/test_serialization_context.py | 10 ++++++++-- 7 files changed, 37 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8597d2c2..f016f34ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Current workflow and activity payload converter accessors now return the configured converter + without SDK-internal transfer type conversion. + ### Security ## [1.33.0] - 2026-09-14 diff --git a/temporalio/activity.py b/temporalio/activity.py index 3f69bc17f..82678d8a1 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -462,7 +462,9 @@ def payload_converter() -> temporalio.converter.PayloadConverter: 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 + return _TemporalTransferTypePayloadConverter.unwrap( + _Context.current().payload_converter + ) def metric_meter() -> temporalio.common.MetricMeter: diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index a8bc35e28..00c63e340 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -612,6 +612,13 @@ def wrap(payload_converter: PayloadConverter) -> PayloadConverter: return payload_converter return _TemporalTransferTypePayloadConverter(payload_converter) + @staticmethod + def unwrap(payload_converter: PayloadConverter) -> PayloadConverter: + """Remove this wrapper from a payload converter, if present.""" + if isinstance(payload_converter, _TemporalTransferTypePayloadConverter): + return payload_converter._inner_payload_converter + return payload_converter + def to_payloads( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 1f8f7c6d2..1163cae3c 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -59,7 +59,9 @@ def _current_user_converters() -> _SystemNexusUserConverters: def _current_user_payload_converter() -> temporalio.converter.PayloadConverter: # pyright: ignore[reportUnusedFunction] """Return the active user payload converter for system Nexus model conversion.""" - return _current_user_converters().payload_converter + return _TemporalTransferTypePayloadConverter.unwrap( + _current_user_converters().payload_converter + ) def _current_user_failure_converter() -> temporalio.converter.FailureConverter: # pyright: ignore[reportUnusedFunction] diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 89625d64c..108431021 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -60,6 +60,9 @@ import temporalio.nexus.system import temporalio.workflow from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo +from temporalio.converter._payload_converter import ( + _TemporalTransferTypePayloadConverter, +) from temporalio.nexus.system.workflow_service._system_nexus_interceptor import ( _start_system_nexus_operation, _SystemNexusWorkflowOutboundInterceptorTerminal, @@ -1468,7 +1471,9 @@ def workflow_patch(self, id: str, *, deprecated: bool) -> bool: return use_patch def workflow_payload_converter(self) -> temporalio.converter.PayloadConverter: - return self._workflow_context_payload_converter + return _TemporalTransferTypePayloadConverter.unwrap( + self._workflow_context_payload_converter + ) def workflow_random(self) -> random.Random: self._assert_not_read_only("random") diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 49200aeab..a97a821dc 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -99,10 +99,11 @@ def from_transfer_type( class _TrackingFailureConverter(temporalio.converter.DefaultFailureConverter): def __init__( - self, expected_payload_converter: temporalio.converter.PayloadConverter + self, + expected_payload_converter_type: type[temporalio.converter.PayloadConverter], ) -> None: super().__init__() - self.expected_payload_converter = expected_payload_converter + self.expected_payload_converter_type = expected_payload_converter_type self.to_failure_calls = 0 self.from_failure_calls = 0 @@ -112,7 +113,7 @@ def to_failure( payload_converter: temporalio.converter.PayloadConverter, failure: temporalio.api.failure.v1.Failure, ) -> None: - assert payload_converter is self.expected_payload_converter + assert isinstance(payload_converter, self.expected_payload_converter_type) self.to_failure_calls += 1 super().to_failure(exception, payload_converter, failure) @@ -121,7 +122,7 @@ def from_failure( failure: temporalio.api.failure.v1.Failure, payload_converter: temporalio.converter.PayloadConverter, ) -> BaseException: - assert payload_converter is self.expected_payload_converter + assert isinstance(payload_converter, self.expected_payload_converter_type) self.from_failure_calls += 1 return super().from_failure(failure, payload_converter) @@ -708,7 +709,7 @@ def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: def test_system_nexus_uses_user_failure_converter() -> None: payload_converter = temporalio.converter.default().payload_converter - failure_converter = _TrackingFailureConverter(payload_converter) + failure_converter = _TrackingFailureConverter(DefaultPayloadConverter) system_converter = nexus_system._get_payload_converter( payload_converter, failure_converter ) @@ -745,7 +746,7 @@ def to_failure( payload_converter: temporalio.converter.PayloadConverter, failure: temporalio.api.failure.v1.Failure, ) -> None: - assert payload_converter is inner_data_converter.payload_converter + assert isinstance(payload_converter, DefaultPayloadConverter) raise ValueError("conversion failed") inner_system_converter = nexus_system._get_payload_converter( diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 793663ca9..c99d023e1 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -160,7 +160,9 @@ def __init__(self): @activity.defn async def passthrough_activity(input: TraceData) -> TraceData: - activity.payload_converter().to_payload(input) + payload_converter = activity.payload_converter() + assert isinstance(payload_converter, SerializationContextCompositePayloadConverter) + payload_converter.to_payload(input) activity.heartbeat(input) # Wait for the heartbeat to be processed so that it modifies the data before the activity returns await asyncio.sleep(0.2) @@ -178,7 +180,11 @@ async def run(self, data: TraceData) -> TraceData: class PayloadConversionWorkflow: @workflow.run async def run(self, data: TraceData) -> TraceData: - workflow.payload_converter().to_payload(data) + payload_converter = workflow.payload_converter() + assert isinstance( + payload_converter, SerializationContextCompositePayloadConverter + ) + payload_converter.to_payload(data) data = await workflow.execute_activity( passthrough_activity, data, From 8fe740ea17c26a1ba1ef21f0b39f5d67d9a6357b Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 15:54:35 -0700 Subject: [PATCH 9/9] Fix release changelog notes command (#1866) --- .github/scripts/release_verify.py | 2 -- tests/test_prepare_release.py | 2 -- 2 files changed, 4 deletions(-) diff --git a/.github/scripts/release_verify.py b/.github/scripts/release_verify.py index b3a98e49c..0e1319164 100644 --- a/.github/scripts/release_verify.py +++ b/.github/scripts/release_verify.py @@ -271,8 +271,6 @@ def _sdk_core_changelog_entries( "run", "--quiet", "-p", - "temporalio-sdk-core", - "--bin", "changelog-release-notes", "--", "--from", diff --git a/tests/test_prepare_release.py b/tests/test_prepare_release.py index b2f80be13..b4c121e8f 100644 --- a/tests/test_prepare_release.py +++ b/tests/test_prepare_release.py @@ -55,8 +55,6 @@ def check_output(args: list[str], *, cwd: pathlib.Path, **_kwargs: object) -> st "run", "--quiet", "-p", - "temporalio-sdk-core", - "--bin", "changelog-release-notes", "--", "--from",