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/CHANGELOG.md b/CHANGELOG.md index f1a526e7e..f016f34ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,24 @@ to include examples, links to docs, or any other relevant information. ### Added +- Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run. +### Changed + +### Deprecated + +### :boom: Breaking Changes + +### 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 + +### Added + #### Standalone Activity operator commands - `ActivityHandle` now supports operator commands for standalone activities: `pause`, @@ -27,17 +45,24 @@ 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 +- 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. - 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`. - -### Deprecated +- The experimental `GetNexusOperationResultInput` now includes the Nexus endpoint, service, and + operation. ### :boom: Breaking Changes @@ -56,6 +81,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 @@ -87,8 +115,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..b7077785f 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" @@ -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/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/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/_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 b9d82d6ea..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 ( @@ -202,11 +200,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 +225,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 +236,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 +294,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 +308,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 +320,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] @@ -657,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 @@ -807,27 +784,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 +825,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/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/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`. """ 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..cb56b0045 --- /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..40f349797 --- /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..ba11f43fe --- /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/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/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/_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/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/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/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/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..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") @@ -2194,14 +2199,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 +2471,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 +3667,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 +3676,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/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..52055ad58 --- /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" 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/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/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_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", diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 8d65d5f1f..c99d023e1 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 @@ -156,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) @@ -174,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, @@ -1688,25 +1698,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) - async def _assert_context_iff_not_nexus( + 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 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 +1790,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 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 ( diff --git a/uv.lock b/uv.lock index 767c4e587..ab1c17005 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 = [ @@ -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]] @@ -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" }, @@ -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" }, @@ -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 = [