diff --git a/CHANGELOG.md b/CHANGELOG.md index ed9ded4a5..f1d184c32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,84 @@ to include examples, links to docs, or any other relevant information. through `ExternalOutputStreamClient`. Workflow output is staged outside History and becomes readable only after its compact Workflow Task marker is committed. +### 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`. +- The experimental `GetNexusOperationResultInput` now includes the Nexus endpoint, service, and + operation. + +### :boom: Breaking Changes + +- Experimental external storage: `ExternalStorage.driver_selector` is now called with a + `StorageDriverSelectContext` instead of a `StorageDriverStoreContext`. Update the annotation; + the new type carries the same `target` field. Since selectors are plain callables, a stale + annotation fails type checking rather than at runtime. +- `client.ActivityExecution` and `client.ActivityExecutionDescription` had some fields removed or renamed + to match RPC API. + - Dataclass parameters for these types were changed to `frozen=True, eq=False, kw_only=True`. + - `scheduled_time` was renamed `schedule_time`. + - `last_failure` was changed from field to method that runs data converter on demand. + - `state_transition_count`, `eager_execution_requested`, `paused` and `long_poll_token` were removed. +- ActivityHandle.describe() long-poll token was removed. The functionality can still be used manually + through raw gRPC API. + +### 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 + resumes from the carried transcript and never re-executes prior dispatches, + so a carried cache entry could only serve stale results). Patch-gated + (`deepagents.retire-result-cache`), so histories recorded before this change + replay unchanged; note that deferring the patch keeps the full legacy dedup + cache — including the stale-result behavior this entry describes — and that + a chain upgraded mid-continue-as-new re-executes rather than reuses a + repeated identical call (the conservative direction). +- `contrib.deepagents`: summarization middleware configured with a model name string now routes its LLM calls through Activities instead of running them in the Workflow. + +- **Experimental**: External storage metrics now report the wall-clock time storage was in flight. + Previously each batch's duration was summed, over-reporting the time whenever storage operations + ran concurrently. +- System Nexus Signal-with-Start workflow operations now give custom payload + converters the target workflow's serialization context when encoding their + inner request payloads. +- Cancelling an activity from a signal while the workflow itself is cancelled + no longer causes a nondeterminism error from duplicate activity-cancellation + commands. +- `StrandsPlugin` now disables Botocore retries for its default Bedrock model so + model request retries are handled exclusively by Temporal. +- `temporalio.contrib.openai_agents` now honors the `retry-after-ms` and + `retry-after` headers when OpenAI returns `x-should-retry: true`. Previously + the delay the server asked for was discarded on that path and the activity + retried on its configured interval instead. +- Nexus-context workflow/activity starts no longer set `on_conflict_options` when there are no links + or callbacks to attach. +- The workflow sandbox now passes `pydantic_core` through by default, alongside `pydantic`. + +## [1.32.0] - 2026-08-24 + +### Added + +- Added `temporalio.converter.create_payload_validation_error` to create the + non-retryable application error used when a converted payload fails validation. +- Added experimental `temporalio.contrib.opentelemetry.ReplaySafeMeterProvider` and + `ReplaySafeLoggerProvider` (and exported `ReplaySafeTracerProvider`): wrap an + OpenTelemetry provider so metrics and log events recorded from workflow code (e.g. by + Google ADK) are not duplicated on replay. `GoogleAdkPlugin` warns when a global OTel + provider is not replay-safe. +- Added `LoggingConfig.format` to select compact, pretty, or newline-delimited JSON output for + Core logs written to the console. - Added the `Runtime(disable_environment_info=...)` option to control whether runtime, hosting, and platform information is included in worker heartbeats. diff --git a/pyproject.toml b/pyproject.toml index a6ef530b9..70f3d2c97 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,7 +45,7 @@ lambda-worker-otel = [ "opentelemetry-sdk-extension-aws>=2.0.0,<3", ] 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] @@ -188,8 +188,13 @@ exclude = [ # Ignore generated code 'temporalio/api', 'temporalio/bridge/proto', + 'temporalio/nexus/system/workflow_service', ] +[[tool.mypy.overrides]] +module = "temporalio.nexus.system.workflow_service.*" +ignore_errors = true + [tool.pydocstyle] convention = "google" # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088 diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index 7ac1bd2de..e57ca03b3 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -35,15 +35,28 @@ / "v1" / "request_response.proto" ) +NEX_GEN_VERSION = "0.2.4" def nex_gen_command() -> list[str]: if bin_path := os.environ.get("NEX_GEN_BIN"): return [bin_path] - if shutil.which("nex-gen") is None: - subprocess.check_call(["cargo", "install", "--locked", "nex-gen", "--force"]) - return ["nex-gen"] + if shutil.which("nexgen") is None: + subprocess.check_call( + [ + "cargo", + "install", + "--locked", + "nexgen", + "--version", + NEX_GEN_VERSION, + "--features", + "advanced", + "--force", + ] + ) + return ["nexgen"] def build_descriptor_set(descriptor_path: Path) -> None: @@ -114,13 +127,11 @@ def generate_nexus_system_api() -> None: subprocess.check_call( [ *command, - "generate", - "--lang", "python", - "--input", str(wit_path), - "--input", str(wit_deps_dir), + "--native-api", + "--system-nexus", "--support-file", str(python_support_path), "--descriptors", diff --git a/scripts/gen_payload_visitor.py b/scripts/gen_payload_visitor.py index 0001659f7..7fab85dcd 100644 --- a/scripts/gen_payload_visitor.py +++ b/scripts/gen_payload_visitor.py @@ -2,7 +2,7 @@ import sys from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path -from typing import cast +from typing import cast, get_args, get_origin, get_type_hints import google.protobuf.message import nexusrpc @@ -19,14 +19,15 @@ from temporalio.bridge.proto.workflow_completion.workflow_completion_pb2 import ( WorkflowActivationCompletion, ) +from temporalio.converter._payload_converter import _get_transfer_type_converter def discover_system_nexus_roots() -> list[Descriptor]: - module_path = ( - base_dir / "temporalio" / "nexus" / "system" / "workflow_service" / "service.py" - ) + module_path = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" spec = spec_from_file_location( - "temporalio_nexus_system_workflow_service", module_path + "temporalio_nexus_system_workflow_service", + module_path / "__init__.py", + submodule_search_locations=[str(module_path)], ) if spec is None or spec.loader is None: raise RuntimeError(f"Cannot load generated system service from {module_path}") @@ -35,10 +36,14 @@ def discover_system_nexus_roots() -> list[Descriptor]: spec.loader.exec_module(module) roots: list[Descriptor] = [] - for operation in vars(module.WorkflowService).values(): - if not isinstance(operation, nexusrpc.Operation): + for annotation in get_type_hints(module._services.WorkflowService).values(): + if get_origin(annotation) is not nexusrpc.Operation: continue - for proto_type in (operation.input_type, operation.output_type): + for operation_type in get_args(annotation): + converter = _get_transfer_type_converter(operation_type) + proto_type = ( + converter.transfer_type if converter is not None else operation_type + ) if isinstance(proto_type, type) and issubclass( proto_type, google.protobuf.message.Message ): @@ -119,12 +124,20 @@ def generate(self, roots: list[Descriptor]) -> str: The generated code defines async visitor functions for each reachable protobuf message type starting from WorkflowActivation, including support - for repeated fields and map entries, and a convenience entrypoint - function `visit`. + for repeated fields and map entries. Payload-free roots get no-op methods + so the `visit` entrypoint recognizes them as supported. """ - for r in roots: - self.walk(r) + for root in roots: + if not self.walk(root): + self.methods.append( + f"""\ + async def _visit_{name_for(root)}( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass +""" + ) header = """ from __future__ import annotations diff --git a/scripts/nex_gen_support.py b/scripts/nex_gen_support.py index 58b51e263..fad21151e 100644 --- a/scripts/nex_gen_support.py +++ b/scripts/nex_gen_support.py @@ -10,6 +10,12 @@ import temporalio.api.workflow.v1 import temporalio.common import temporalio.converter +import temporalio.nexus.system + + +class SignalWithStartWorkflowModelRequest(typing.Protocol): + namespace: str + id: str def retry_policy_from_proto( @@ -52,6 +58,12 @@ def workflow_type_to_proto( return common_pb2.WorkflowType(name=workflow_function_name(workflow_type)) +def workflow_type_from_proto( + proto: common_pb2.WorkflowType, +) -> str: + return proto.name + + def task_queue_from_proto( proto: taskqueue_pb2.TaskQueue, ) -> str: @@ -70,12 +82,33 @@ def workflow_namespace() -> str: return info().namespace +def signal_with_start_workflow_serialization_context( + request: SignalWithStartWorkflowModelRequest, +) -> temporalio.converter.WorkflowSerializationContext: + return temporalio.converter.WorkflowSerializationContext( + namespace=request.namespace, + workflow_id=request.id, + ) + + def payloads_to_proto( values: collections.abc.Sequence[typing.Any], ) -> common_pb2.Payloads: - from temporalio.workflow import payload_converter + return ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + values + ) + ) - return payload_converter().to_payloads_wrapper(values) + +def payloads_from_proto( + proto: common_pb2.Payloads, +) -> list[object]: + return list( + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + proto + ) + ) def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: @@ -87,20 +120,23 @@ def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: def _value_to_payload(value: object | common_pb2.Payload) -> common_pb2.Payload: if isinstance(value, common_pb2.Payload): return _clone_payload(value) - from temporalio.workflow import payload_converter - - payloads = payload_converter().to_payloads_wrapper([value]) + payloads = ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + [value] + ) + ) return _clone_payload(payloads.payloads[0]) def _payload_to_value(payload: common_pb2.Payload) -> object: wrapper = common_pb2.Payloads() wrapper.payloads.add().CopyFrom(payload) - from temporalio.workflow import payload_converter return typing.cast( object, - payload_converter().from_payloads_wrapper(wrapper)[0], + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + wrapper + )[0], ) @@ -131,6 +167,21 @@ def memo_to_proto( return message +def header_from_proto( + proto: common_pb2.Header, +) -> collections.abc.Mapping[str, object]: + return {key: _payload_to_value(value) for key, value in proto.fields.items()} + + +def header_to_proto( + header: collections.abc.Mapping[str, object], +) -> common_pb2.Header: + message = common_pb2.Header() + for key, value in header.items(): + message.fields[key].CopyFrom(_value_to_payload(value)) + return message + + def duration_from_proto(proto: google.protobuf.duration_pb2.Duration) -> timedelta: return proto.ToTimedelta() @@ -177,6 +228,12 @@ def search_attributes_to_proto( return proto +def search_attributes_from_proto( + proto: common_pb2.SearchAttributes, +) -> temporalio.common.TypedSearchAttributes: + return temporalio.converter.decode_typed_search_attributes(proto) + + def priority_from_proto( proto: common_pb2.Priority, ) -> temporalio.common.Priority: @@ -193,3 +250,25 @@ def versioning_override_to_proto( versioning_override: temporalio.common.VersioningOverride, ) -> temporalio.api.workflow.v1.VersioningOverride: return versioning_override._to_proto() # pyright: ignore[reportPrivateUsage] + + +def versioning_override_from_proto( + proto: temporalio.api.workflow.v1.VersioningOverride, +) -> temporalio.common.VersioningOverride: + if proto.HasField("pinned") and proto.pinned.HasField("version"): + version = proto.pinned.version + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion( + deployment_name=version.deployment_name, + build_id=version.build_id, + ) + ) + if proto.pinned_version: + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion.from_canonical_string( + proto.pinned_version + ) + ) + if proto.auto_upgrade: + return temporalio.common.AutoUpgradeVersioningOverride() + raise ValueError("unknown versioning override proto shape") diff --git a/temporalio/bridge/_visitor.py b/temporalio/bridge/_visitor.py index ef1e21dbd..f26fc390f 100644 --- a/temporalio/bridge/_visitor.py +++ b/temporalio/bridge/_visitor.py @@ -621,3 +621,8 @@ async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutio await self._visit_temporal_api_common_v1_Header(fs, o.header) if o.HasField("user_metadata"): await self._visit_temporal_api_sdk_v1_UserMetadata(fs, o.user_metadata) + + async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutionResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass 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/contrib/openai_agents/testing.py b/temporalio/contrib/openai_agents/testing.py index 110ca20b7..3fcd6a573 100644 --- a/temporalio/contrib/openai_agents/testing.py +++ b/temporalio/contrib/openai_agents/testing.py @@ -1,5 +1,6 @@ """Testing utilities for OpenAI agents.""" +import uuid from collections.abc import AsyncIterator, Callable, Sequence from typing import Any @@ -73,7 +74,9 @@ def tool_call(arguments: str, name: str) -> ModelResponse: return ResponseBuilders.model_response( ResponseFunctionToolCall( arguments=arguments, - call_id="call", + # The Agents SDK rejects two invocations sharing one completed + # call id, and a script that calls two tools would hand them one. + call_id=f"call-{uuid.uuid4()}", name=name, type="function_call", id="id", diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index 7c24fa125..8397ef186 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -1,8 +1,15 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen v0.2.4. DO NOT EDIT! from __future__ import annotations -from . import service as _service +import collections.abc +import typing + +import nexusrpc + +import temporalio.converter + +from . import services as _services from .operations.signal_with_start_workflow import signal_with_start_workflow __all__ = [ @@ -10,9 +17,33 @@ ] +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +_SerializationContextFactory = collections.abc.Callable[ + [_InputT], temporalio.converter.SerializationContext +] + + +class _NexusOperationInfo(typing.Generic[_InputT, _OutputT]): + def __init__( + self, + *, + operation: nexusrpc.Operation[_InputT, _OutputT], + serialization_context: _SerializationContextFactory[_InputT] | None = None, + ) -> None: + self.operation: nexusrpc.Operation[_InputT, _OutputT] = operation + self.serialization_context: _SerializationContextFactory[_InputT] | None = ( + serialization_context + ) + + __nexus_operation_registry__ = { ( "temporal.api.workflowservice.v1.WorkflowService", "SignalWithStartWorkflowExecution", - ): _service.WorkflowService.signal_with_start_workflow, + ): _NexusOperationInfo( + operation=_services.WorkflowService.signal_with_start_workflow, + ), } diff --git a/temporalio/nexus/system/workflow_service/_resources/__init__.py b/temporalio/nexus/system/workflow_service/_resources/__init__.py deleted file mode 100644 index 373efbd33..000000000 --- a/temporalio/nexus/system/workflow_service/_resources/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -# Generated by nex-gen. DO NOT EDIT! - -from __future__ import annotations - -__all__ = [] diff --git a/temporalio/nexus/system/workflow_service/_support/__init__.py b/temporalio/nexus/system/workflow_service/_support/__init__.py index 530c33e80..6af0c0c56 100644 --- a/temporalio/nexus/system/workflow_service/_support/__init__.py +++ b/temporalio/nexus/system/workflow_service/_support/__init__.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen v0.2.4. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py index 58b51e263..fad21151e 100644 --- a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py +++ b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py @@ -10,6 +10,12 @@ import temporalio.api.workflow.v1 import temporalio.common import temporalio.converter +import temporalio.nexus.system + + +class SignalWithStartWorkflowModelRequest(typing.Protocol): + namespace: str + id: str def retry_policy_from_proto( @@ -52,6 +58,12 @@ def workflow_type_to_proto( return common_pb2.WorkflowType(name=workflow_function_name(workflow_type)) +def workflow_type_from_proto( + proto: common_pb2.WorkflowType, +) -> str: + return proto.name + + def task_queue_from_proto( proto: taskqueue_pb2.TaskQueue, ) -> str: @@ -70,12 +82,33 @@ def workflow_namespace() -> str: return info().namespace +def signal_with_start_workflow_serialization_context( + request: SignalWithStartWorkflowModelRequest, +) -> temporalio.converter.WorkflowSerializationContext: + return temporalio.converter.WorkflowSerializationContext( + namespace=request.namespace, + workflow_id=request.id, + ) + + def payloads_to_proto( values: collections.abc.Sequence[typing.Any], ) -> common_pb2.Payloads: - from temporalio.workflow import payload_converter + return ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + values + ) + ) - return payload_converter().to_payloads_wrapper(values) + +def payloads_from_proto( + proto: common_pb2.Payloads, +) -> list[object]: + return list( + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + proto + ) + ) def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: @@ -87,20 +120,23 @@ def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: def _value_to_payload(value: object | common_pb2.Payload) -> common_pb2.Payload: if isinstance(value, common_pb2.Payload): return _clone_payload(value) - from temporalio.workflow import payload_converter - - payloads = payload_converter().to_payloads_wrapper([value]) + payloads = ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + [value] + ) + ) return _clone_payload(payloads.payloads[0]) def _payload_to_value(payload: common_pb2.Payload) -> object: wrapper = common_pb2.Payloads() wrapper.payloads.add().CopyFrom(payload) - from temporalio.workflow import payload_converter return typing.cast( object, - payload_converter().from_payloads_wrapper(wrapper)[0], + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + wrapper + )[0], ) @@ -131,6 +167,21 @@ def memo_to_proto( return message +def header_from_proto( + proto: common_pb2.Header, +) -> collections.abc.Mapping[str, object]: + return {key: _payload_to_value(value) for key, value in proto.fields.items()} + + +def header_to_proto( + header: collections.abc.Mapping[str, object], +) -> common_pb2.Header: + message = common_pb2.Header() + for key, value in header.items(): + message.fields[key].CopyFrom(_value_to_payload(value)) + return message + + def duration_from_proto(proto: google.protobuf.duration_pb2.Duration) -> timedelta: return proto.ToTimedelta() @@ -177,6 +228,12 @@ def search_attributes_to_proto( return proto +def search_attributes_from_proto( + proto: common_pb2.SearchAttributes, +) -> temporalio.common.TypedSearchAttributes: + return temporalio.converter.decode_typed_search_attributes(proto) + + def priority_from_proto( proto: common_pb2.Priority, ) -> temporalio.common.Priority: @@ -193,3 +250,25 @@ def versioning_override_to_proto( versioning_override: temporalio.common.VersioningOverride, ) -> temporalio.api.workflow.v1.VersioningOverride: return versioning_override._to_proto() # pyright: ignore[reportPrivateUsage] + + +def versioning_override_from_proto( + proto: temporalio.api.workflow.v1.VersioningOverride, +) -> temporalio.common.VersioningOverride: + if proto.HasField("pinned") and proto.pinned.HasField("version"): + version = proto.pinned.version + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion( + deployment_name=version.deployment_name, + build_id=version.build_id, + ) + ) + if proto.pinned_version: + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion.from_canonical_string( + proto.pinned_version + ) + ) + if proto.auto_upgrade: + return temporalio.common.AutoUpgradeVersioningOverride() + raise ValueError("unknown versioning override proto shape") diff --git a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py new file mode 100644 index 000000000..c70b106db --- /dev/null +++ b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py @@ -0,0 +1,95 @@ +# Generated by nexgen v0.2.4. DO NOT EDIT! + +from __future__ import annotations + +import abc +import typing + +from . import models + +if typing.TYPE_CHECKING: + import temporalio.workflow + from temporalio.worker._interceptor import StartNexusOperationInput + + +__all__ = [ + "_start_system_nexus_operation", + "_SystemNexusWorkflowOutboundInterceptorBase", + "_SystemNexusWorkflowOutboundInterceptorTerminal", +] + + +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +async def _start_system_nexus_operation( + interceptor: _SystemNexusWorkflowOutboundInterceptorBase, + input: StartNexusOperationInput[_InputT, _OutputT], +) -> temporalio.workflow.NexusOperationHandle[_OutputT]: + if ( + input.service == "temporal.api.workflowservice.v1.WorkflowService" + and input.operation_name == "SignalWithStartWorkflowExecution" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.SignalWithStartWorkflowRequest, models.SignalWithStartWorkflowResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_signal_with_start_workflow(typed_input.input), + ) + raise ValueError( + f"unsupported System Nexus operation: {input.service}/{input.operation_name}" + ) + + +class _SystemNexusWorkflowOutboundInterceptorBase(abc.ABC): + @abc.abstractmethod + def _next_system_nexus_interceptor( + self, + ) -> _SystemNexusWorkflowOutboundInterceptorBase: ... + + async def start_signal_with_start_workflow( + self, request: models.SignalWithStartWorkflowRequest + ) -> temporalio.workflow.NexusOperationHandle[ + models.SignalWithStartWorkflowResponse + ]: + """Intercept the SignalWithStartWorkflow operation.""" + return await self._next_system_nexus_interceptor().start_signal_with_start_workflow( + request + ) + + +class _SystemNexusWorkflowOutboundInterceptorTerminal(abc.ABC): + @abc.abstractmethod + async def _intercept_system_nexus_operation( + self, + input: StartNexusOperationInput[_InputT, _OutputT], + ) -> temporalio.workflow.NexusOperationHandle[_OutputT]: ... + + async def start_signal_with_start_workflow( + self, request: models.SignalWithStartWorkflowRequest + ) -> temporalio.workflow.NexusOperationHandle[ + models.SignalWithStartWorkflowResponse + ]: + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.workflowservice.v1.WorkflowService", + operation="SignalWithStartWorkflowExecution", + input=request, + output_type=models.SignalWithStartWorkflowResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index 05e1e3088..df3694863 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen v0.2.4. DO NOT EDIT! from __future__ import annotations @@ -7,25 +7,39 @@ import datetime import typing +import typing_extensions + import temporalio.api.sdk.v1.user_metadata_pb2 import temporalio.api.workflowservice.v1.request_response_pb2 import temporalio.common +import temporalio.converter from ._support import ( + duration_from_proto, duration_to_proto, + memo_from_proto, memo_to_proto, payload_from_proto, payload_to_proto, + payloads_from_proto, payloads_to_proto, + priority_from_proto, priority_to_proto, + retry_policy_from_proto, retry_policy_to_proto, + search_attributes_from_proto, search_attributes_to_proto, signal_function_to_proto, + task_queue_from_proto, task_queue_to_proto, + versioning_override_from_proto, versioning_override_to_proto, + workflow_id_conflict_policy_from_proto, workflow_id_conflict_policy_to_proto, + workflow_id_reuse_policy_from_proto, workflow_id_reuse_policy_to_proto, workflow_namespace, + workflow_type_from_proto, workflow_type_to_proto, ) @@ -59,83 +73,262 @@ class SignalWithStartWorkflowRequest: versioning_override: temporalio.common.VersioningOverride | None = None start_delay: datetime.timedelta | None = None user_metadata: UserMetadata | None = None + namespace: str = dataclasses.field(default_factory=workflow_namespace) + - def to_proto( +class _SignalWithStartWorkflowRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + SignalWithStartWorkflowRequest, + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, + ] +): + transfer_type: ( + type[ + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest + ] + | None + ) = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest + + @typing_extensions.override + def from_transfer_type( self, + value: temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, + type_hint: type[SignalWithStartWorkflowRequest], + ) -> SignalWithStartWorkflowRequest: + if not value.HasField("workflow_type"): + raise ValueError( + "missing required field SignalWithStartWorkflowRequest.workflow" + ) + workflow = workflow_type_from_proto(value.workflow_type) + if not value.workflow_id: + raise ValueError("missing required field SignalWithStartWorkflowRequest.id") + id = value.workflow_id + if not value.HasField("task_queue"): + raise ValueError( + "missing required field SignalWithStartWorkflowRequest.task_queue" + ) + task_queue = task_queue_from_proto(value.task_queue) + if not value.signal_name: + raise ValueError( + "missing required field SignalWithStartWorkflowRequest.signal" + ) + signal = value.signal_name + return SignalWithStartWorkflowRequest( + workflow=workflow, + args=payloads_from_proto(value.input) if value.HasField("input") else None, + id=id, + task_queue=task_queue, + signal=signal, + signal_args=payloads_from_proto(value.signal_input) + if value.HasField("signal_input") + else None, + execution_timeout=duration_from_proto(value.workflow_execution_timeout) + if value.HasField("workflow_execution_timeout") + else None, + run_timeout=duration_from_proto(value.workflow_run_timeout) + if value.HasField("workflow_run_timeout") + else None, + task_timeout=duration_from_proto(value.workflow_task_timeout) + if value.HasField("workflow_task_timeout") + else None, + request_id=value.request_id if bool(value.request_id) else None, + id_reuse_policy=workflow_id_reuse_policy_from_proto( + value.workflow_id_reuse_policy + ), + id_conflict_policy=workflow_id_conflict_policy_from_proto( + value.workflow_id_conflict_policy + ) + if value.workflow_id_conflict_policy != 0 + else None, + retry_policy=retry_policy_from_proto(value.retry_policy) + if value.HasField("retry_policy") + else None, + cron_schedule=value.cron_schedule if bool(value.cron_schedule) else None, + memo=memo_from_proto(value.memo) if value.HasField("memo") else None, + search_attributes=search_attributes_from_proto(value.search_attributes) + if value.HasField("search_attributes") + else None, + priority=priority_from_proto(value.priority) + if value.HasField("priority") + else None, + versioning_override=versioning_override_from_proto( + value.versioning_override + ) + if value.HasField("versioning_override") + else None, + start_delay=duration_from_proto(value.workflow_start_delay) + if value.HasField("workflow_start_delay") + else None, + user_metadata=_UserMetadataTransferTypeConverter().from_transfer_type( + value.user_metadata, UserMetadata + ) + if value.HasField("user_metadata") + else None, + namespace=value.namespace, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: SignalWithStartWorkflowRequest, ) -> temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest: message = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest() - message.workflow_type.CopyFrom(workflow_type_to_proto(self.workflow)) - if self.args is not None: - message.input.CopyFrom(payloads_to_proto(self.args)) - message.workflow_id = self.id - message.task_queue.CopyFrom(task_queue_to_proto(self.task_queue)) - message.signal_name = signal_function_to_proto(self.signal) - if self.signal_args is not None: - message.signal_input.CopyFrom(payloads_to_proto(self.signal_args)) - if self.execution_timeout is not None: + message.workflow_type.CopyFrom(workflow_type_to_proto(value.workflow)) + if value.args is not None: + message.input.CopyFrom(payloads_to_proto(value.args)) + message.workflow_id = value.id + message.task_queue.CopyFrom(task_queue_to_proto(value.task_queue)) + message.signal_name = signal_function_to_proto(value.signal) + if value.signal_args is not None: + message.signal_input.CopyFrom(payloads_to_proto(value.signal_args)) + if value.execution_timeout is not None: message.workflow_execution_timeout.CopyFrom( - duration_to_proto(self.execution_timeout) + duration_to_proto(value.execution_timeout) + ) + if value.run_timeout is not None: + message.workflow_run_timeout.CopyFrom(duration_to_proto(value.run_timeout)) + if value.task_timeout is not None: + message.workflow_task_timeout.CopyFrom( + duration_to_proto(value.task_timeout) ) - if self.run_timeout is not None: - message.workflow_run_timeout.CopyFrom(duration_to_proto(self.run_timeout)) - if self.task_timeout is not None: - message.workflow_task_timeout.CopyFrom(duration_to_proto(self.task_timeout)) - if self.request_id is not None: - message.request_id = self.request_id + if value.request_id is not None: + message.request_id = value.request_id message.workflow_id_reuse_policy = workflow_id_reuse_policy_to_proto( - self.id_reuse_policy + value.id_reuse_policy ) - if self.id_conflict_policy is not None: + if value.id_conflict_policy is not None: message.workflow_id_conflict_policy = workflow_id_conflict_policy_to_proto( - self.id_conflict_policy + value.id_conflict_policy ) - if self.retry_policy is not None: - message.retry_policy.CopyFrom(retry_policy_to_proto(self.retry_policy)) - if self.cron_schedule is not None: - message.cron_schedule = self.cron_schedule - if self.memo is not None: - message.memo.CopyFrom(memo_to_proto(self.memo)) - if self.search_attributes is not None: + if value.retry_policy is not None: + message.retry_policy.CopyFrom(retry_policy_to_proto(value.retry_policy)) + if value.cron_schedule is not None: + message.cron_schedule = value.cron_schedule + if value.memo is not None: + message.memo.CopyFrom(memo_to_proto(value.memo)) + if value.search_attributes is not None: message.search_attributes.CopyFrom( - search_attributes_to_proto(self.search_attributes) + search_attributes_to_proto(value.search_attributes) ) - if self.priority is not None: - message.priority.CopyFrom(priority_to_proto(self.priority)) - if self.versioning_override is not None: + if value.priority is not None: + message.priority.CopyFrom(priority_to_proto(value.priority)) + if value.versioning_override is not None: message.versioning_override.CopyFrom( - versioning_override_to_proto(self.versioning_override) + versioning_override_to_proto(value.versioning_override) ) - if self.start_delay is not None: - message.workflow_start_delay.CopyFrom(duration_to_proto(self.start_delay)) - if self.user_metadata is not None: - message.user_metadata.CopyFrom(self.user_metadata.to_proto()) - message.namespace = workflow_namespace() + if value.start_delay is not None: + message.workflow_start_delay.CopyFrom(duration_to_proto(value.start_delay)) + if value.user_metadata is not None: + message.user_metadata.CopyFrom( + _UserMetadataTransferTypeConverter().to_transfer_type( + value.user_metadata + ) + ) + message.namespace = value.namespace return message +_ = temporalio.converter.transfer_type_convertible( + _SignalWithStartWorkflowRequestTransferTypeConverter +)(SignalWithStartWorkflowRequest) + + @dataclasses.dataclass(slots=True) class UserMetadata: static_summary: typing.Any | None = None static_details: typing.Any | None = None - @classmethod - def from_proto( - cls, - proto: temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata, + +class _UserMetadataTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + UserMetadata, temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata + ] +): + transfer_type: type[temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata] | None = ( + temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata, + type_hint: type[UserMetadata], ) -> UserMetadata: - return cls( - static_summary=payload_from_proto(proto.summary) - if proto.HasField("summary") + return UserMetadata( + static_summary=payload_from_proto(value.summary) + if value.HasField("summary") else None, - static_details=payload_from_proto(proto.details) - if proto.HasField("details") + static_details=payload_from_proto(value.details) + if value.HasField("details") else None, ) - def to_proto(self) -> temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata: + @typing_extensions.override + def to_transfer_type( + self, + value: UserMetadata, + ) -> temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata: message = temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata() - if self.static_summary is not None: - message.summary.CopyFrom(payload_to_proto(self.static_summary)) - if self.static_details is not None: - message.details.CopyFrom(payload_to_proto(self.static_details)) + if value.static_summary is not None: + message.summary.CopyFrom(payload_to_proto(value.static_summary)) + if value.static_details is not None: + message.details.CopyFrom(payload_to_proto(value.static_details)) + return message + + +_ = temporalio.converter.transfer_type_convertible(_UserMetadataTransferTypeConverter)( + UserMetadata +) + + +@dataclasses.dataclass(slots=True) +class SignalWithStartWorkflowResponse: + """ + .. warning:: + This API is experimental and subject to change. + """ + + run_id: str | None = None + started: bool | None = None + + +class _SignalWithStartWorkflowResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + SignalWithStartWorkflowResponse, + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + ] +): + transfer_type: ( + type[ + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse + ] + | None + ) = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + type_hint: type[SignalWithStartWorkflowResponse], + ) -> SignalWithStartWorkflowResponse: + return SignalWithStartWorkflowResponse( + run_id=value.run_id if bool(value.run_id) else None, + started=value.started if bool(value.started) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: SignalWithStartWorkflowResponse, + ) -> temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse: + message = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse() + if value.run_id is not None: + message.run_id = value.run_id + if value.started is not None: + message.started = value.started return message + + +_ = temporalio.converter.transfer_type_convertible( + _SignalWithStartWorkflowResponseTransferTypeConverter +)(SignalWithStartWorkflowResponse) diff --git a/temporalio/nexus/system/workflow_service/operations/__init__.py b/temporalio/nexus/system/workflow_service/operations/__init__.py index 67c9cc56b..28d71c338 100644 --- a/temporalio/nexus/system/workflow_service/operations/__init__.py +++ b/temporalio/nexus/system/workflow_service/operations/__init__.py @@ -1,3 +1,3 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen v0.2.4. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py index 0865e5a88..9060fc11c 100644 --- a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py +++ b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen v0.2.4. DO NOT EDIT! from __future__ import annotations @@ -8,7 +8,6 @@ import typing_extensions -import temporalio.api.workflowservice.v1.request_response_pb2 import temporalio.common if typing.TYPE_CHECKING: @@ -16,6 +15,7 @@ from ..models import ( SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, UserMetadata, ) @@ -33,15 +33,14 @@ async def _signal_with_start_workflow( get_external_workflow_handle, ) - request_proto = request.to_proto() nexus_client = create_nexus_client( service="temporal.api.workflowservice.v1.WorkflowService", endpoint="__temporal_system", ) handle = await nexus_client.start_operation( operation="SignalWithStartWorkflowExecution", - input=request_proto, - output_type=temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + input=request, + output_type=SignalWithStartWorkflowResponse, ) result = await handle return get_external_workflow_handle(request.id, run_id=result.run_id) @@ -77,17 +76,17 @@ async def signal_with_start_workflow( # Overload case: -# - workflow name with optional list-form workflow arguments -# - signal name with optional list-form signal arguments +# - workflow name with positional workflow arguments +# - signal method callable with no signal arguments @typing.overload async def signal_with_start_workflow( workflow: str, - *, - args: list[typing.Any] | None = ..., + *args: object, id: str, task_queue: str, - signal: str, - signal_args: list[typing.Any] | None = ..., + signal: collections.abc.Callable[ + [SelfType], None | collections.abc.Awaitable[None] + ], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -103,23 +102,22 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[object]: ... +) -> ExternalWorkflowHandle[SelfType]: ... # Overload case: -# - workflow method callable with typed positional workflow arguments -# - signal name with optional list-form signal arguments +# - workflow name with positional workflow arguments +# - signal method callable with a typed single signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], - *args: typing_extensions.Unpack[WorkflowArgs], + workflow: str, + *args: object, id: str, task_queue: str, - signal: str, - signal_args: list[typing.Any] | None = ..., + signal: collections.abc.Callable[ + [SelfType, SignalArg], None | collections.abc.Awaitable[None] + ], + signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -139,20 +137,16 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with list-form workflow arguments -# - signal name with optional list-form signal arguments +# - workflow name with positional workflow arguments +# - signal callable with list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], - *, - args: list[typing.Any], + workflow: str, + *args: object, id: str, task_queue: str, - signal: str, - signal_args: list[typing.Any] | None = ..., + signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], + signal_args: list[typing.Any], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -168,21 +162,21 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[SelfType]: ... +) -> ExternalWorkflowHandle[object]: ... # Overload case: -# - workflow name with positional workflow arguments -# - signal method callable with no signal arguments +# - workflow name with optional list-form workflow arguments +# - signal name with optional list-form signal arguments @typing.overload async def signal_with_start_workflow( workflow: str, - *args: object, + *, + args: list[typing.Any] | None = ..., id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType], None | collections.abc.Awaitable[None] - ], + signal: str, + signal_args: list[typing.Any] | None = ..., execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -198,7 +192,7 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[SelfType]: ... +) -> ExternalWorkflowHandle[object]: ... # Overload case: @@ -233,20 +227,19 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with typed positional workflow arguments -# - signal method callable with no signal arguments +# - workflow name with optional list-form workflow arguments +# - signal method callable with a typed single signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], - *args: typing_extensions.Unpack[WorkflowArgs], + workflow: str, + *, + args: list[typing.Any] | None = ..., id: str, task_queue: str, signal: collections.abc.Callable[ - [SelfType], None | collections.abc.Awaitable[None] + [SelfType, SignalArg], None | collections.abc.Awaitable[None] ], + signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -266,21 +259,17 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with list-form workflow arguments -# - signal method callable with no signal arguments +# - workflow name with optional list-form workflow arguments +# - signal callable with list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], + workflow: str, *, - args: list[typing.Any], + args: list[typing.Any] | None = ..., id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType], None | collections.abc.Awaitable[None] - ], + signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], + signal_args: list[typing.Any], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -296,22 +285,23 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[SelfType]: ... +) -> ExternalWorkflowHandle[object]: ... # Overload case: -# - workflow name with positional workflow arguments -# - signal method callable with a typed single signal arguments +# - workflow method callable with typed positional workflow arguments +# - signal name with optional list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, - *args: object, + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], + *args: typing_extensions.Unpack[WorkflowArgs], id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType, SignalArg], None | collections.abc.Awaitable[None] - ], - signal_args: SignalArg, + signal: str, + signal_args: list[typing.Any] | None = ..., execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -331,19 +321,20 @@ async def signal_with_start_workflow( # Overload case: -# - workflow name with optional list-form workflow arguments -# - signal method callable with a typed single signal arguments +# - workflow method callable with typed positional workflow arguments +# - signal method callable with no signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, - *, - args: list[typing.Any] | None = ..., + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], + *args: typing_extensions.Unpack[WorkflowArgs], id: str, task_queue: str, signal: collections.abc.Callable[ - [SelfType, SignalArg], None | collections.abc.Awaitable[None] + [SelfType], None | collections.abc.Awaitable[None] ], - signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -397,22 +388,19 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with list-form workflow arguments -# - signal method callable with a typed single signal arguments +# - workflow method callable with typed positional workflow arguments +# - signal callable with list-form signal arguments @typing.overload async def signal_with_start_workflow( workflow: collections.abc.Callable[ [SelfType, typing_extensions.Unpack[WorkflowArgs]], collections.abc.Awaitable[WorkflowResult], ], - *, - args: list[typing.Any], + *args: typing_extensions.Unpack[WorkflowArgs], id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType, SignalArg], None | collections.abc.Awaitable[None] - ], - signal_args: SignalArg, + signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], + signal_args: list[typing.Any], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -432,16 +420,20 @@ async def signal_with_start_workflow( # Overload case: -# - workflow name with positional workflow arguments -# - signal callable with list-form signal arguments +# - workflow method callable with list-form workflow arguments +# - signal name with optional list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, - *args: object, + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], + *, + args: list[typing.Any], id: str, task_queue: str, - signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], - signal_args: list[typing.Any], + signal: str, + signal_args: list[typing.Any] | None = ..., execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -457,21 +449,25 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[object]: ... +) -> ExternalWorkflowHandle[SelfType]: ... # Overload case: -# - workflow name with optional list-form workflow arguments -# - signal callable with list-form signal arguments +# - workflow method callable with list-form workflow arguments +# - signal method callable with no signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], *, - args: list[typing.Any] | None = ..., + args: list[typing.Any], id: str, task_queue: str, - signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], - signal_args: list[typing.Any], + signal: collections.abc.Callable[ + [SelfType], None | collections.abc.Awaitable[None] + ], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -487,23 +483,26 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[object]: ... +) -> ExternalWorkflowHandle[SelfType]: ... # Overload case: -# - workflow method callable with typed positional workflow arguments -# - signal callable with list-form signal arguments +# - workflow method callable with list-form workflow arguments +# - signal method callable with a typed single signal arguments @typing.overload async def signal_with_start_workflow( workflow: collections.abc.Callable[ [SelfType, typing_extensions.Unpack[WorkflowArgs]], collections.abc.Awaitable[WorkflowResult], ], - *args: typing_extensions.Unpack[WorkflowArgs], + *, + args: list[typing.Any], id: str, task_queue: str, - signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], - signal_args: list[typing.Any], + signal: collections.abc.Callable[ + [SelfType, SignalArg], None | collections.abc.Awaitable[None] + ], + signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -629,6 +628,11 @@ async def signal_with_start_workflow( Returns: A workflow handle to the started workflow. """ + if positional_args and args is not None: + raise TypeError("cannot specify both positional arguments and args") + normalized_args: list[typing.Any] | None = ( + list(positional_args) if positional_args else args + ) normalized_signal_args: list[typing.Any] | None if signal_args is None: normalized_signal_args = None @@ -636,11 +640,6 @@ async def signal_with_start_workflow( normalized_signal_args = typing.cast(list[typing.Any], signal_args) else: normalized_signal_args = [signal_args] - if positional_args and args is not None: - raise TypeError("cannot specify both positional arguments and args") - normalized_args: list[typing.Any] | None = ( - list(positional_args) if positional_args else args - ) user_metadata = ( None if static_summary is None and static_details is None diff --git a/temporalio/nexus/system/workflow_service/service.py b/temporalio/nexus/system/workflow_service/services.py similarity index 58% rename from temporalio/nexus/system/workflow_service/service.py rename to temporalio/nexus/system/workflow_service/services.py index 7ce5849ca..c538493e1 100644 --- a/temporalio/nexus/system/workflow_service/service.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -1,10 +1,13 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen v0.2.4. DO NOT EDIT! from __future__ import annotations from nexusrpc import Operation, service -import temporalio.api.workflowservice.v1.request_response_pb2 +from .models import ( + SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, +) @service(name="temporal.api.workflowservice.v1.WorkflowService") @@ -16,6 +19,6 @@ class WorkflowService: # .. warning:: This API is experimental and subject to change. signal_with_start_workflow: Operation[ - temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, - temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, ] = Operation(name="SignalWithStartWorkflowExecution") diff --git a/tests/contrib/google_adk_agents/test_adk_streaming.py b/tests/contrib/google_adk_agents/test_adk_streaming.py index 30aecd9f4..d426990bb 100644 --- a/tests/contrib/google_adk_agents/test_adk_streaming.py +++ b/tests/contrib/google_adk_agents/test_adk_streaming.py @@ -12,7 +12,12 @@ import pytest from google.adk import Agent -from google.adk.agents.run_config import RunConfig, StreamingMode + +# ADK 2.7 marks StreamingMode as a private re-export. +from google.adk.agents.run_config import ( + RunConfig, + StreamingMode, # pyright: ignore[reportPrivateImportUsage] +) from google.adk.models import BaseLlm, LLMRegistry from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse diff --git a/tests/contrib/google_genai/test_gemini.py b/tests/contrib/google_genai/test_gemini.py index 0881f63b3..d87d37675 100644 --- a/tests/contrib/google_genai/test_gemini.py +++ b/tests/contrib/google_genai/test_gemini.py @@ -292,8 +292,8 @@ async def gemini_file_search_stores_upload( self, req: _GeminiUploadToFileSearchStoreRequest ) -> types.UploadToFileSearchStoreOperation: self.file_search_store_upload_requests.append(req) - return types.UploadToFileSearchStoreOperation.model_construct( - name="operations/test-op", + return types.UploadToFileSearchStoreOperation.model_validate( + {"name": "operations/test-op"} ) @activity.defn @@ -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.""" @@ -1251,6 +1261,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 # =========================================================================== @@ -1387,8 +1415,8 @@ async def _gen(): ) gemini.aio.files.download = AsyncMock(return_value=b"mock download content") # type: ignore[method-assign] gemini.aio.file_search_stores.upload_to_file_search_store = AsyncMock( # type: ignore[method-assign] - return_value=types.UploadToFileSearchStoreOperation.model_construct( - name="operations/mock-op" + return_value=types.UploadToFileSearchStoreOperation.model_validate( + {"name": "operations/mock-op"} ) ) diff --git a/tests/contrib/openai_agents/test_openai.py b/tests/contrib/openai_agents/test_openai.py index 25597ee55..0baae0e6f 100644 --- a/tests/contrib/openai_agents/test_openai.py +++ b/tests/contrib/openai_agents/test_openai.py @@ -1431,8 +1431,16 @@ def status_error(status: int): import httpx raise APIStatusError( message="Something went wrong.", - response=httpx.Response( - status_code=status, request=httpx.Request("GET", url="") + # Under the latest dependency set openai carries its own httpx + # as `httpx2`, so the parameter's `Response` is a different + # class from the one built here and the two are not assignable. + # Only the annotation differs; openai reads the object the same + # way either way. + response=cast( + Any, + httpx.Response( + status_code=status, request=httpx.Request("GET", url="") + ), ), body=None, ) diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index ff6b36e41..dfa466500 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -14,6 +14,7 @@ import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2 import temporalio.converter import temporalio.nexus.system as nexus_system +import temporalio.nexus.system.workflow_service.models as workflow_service_models from temporalio import workflow from temporalio.bridge._visitor import PayloadVisitor from temporalio.bridge._visitor_functions import VisitorFunctions @@ -137,12 +138,12 @@ def _assert_start_nexus_operation_interceptor_trace() -> None: assert trace_name == "workflow.start_nexus_operation" trace_input = cast(StartNexusOperationInput[Any, Any], trace_value) request = cast( - workflowservice_pb2.SignalWithStartWorkflowExecutionRequest, + workflow_service_models.SignalWithStartWorkflowRequest, trace_input.input, ) - assert request.workflow_id == "system-nexus-workflow-id" - assert request.signal_name == "test-signal" - assert request.workflow_type.name == "test-workflow" + assert request.id == "system-nexus-workflow-id" + assert request.signal == "test-signal" + assert request.workflow == "test-workflow" class _MarkingPayloadVisitor(VisitorFunctions): @@ -343,13 +344,10 @@ def _proto_scalar_sample(field: FieldDescriptor, *, path: str) -> Any: def _field_is_repeated(field: FieldDescriptor) -> bool: - return bool( - getattr( - field, - "is_repeated", - getattr(field, "label") == FieldDescriptor.LABEL_REPEATED, - ) - ) + is_repeated = getattr(field, "is_repeated", None) + if is_repeated is not None: + return bool(is_repeated) + return getattr(field, "label") == FieldDescriptor.LABEL_REPEATED @pytest.mark.parametrize( diff --git a/uv.lock b/uv.lock index 9c625ed14..df077d1e8 100644 --- a/uv.lock +++ b/uv.lock @@ -1354,7 +1354,7 @@ requests = [ [[package]] name = "google-genai" -version = "2.11.0" +version = "2.21.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, @@ -1368,9 +1368,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a6/01/e7b5f3aac89200c78318ed7643401e7f5ed3131b0cd353c07483606b1e61/google_genai-2.11.0.tar.gz", hash = "sha256:4c5e524d24b145c96be327f9a7f8f04b0fe4efee0533877795e9848afed01749", size = 622366, upload-time = "2026-07-09T17:49:43.862Z" } +sdist = { url = "https://files.pythonhosted.org/packages/60/a7/a45f64f22ab9302b55fcbeb32acb6f313690a7748629b01e451aad1817a3/google_genai-2.21.0.tar.gz", hash = "sha256:0ecc11c6a5b9f5e3cc58e77ae5fead00c6719f8a1b2b654b803f514a9a6b64c0", size = 677301, upload-time = "2026-08-31T21:49:14.508Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/93/ef/d296c23390160a8b0b1dafb36dd3cb36a39ed40c81cd27e04e6233334186/google_genai-2.11.0-py3-none-any.whl", hash = "sha256:5bc8186100e1d34d691fbe0cba392b7e04e98d286ca952323a6672d054accf95", size = 984162, upload-time = "2026-07-09T17:49:42.15Z" }, + { url = "https://files.pythonhosted.org/packages/fb/d7/c2419dd5fedd5803ce09810e4e39561a0a13a960b3f48d2581c12e42af3e/google_genai-2.21.0-py3-none-any.whl", hash = "sha256:36b575034be46a03acd603a852e22a6359f2cdd6b26bb1d65d9b7e0cc7ab3648", size = 1080223, upload-time = "2026-08-31T21:49:12.699Z" }, ] [[package]] @@ -4820,7 +4820,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" },