From eee86db9e8596c543015470bf4e1bf202ad563a5 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 26 Aug 2026 16:24:33 -0700 Subject: [PATCH 1/6] feat: add System Nexus operation interceptors --- CHANGELOG.md | 4 + scripts/gen_nexus_system_api.py | 13 +- .../contrib/opentelemetry/_interceptor.py | 36 ++++ .../opentelemetry/_otel_interceptor.py | 13 ++ .../wit/deps/nexus-temporal-types/model.wit | 192 ++++++++++++++++++ .../wit/temporal-proto-models-nexusrpc.yaml | 19 ++ .../nexus/system/wit/workflow-service.wit | 130 ++++++++++++ .../_system_nexus_interceptor.py | 96 +++++++++ temporalio/worker/_interceptor.py | 12 +- temporalio/worker/_workflow_instance.py | 31 ++- .../opentelemetry/test_opentelemetry.py | 63 ++++++ .../test_opentelemetry_plugin.py | 63 ++++++ tests/nexus/test_temporal_system_nexus.py | 26 ++- 13 files changed, 673 insertions(+), 25 deletions(-) create mode 100644 temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit create mode 100644 temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml create mode 100644 temporalio/nexus/system/wit/workflow-service.wit create mode 100644 temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py diff --git a/CHANGELOG.md b/CHANGELOG.md index be3444177..a3ca40991 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,10 @@ to include examples, links to docs, or any other relevant information. ### Changed +- 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. + ### Deprecated ### :boom: Breaking Changes diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index 82f40a8f0..9e2848c82 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -11,17 +11,7 @@ base_dir = Path(__file__).parent.parent sys.path.insert(0, str(base_dir)) -wit_input_dir = ( - base_dir - / "temporalio" - / "bridge" - / "sdk-core" - / "crates" - / "protos" - / "protos" - / "api_upstream" - / "nexus" -) +wit_input_dir = base_dir / "temporalio" / "nexus" / "system" / "wit" wit_path = wit_input_dir / "workflow-service.wit" wit_deps_dir = wit_input_dir / "deps" python_support_path = base_dir / "scripts" / "nex_gen_support.py" @@ -131,6 +121,7 @@ def generate_nexus_system_api() -> None: str(wit_path), str(wit_deps_dir), "--native-api", + "--system-nexus", "--support-file", str(python_support_path), "--descriptors", diff --git a/temporalio/contrib/opentelemetry/_interceptor.py b/temporalio/contrib/opentelemetry/_interceptor.py index eb22f8be6..6dca4596e 100644 --- a/temporalio/contrib/opentelemetry/_interceptor.py +++ b/temporalio/contrib/opentelemetry/_interceptor.py @@ -33,6 +33,7 @@ import temporalio.client import temporalio.converter import temporalio.exceptions +import temporalio.nexus.system.workflow_service.models import temporalio.worker import temporalio.workflow from temporalio.exceptions import ApplicationError, ApplicationErrorCategory @@ -433,6 +434,10 @@ class _InputWithStringHeaders(Protocol): headers: Mapping[str, str] | None +class _InputWithModelHeaders(Protocol): + headers: Mapping[str, Any] | None + + class _InputWithOperationContext(Generic[_ContextT], Protocol): ctx: _ContextT @@ -684,6 +689,18 @@ def _context_carrier_to_headers( } return headers + def _context_carrier_to_model_headers( + self, + carrier: _CarrierDict, + headers: Mapping[str, Any] | None, + ) -> Mapping[str, Any]: + if carrier: + return { + **(headers or {}), + self.header_key: carrier, + } + return headers or {} + def _completed_span( self, span_name: str, @@ -691,6 +708,7 @@ def _completed_span( link_context_carrier: _CarrierDict | None = None, add_to_outbound: _InputWithHeaders | None = None, add_to_outbound_str: _InputWithStringHeaders | None = None, + add_to_outbound_model: _InputWithModelHeaders | None = None, new_span_even_on_replay: bool = False, additional_attributes: opentelemetry.util.types.Attributes = None, exception: Exception | None = None, @@ -742,6 +760,11 @@ def _completed_span( updated_context_carrier, add_to_outbound_str.headers ) + if add_to_outbound_model: + add_to_outbound_model.headers = self._context_carrier_to_model_headers( + updated_context_carrier, add_to_outbound_model.headers + ) + def _set_on_context( self, context: opentelemetry.context.Context ) -> opentelemetry.context.Context: @@ -830,6 +853,19 @@ async def start_nexus_operation( return await super().start_nexus_operation(input) + async def start_signal_with_start_workflow( + self, + request: temporalio.nexus.system.workflow_service.models.SignalWithStartWorkflowRequest, + ) -> temporalio.workflow.NexusOperationHandle[ + temporalio.nexus.system.workflow_service.models.SignalWithStartWorkflowResponse + ]: + self.root._completed_span( + "SignalWithStartWorkflow", + kind=opentelemetry.trace.SpanKind.CLIENT, + add_to_outbound_model=request, + ) + return await super().start_signal_with_start_workflow(request) + def _carrier_to_nexus_headers( carrier: _CarrierDict, initial: Mapping[str, str] | None = None diff --git a/temporalio/contrib/opentelemetry/_otel_interceptor.py b/temporalio/contrib/opentelemetry/_otel_interceptor.py index c120fcd03..ff07f0f1d 100644 --- a/temporalio/contrib/opentelemetry/_otel_interceptor.py +++ b/temporalio/contrib/opentelemetry/_otel_interceptor.py @@ -32,6 +32,7 @@ import temporalio.api.common.v1 import temporalio.client import temporalio.converter +import temporalio.nexus.system.workflow_service.models import temporalio.worker import temporalio.workflow from temporalio.contrib.opentelemetry._tracer_provider import ( @@ -600,3 +601,15 @@ async def start_nexus_operation( ): input.headers = _context_to_nexus_headers(input.headers or {}) return await super().start_nexus_operation(input) + + async def start_signal_with_start_workflow( + self, + request: temporalio.nexus.system.workflow_service.models.SignalWithStartWorkflowRequest, + ) -> temporalio.workflow.NexusOperationHandle[ + temporalio.nexus.system.workflow_service.models.SignalWithStartWorkflowResponse + ]: + with self._workflow_maybe_span( + "SignalWithStartWorkflow", kind=opentelemetry.trace.SpanKind.CLIENT + ): + request.headers = _context_to_headers(request.headers or {}) + return await super().start_signal_with_start_workflow(request) diff --git a/temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit b/temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit new file mode 100644 index 000000000..9909d34c6 --- /dev/null +++ b/temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit @@ -0,0 +1,192 @@ +package nexus:temporal-types@1.0.0; + +interface model { + /// String-shaped placeholder for semantic types that generators reinterpret. + type placeholder = string; + + /// @nexus.proto "temporal.api.common.v1.Payload" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="typing.Any" + /// typescript="common.Payload" + /// dotnet="object?" + /// dotnet-from="ProtoExtensions.FromPayload" + /// dotnet-to="ProtoExtensions.ToPayload" + /// typescript-import="@temporalio/common" + type payload = placeholder; + + /// @nexus.proto "temporal.api.common.v1.Payloads" + /// typescript-import="@temporalio/proto" + /// @nexus.type dotnet="IReadOnlyCollection" dotnet-from="ProtoExtensions.FromPayloads" dotnet-to="ProtoExtensions.ToPayloads" + type payloads = list; + + /// Temporal failure represented by the target SDK's native exception/error type. + /// The SDK failure converter owns the recursive cause and failure-info structure. + /// @nexus.proto "temporal.api.failure.v1.Failure" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="BaseException" + /// typescript="Error" + /// go="error" + /// dotnet="System.Exception" + /// dotnet-from="ProtoExtensions.FromFailureProto" + /// dotnet-to="ProtoExtensions.ToFailureProto" + type failure = placeholder; + + /// Callable result annotation for workflow functions. + /// @nexus.type + /// python="collections.abc.Awaitable[WorkflowResult]" + /// typescript="Promise" + /// dotnet="System.Threading.Tasks.Task" + type workflow-result = placeholder; + + /// Receiver/context argument for workflow callable method forms. + /// @nexus.type python="typing.Any" typescript="any" dotnet="object" + type callable-prefix = placeholder; + + /// @nexus.function-args + /// varargs=true + /// param="args" + /// typescript-drop-prefix=true + workflow-call: async func(callable-prefix: callable-prefix, args: payloads) -> workflow-result; + + /// Callable result annotation for signal functions. + /// @nexus.type python="None | collections.abc.Awaitable[None]" typescript="void" dotnet="void" + type signal-result = placeholder; + + /// @nexus.function-args + /// varargs=true + /// param="signal-args" + /// typescript-drop-prefix=true + signal-call: func(callable-prefix: callable-prefix, signal-args: payloads) -> signal-result; + + /// @nexus.proto "temporal.api.common.v1.WorkflowType" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="str" + /// typescript="string" + /// dotnet="string" + /// dotnet-from="ProtoExtensions.FromWorkflowTypeProto" + /// dotnet-to="ProtoExtensions.ToWorkflowTypeProto" + type workflow-type = placeholder; + + /// @nexus.function + /// primary=true + /// signature="workflow-call" + /// args-field="input" + /// result-type-parameter="WorkflowResult" + /// alternate-type="workflow-type" + /// dotnet-name-extractor="TemporalFunctionNames.WorkflowName" + /// dotnet-call-extractor="TemporalFunctionNames.ExtractCall" + /// @nexus.add-rpc-compatible-with "workflow-type" + type workflow-function = placeholder; + + /// @nexus.function + /// signature="signal-call" + /// args-field="signal-input" + /// alternate-type="string" + /// python-converter="signal_function_to_proto" + /// typescript-name-extractor="signalFunctionName" + /// dotnet-name-extractor="TemporalFunctionNames.SignalName" + /// dotnet-call-extractor="TemporalFunctionNames.ExtractCall" + /// typescript-value-type="workflow.SignalDefinition" + /// typescript-args-type="Value extends workflow.SignalDefinition ? Args : never" + /// typescript-import="@temporalio/workflow" + /// @nexus.add-rpc-compatible-with "string" + type signal-function = placeholder; + + /// @nexus.proto "temporal.api.common.v1.RetryPolicy" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="temporalio.common.RetryPolicy" + /// typescript="common.RetryPolicy" + /// dotnet="Temporalio.Common.RetryPolicy" + /// dotnet-from="ProtoExtensions.FromRetryPolicyProto" + /// typescript-import="@temporalio/common" + type retry-policy = placeholder; + + /// @nexus.proto "temporal.api.taskqueue.v1.TaskQueue" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="str" + /// typescript="string" + /// dotnet="string" + /// dotnet-from="ProtoExtensions.FromTaskQueueProto" + /// dotnet-to="ProtoExtensions.ToTaskQueueProto" + type task-queue = placeholder; + + /// @nexus.proto "temporal.api.common.v1.Memo" typescript-import="@temporalio/proto" + /// @nexus.type python="collections.abc.Mapping[str, typing.Any]" typescript="Record" dotnet="IReadOnlyDictionary" dotnet-from="ProtoExtensions.FromMemoProto" + type memo = placeholder; + + /// @nexus.proto "temporal.api.common.v1.Header" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="collections.abc.Mapping[str, typing.Any]" + /// typescript="common.Headers" + /// go="map[string]any" + /// dotnet="IReadOnlyDictionary" + /// dotnet-from="ProtoExtensions.FromHeaderProto" + /// dotnet-to="ProtoExtensions.ToHeaderProto" + /// typescript-import="@temporalio/common" + type header = placeholder; + + /// @nexus.proto "temporal.api.common.v1.SearchAttributes" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="temporalio.common.TypedSearchAttributes" + /// typescript="common.TypedSearchAttributes" + /// dotnet="Temporalio.Common.SearchAttributeCollection" + /// dotnet-from="ProtoExtensions.FromSearchAttributesProto" + /// typescript-import="@temporalio/common" + type search-attributes = placeholder; + + /// @nexus.proto "temporal.api.common.v1.Priority" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="temporalio.common.Priority" + /// typescript="common.Priority" + /// dotnet="Temporalio.Common.Priority" + /// dotnet-from="ProtoExtensions.FromPriorityProto" + /// typescript-import="@temporalio/common" + type priority = placeholder; + + /// @nexus.proto "temporal.api.workflow.v1.VersioningOverride" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="temporalio.common.VersioningOverride" + /// typescript="common.VersioningOverride" + /// dotnet="Temporalio.Common.VersioningOverride" + /// dotnet-from="ProtoExtensions.FromVersioningOverrideProto" + /// typescript-import="@temporalio/common" + type versioning-override = placeholder; + + /// @nexus.proto "google.protobuf.Duration" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="datetime.timedelta" + /// typescript="common.Duration" + /// dotnet="System.TimeSpan" + /// dotnet-from="ProtoExtensions.FromDurationProto" + /// typescript-import="@temporalio/common" + type duration = placeholder; + + /// @nexus.proto "temporal.api.enums.v1.WorkflowIdReusePolicy" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="temporalio.common.WorkflowIDReusePolicy" + /// typescript="common.WorkflowIdReusePolicy" + /// dotnet="Temporalio.Api.Enums.V1.WorkflowIdReusePolicy" + /// typescript-import="@temporalio/common" + type workflow-id-reuse-policy = placeholder; + + /// @nexus.proto "temporal.api.enums.v1.WorkflowIdConflictPolicy" typescript-import="@temporalio/proto" + /// @nexus.type + /// python="temporalio.common.WorkflowIDConflictPolicy" + /// typescript="common.WorkflowIdConflictPolicy" + /// dotnet="Temporalio.Api.Enums.V1.WorkflowIdConflictPolicy" + /// typescript-import="@temporalio/common" + type workflow-id-conflict-policy = placeholder; + + /// @nexus.proto "temporal.api.sdk.v1.UserMetadata" typescript-import="@temporalio/proto" + /// @nexus.flatten-in-api + record user-metadata { + /// @nexus.doc "Single-line fixed summary for the workflow execution that may appear in UI and CLI. This can be in single-line Temporal Markdown format." + /// @nexus.proto-field "summary" + /// @nexus.flattened-type python="str" typescript="string" dotnet="string" + static-summary: option, + /// @nexus.doc "General fixed details for the workflow execution that may appear in UI and CLI. This can be in Temporal Markdown format and can span multiple lines. This value is fixed on the workflow execution and cannot be updated." + /// @nexus.proto-field "details" + /// @nexus.flattened-type python="str" typescript="string" dotnet="string" + static-details: option, + } +} diff --git a/temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml b/temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml new file mode 100644 index 000000000..ebc74b68b --- /dev/null +++ b/temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml @@ -0,0 +1,19 @@ +nexusrpc: 1.0.0 +services: + temporal.api.workflowservice.v1.WorkflowService: + operations: + SignalWithStartWorkflowExecution: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.SignalWithStartWorkflowExecutionRequest + $goRef: go.temporal.io/api/workflowservice/v1.SignalWithStartWorkflowExecutionRequest + $javaRef: io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest + $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.SignalWithStartWorkflowExecutionResponse + $goRef: go.temporal.io/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse + $javaRef: io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse + $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse' diff --git a/temporalio/nexus/system/wit/workflow-service.wit b/temporalio/nexus/system/wit/workflow-service.wit new file mode 100644 index 000000000..190aae88c --- /dev/null +++ b/temporalio/nexus/system/wit/workflow-service.wit @@ -0,0 +1,130 @@ +package temporal:nexus@1.0.0; + +world system { + export workflow-service; +} + +/// @nexus.endpoint "__temporal_system" +/// @nexus.service-name "temporal.api.workflowservice.v1.WorkflowService" +/// @nexus.namespace dotnet="Temporalio.Workflows" +/// @nexus.operations-class dotnet="Workflow" +/// @nexus.delay-load-temporalio-workflow +/// @nexus.experimental +interface workflow-service { + use nexus:temporal-types/model@1.0.0.{ + duration, + header, + memo, + payloads, + placeholder, + priority, + retry-policy, + search-attributes, + signal-function, + task-queue, + user-metadata, + versioning-override, + workflow-function, + workflow-id-conflict-policy, + workflow-id-reuse-policy, + }; + + /// @nexus.doc "Request fields for signaling a workflow, starting it first if needed." + /// @nexus.experimental + /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest" typescript-import="@temporalio/proto" + record signal-with-start-workflow-request { + /// @nexus.doc + /// python="Workflow type name or callable identifying the workflow to start." + /// typescript="Workflow type name or workflow function identifying the workflow to start." + /// dotnet="Workflow type name or workflow expression identifying the workflow to start." + /// @nexus.proto-field "workflow_type" + workflow: workflow-function, + /// @nexus.doc "Unique identifier for the workflow execution." + /// @nexus.proto-field "workflow_id" + id: string, + /// @nexus.doc "Task queue to run the workflow on." + task-queue: task-queue, + /// @nexus.doc + /// python="Signal name or callable to send with the start request." + /// typescript="Signal name or signal definition to send with the start request." + /// dotnet="Signal name or signal expression to send with the start request." + /// @nexus.proto-field "signal_name" + signal: signal-function, + /// @nexus.doc "Total workflow execution timeout, including retries and continue-as-new." + /// @nexus.proto-field "workflow_execution_timeout" + execution-timeout: option, + /// @nexus.doc "Timeout of a single workflow run." + /// @nexus.proto-field "workflow_run_timeout" + run-timeout: option, + /// @nexus.doc "Timeout of a single workflow task." + /// @nexus.proto-field "workflow_task_timeout" + task-timeout: option, + /// @nexus.omit + identity: placeholder, + /// @nexus.omit + request-id: placeholder, + /// @nexus.doc "Behavior when a closed workflow with the same ID exists. Default is allow-duplicate." + /// @nexus.proto-field "workflow_id_reuse_policy" + /// @nexus.default "allow-duplicate" + id-reuse-policy: workflow-id-reuse-policy, + /// @nexus.doc "Behavior when a workflow is currently running with the same ID. Set to use-existing for idempotent deduplication on workflow ID. Cannot be set if id-reuse-policy is terminate-if-running." + /// @nexus.proto-field "workflow_id_conflict_policy" + id-conflict-policy: option, + /// @nexus.doc "Retry policy for the workflow." + retry-policy: option, + /// @nexus.doc "Cron schedule for recurring workflow executions. See https://docs.temporal.io/cron-job." + cron-schedule: option, + /// @nexus.doc "Memo for the workflow." + memo: option, + /// @nexus.doc "Typed search attributes for the workflow." + search-attributes: option, + /// @nexus.doc "Priority of the workflow execution." + priority: option, + /// @nexus.doc "Override for workflow versioning behavior." + versioning-override: option, + /// @nexus.doc "Amount of time to wait before starting the workflow. This does not work with cron-schedule." + /// @nexus.proto-field "workflow_start_delay" + start-delay: option, + user-metadata: option, + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + namespace: string, + /// @nexus.omit + control: placeholder, + /// @nexus.api-omit + /// @nexus.proto-field "header" + headers: option
, + /// @nexus.omit + links: placeholder, + /// @nexus.omit + time-skipping-config: placeholder, + } + + /// @nexus.experimental + /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse" typescript-import="@temporalio/proto" + record signal-with-start-workflow-response { + run-id: option, + started: option, + /// @nexus.omit + signal-link: placeholder, + /// @nexus.omit + first-execution-run-id: placeholder, + } + + /// @nexus.doc + /// "Signal a workflow, starting it first if needed." + /// returns="A workflow handle to the started workflow." + /// @nexus.output-transform + /// python-type="temporalio.workflow.ExternalWorkflowHandle[WorkflowResult]" + /// python="temporalio.workflow.get_external_workflow_handle(request.id, run_id=result.run_id)" + /// typescript-type="workflow.ExternalWorkflowHandle" + /// typescript="workflow.getExternalWorkflowHandle(request.id, result.runId ?? undefined)" + /// typescript-import="@temporalio/workflow" + /// dotnet-type="Temporalio.Workflows.ExternalWorkflowHandle" + /// dotnet="Temporalio.Workflows.Workflow.GetExternalWorkflowHandle(request.Id, result.RunId)" + /// @nexus.operation name="SignalWithStartWorkflowExecution" + /// @nexus.serialization-context python="signal_with_start_workflow_serialization_context" dotnet="WorkflowServiceSerializationContexts.SignalWithStartWorkflow" + /// @nexus.experimental + signal-with-start-workflow: func( + request: signal-with-start-workflow-request, + ) -> signal-with-start-workflow-response; +} 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..afe773813 --- /dev/null +++ b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py @@ -0,0 +1,96 @@ +# Generated by nexgen v0.2.2. DO NOT EDIT! + +from __future__ import annotations + +import abc +import typing + +from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + +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 System Nexus temporal.api.workflowservice.v1.WorkflowService/SignalWithStartWorkflowExecution operation.""" + return await self._next_system_nexus_interceptor().start_signal_with_start_workflow( + request + ) + + +class _SystemNexusWorkflowOutboundInterceptorTerminal(abc.ABC): + @abc.abstractmethod + async def _outbound_start_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.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._outbound_start_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/worker/_interceptor.py b/temporalio/worker/_interceptor.py index 4acf3c5d1..1bac84673 100644 --- a/temporalio/worker/_interceptor.py +++ b/temporalio/worker/_interceptor.py @@ -21,6 +21,9 @@ import temporalio.nexus import temporalio.nexus._util import temporalio.workflow +from temporalio.nexus.system.workflow_service._system_nexus_interceptor import ( + _SystemNexusWorkflowOutboundInterceptorBase, +) from temporalio.workflow import ContinueAsNewVersioningBehavior, VersioningIntent @@ -414,12 +417,14 @@ async def handle_update_handler(self, input: HandleUpdateInput) -> Any: return await self.next.handle_update_handler(input) -class WorkflowOutboundInterceptor: +class WorkflowOutboundInterceptor(_SystemNexusWorkflowOutboundInterceptorBase): """Outbound interceptor to wrap calls made from within workflows. This should be extended by any workflow outbound interceptors. """ + next: WorkflowOutboundInterceptor + def __init__(self, next: WorkflowOutboundInterceptor) -> None: """Create the outbound interceptor. @@ -429,6 +434,11 @@ def __init__(self, next: WorkflowOutboundInterceptor) -> None: """ self.next = next + def _next_system_nexus_interceptor( + self, + ) -> _SystemNexusWorkflowOutboundInterceptorBase: + return self.next + def continue_as_new(self, input: ContinueAsNewInput) -> NoReturn: """Called for every :py:func:`temporalio.workflow.continue_as_new` call.""" self.next.continue_as_new(input) diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 69b416f26..fca73c266 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -60,6 +60,10 @@ import temporalio.nexus.system import temporalio.workflow from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo +from temporalio.nexus.system.workflow_service._system_nexus_interceptor import ( + _start_system_nexus_operation, + _SystemNexusWorkflowOutboundInterceptorTerminal, +) from temporalio.service import __version__ from ..api.failure.v1.message_pb2 import Failure @@ -1732,7 +1736,23 @@ async def workflow_start_nexus_operation( headers: Mapping[str, str] | None, summary: str | None, ) -> temporalio.workflow.NexusOperationHandle[OutputT]: - # start_nexus_operation + if temporalio.nexus.system.is_system_endpoint(endpoint): + return await _start_system_nexus_operation( + self._outbound, + StartNexusOperationInput( + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + service=service, + operation=operation, + input=input, + output_type=output_type, + schedule_to_close_timeout=schedule_to_close_timeout, + schedule_to_start_timeout=schedule_to_start_timeout, + start_to_close_timeout=start_to_close_timeout, + cancellation_type=cancellation_type, + headers=None, + summary=summary, + ), + ) return await self._outbound.start_nexus_operation( StartNexusOperationInput( endpoint=endpoint, @@ -3131,11 +3151,18 @@ async def handle_update_handler(self, input: HandleUpdateInput) -> Any: return handler(*input.args) -class _WorkflowOutboundImpl(WorkflowOutboundInterceptor): +class _WorkflowOutboundImpl( + _SystemNexusWorkflowOutboundInterceptorTerminal, WorkflowOutboundInterceptor +): def __init__(self, instance: _WorkflowInstanceImpl) -> None: # type: ignore # We are intentionally not calling the base class's __init__ here self._instance = instance + async def _outbound_start_nexus_operation( + self, input: StartNexusOperationInput[InputT, OutputT] + ) -> temporalio.workflow.NexusOperationHandle[OutputT]: + return await self._instance._outbound_start_nexus_operation(input) + def continue_as_new(self, input: ContinueAsNewInput) -> NoReturn: self._instance._outbound_continue_as_new(input) diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py index 1bab931ac..0ea9530e6 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry.py +++ b/tests/contrib/opentelemetry/test_opentelemetry.py @@ -88,6 +88,34 @@ class TracingWorkflowActionActivity: fail_on_non_replay_before_complete: bool = False +@workflow.defn +class SignalWithStartHeaderWorkflow: + def __init__(self) -> None: + self._signaled = False + + @workflow.run + async def run(self) -> bool: + await workflow.wait_condition(lambda: self._signaled) + return "_tracer-data" in workflow.info().headers + + @workflow.signal + def notify(self) -> None: + self._signaled = True + + +@workflow.defn +class SignalWithStartCallerWorkflow: + @workflow.run + async def run(self, target_id: str, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + SignalWithStartHeaderWorkflow.run, + id=target_id, + task_queue=task_queue, + signal=SignalWithStartHeaderWorkflow.notify, + ) + return handle.id + + @dataclass class TracingWorkflowActionContinueAsNew: param: TracingWorkflowParam @@ -229,6 +257,41 @@ def update_validator(self) -> None: pass +@pytest.mark.requires_local_server +async def test_workflow_signal_with_start_propagates_trace_headers( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = get_tracer(__name__, tracer_provider=provider) + config = client.config() + config["interceptors"] = [TracingInterceptor(tracer)] + client = Client(**config) + + async with Worker( + client, + task_queue=f"signal-with-start-{uuid.uuid4()}", + workflows=[SignalWithStartCallerWorkflow, SignalWithStartHeaderWorkflow], + workflow_runner=UnsandboxedWorkflowRunner(), + ) as worker: + target_id = f"signal-with-start-target-{uuid.uuid4()}" + with tracer.start_as_current_span("signal-with-start"): + caller = await client.start_workflow( + SignalWithStartCallerWorkflow.run, + args=[target_id, worker.task_queue], + id=f"signal-with-start-caller-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + assert await caller.result() == target_id + assert await client.get_workflow_handle(target_id).result() is True + assert any( + span.name == "SignalWithStartWorkflow" for span in exporter.get_finished_spans() + ) + + async def test_opentelemetry_tracing(client: Client, env: WorkflowEnvironment): # TODO(cretz): Fix if env.supports_time_skipping: diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 12d0c5972..022645fee 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -122,6 +122,34 @@ async def run(self): return +@workflow.defn +class SignalWithStartHeaderWorkflow: + def __init__(self) -> None: + self._signaled = False + + @workflow.run + async def run(self) -> bool: + await workflow.wait_condition(lambda: self._signaled) + return "_tracer-data" in workflow.info().headers + + @workflow.signal + def notify(self) -> None: + self._signaled = True + + +@workflow.defn +class SignalWithStartCallerWorkflow: + @workflow.run + async def run(self, target_id: str, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + SignalWithStartHeaderWorkflow.run, + id=target_id, + task_queue=task_queue, + signal=SignalWithStartHeaderWorkflow.notify, + ) + return handle.id + + async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: Any): # type: ignore[reportUnusedParameter] exporter = InMemorySpanExporter() provider = create_tracer_provider() @@ -169,6 +197,41 @@ async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: An ) +@pytest.mark.requires_local_server +async def test_workflow_signal_with_start_propagates_trace_headers( + client: Client, + env: WorkflowEnvironment, + reset_otel_tracer_provider: Any, # type: ignore[reportUnusedParameter] +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + exporter = InMemorySpanExporter() + provider = create_tracer_provider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + opentelemetry.trace.set_tracer_provider(provider) + config = client.config() + config["plugins"] = [OpenTelemetryPlugin(add_temporal_spans=True)] + client = Client(**config) + + async with new_worker( + client, SignalWithStartCallerWorkflow, SignalWithStartHeaderWorkflow + ) as worker: + target_id = f"signal-with-start-target-{uuid.uuid4()}" + with get_tracer(__name__).start_as_current_span("signal-with-start"): + caller = await client.start_workflow( + SignalWithStartCallerWorkflow.run, + args=[target_id, worker.task_queue], + id=f"signal-with-start-caller-{uuid.uuid4()}", + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=3), + ) + assert await caller.result() == target_id + assert await client.get_workflow_handle(target_id).result() is True + assert any( + span.name == "SignalWithStartWorkflow" for span in exporter.get_finished_spans() + ) + + @workflow.defn class ComprehensiveWorkflow: def __init__(self) -> None: diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 6a9bc9959..69bed1a9c 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -36,7 +36,6 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import ( Interceptor, - StartNexusOperationInput, Worker, WorkflowInboundInterceptor, WorkflowInterceptorClassInput, @@ -251,11 +250,16 @@ def init(self, outbound: WorkflowOutboundInterceptor) -> None: class _TracingWorkflowOutboundInterceptor(WorkflowOutboundInterceptor): - async def start_nexus_operation( - self, input: StartNexusOperationInput[Any, Any] - ) -> workflow.NexusOperationHandle[Any]: - interceptor_traces.append(("workflow.start_nexus_operation", input)) - return await super().start_nexus_operation(input) + async def start_signal_with_start_workflow( + self, request: workflow_service_models.SignalWithStartWorkflowRequest + ) -> workflow.NexusOperationHandle[ + workflow_service_models.SignalWithStartWorkflowResponse + ]: + request.headers = {**(request.headers or {}), "interceptor-header": "value"} + interceptor_traces.append( + ("workflow.start_signal_with_start_workflow", request) + ) + return await super().start_signal_with_start_workflow(request) def _assert_stored_payloads_include( @@ -270,15 +274,15 @@ def _assert_stored_payloads_include( assert expected_payload_data.issubset(stored_payload_data) -def _assert_start_nexus_operation_interceptor_trace() -> None: +def _assert_signal_with_start_workflow_interceptor_trace() -> None: assert len(interceptor_traces) == 1 trace_name, trace_value = interceptor_traces.pop() - assert trace_name == "workflow.start_nexus_operation" - trace_input = cast(StartNexusOperationInput[Any, Any], trace_value) - request = trace_input.input + assert trace_name == "workflow.start_signal_with_start_workflow" + request = cast(workflow_service_models.SignalWithStartWorkflowRequest, trace_value) assert request.id == "system-nexus-workflow-id" assert request.signal == "test-signal" assert request.workflow == "test-workflow" + assert request.headers == {"interceptor-header": "value"} class _MarkingPayloadVisitor(VisitorFunctions): @@ -711,7 +715,7 @@ async def test_external_workflow_handle_signal_with_start_workflow_uses_system_n b'"details-value"', }, ) - _assert_start_nexus_operation_interceptor_trace() + _assert_signal_with_start_workflow_interceptor_trace() # Cloud namespaces created by CI do not have the System Nexus dynamic config. From 29c022f5cd98046753ac73fe32b30e7e29177805 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Fri, 28 Aug 2026 11:04:49 -0700 Subject: [PATCH 2/6] chore: regenerate System Nexus interceptor --- .../nexus/system/workflow_service/_system_nexus_interceptor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py index afe773813..8f21b9bdb 100644 --- a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py +++ b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py @@ -58,7 +58,7 @@ async def start_signal_with_start_workflow( ) -> temporalio.workflow.NexusOperationHandle[ models.SignalWithStartWorkflowResponse ]: - """Intercept the System Nexus temporal.api.workflowservice.v1.WorkflowService/SignalWithStartWorkflowExecution operation.""" + """Intercept the SignalWithStartWorkflow operation.""" return await self._next_system_nexus_interceptor().start_signal_with_start_workflow( request ) From 4cfcfee423ba439c67f6afc6807b3dc4bc03c6b9 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Fri, 28 Aug 2026 11:43:23 -0700 Subject: [PATCH 3/6] chore: use SDK Core WIT source --- scripts/gen_nexus_system_api.py | 12 +- temporalio/bridge/sdk-core | 2 +- .../wit/deps/nexus-temporal-types/model.wit | 192 ------------------ .../wit/temporal-proto-models-nexusrpc.yaml | 19 -- .../nexus/system/wit/workflow-service.wit | 130 ------------ 5 files changed, 12 insertions(+), 343 deletions(-) delete mode 100644 temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit delete mode 100644 temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml delete mode 100644 temporalio/nexus/system/wit/workflow-service.wit diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index 9e2848c82..0c410f009 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -11,7 +11,17 @@ base_dir = Path(__file__).parent.parent sys.path.insert(0, str(base_dir)) -wit_input_dir = base_dir / "temporalio" / "nexus" / "system" / "wit" +wit_input_dir = ( + base_dir + / "temporalio" + / "bridge" + / "sdk-core" + / "crates" + / "protos" + / "protos" + / "api_upstream" + / "nexus" +) wit_path = wit_input_dir / "workflow-service.wit" wit_deps_dir = wit_input_dir / "deps" python_support_path = base_dir / "scripts" / "nex_gen_support.py" diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index b860e3f14..85b71d7ec 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit b860e3f14f68af64bd4d0e372b3c78b2ce9ecff1 +Subproject commit 85b71d7ecd4f2bf677fa1cee17f3fbc1ab10f1b9 diff --git a/temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit b/temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit deleted file mode 100644 index 9909d34c6..000000000 --- a/temporalio/nexus/system/wit/deps/nexus-temporal-types/model.wit +++ /dev/null @@ -1,192 +0,0 @@ -package nexus:temporal-types@1.0.0; - -interface model { - /// String-shaped placeholder for semantic types that generators reinterpret. - type placeholder = string; - - /// @nexus.proto "temporal.api.common.v1.Payload" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="typing.Any" - /// typescript="common.Payload" - /// dotnet="object?" - /// dotnet-from="ProtoExtensions.FromPayload" - /// dotnet-to="ProtoExtensions.ToPayload" - /// typescript-import="@temporalio/common" - type payload = placeholder; - - /// @nexus.proto "temporal.api.common.v1.Payloads" - /// typescript-import="@temporalio/proto" - /// @nexus.type dotnet="IReadOnlyCollection" dotnet-from="ProtoExtensions.FromPayloads" dotnet-to="ProtoExtensions.ToPayloads" - type payloads = list; - - /// Temporal failure represented by the target SDK's native exception/error type. - /// The SDK failure converter owns the recursive cause and failure-info structure. - /// @nexus.proto "temporal.api.failure.v1.Failure" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="BaseException" - /// typescript="Error" - /// go="error" - /// dotnet="System.Exception" - /// dotnet-from="ProtoExtensions.FromFailureProto" - /// dotnet-to="ProtoExtensions.ToFailureProto" - type failure = placeholder; - - /// Callable result annotation for workflow functions. - /// @nexus.type - /// python="collections.abc.Awaitable[WorkflowResult]" - /// typescript="Promise" - /// dotnet="System.Threading.Tasks.Task" - type workflow-result = placeholder; - - /// Receiver/context argument for workflow callable method forms. - /// @nexus.type python="typing.Any" typescript="any" dotnet="object" - type callable-prefix = placeholder; - - /// @nexus.function-args - /// varargs=true - /// param="args" - /// typescript-drop-prefix=true - workflow-call: async func(callable-prefix: callable-prefix, args: payloads) -> workflow-result; - - /// Callable result annotation for signal functions. - /// @nexus.type python="None | collections.abc.Awaitable[None]" typescript="void" dotnet="void" - type signal-result = placeholder; - - /// @nexus.function-args - /// varargs=true - /// param="signal-args" - /// typescript-drop-prefix=true - signal-call: func(callable-prefix: callable-prefix, signal-args: payloads) -> signal-result; - - /// @nexus.proto "temporal.api.common.v1.WorkflowType" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="str" - /// typescript="string" - /// dotnet="string" - /// dotnet-from="ProtoExtensions.FromWorkflowTypeProto" - /// dotnet-to="ProtoExtensions.ToWorkflowTypeProto" - type workflow-type = placeholder; - - /// @nexus.function - /// primary=true - /// signature="workflow-call" - /// args-field="input" - /// result-type-parameter="WorkflowResult" - /// alternate-type="workflow-type" - /// dotnet-name-extractor="TemporalFunctionNames.WorkflowName" - /// dotnet-call-extractor="TemporalFunctionNames.ExtractCall" - /// @nexus.add-rpc-compatible-with "workflow-type" - type workflow-function = placeholder; - - /// @nexus.function - /// signature="signal-call" - /// args-field="signal-input" - /// alternate-type="string" - /// python-converter="signal_function_to_proto" - /// typescript-name-extractor="signalFunctionName" - /// dotnet-name-extractor="TemporalFunctionNames.SignalName" - /// dotnet-call-extractor="TemporalFunctionNames.ExtractCall" - /// typescript-value-type="workflow.SignalDefinition" - /// typescript-args-type="Value extends workflow.SignalDefinition ? Args : never" - /// typescript-import="@temporalio/workflow" - /// @nexus.add-rpc-compatible-with "string" - type signal-function = placeholder; - - /// @nexus.proto "temporal.api.common.v1.RetryPolicy" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="temporalio.common.RetryPolicy" - /// typescript="common.RetryPolicy" - /// dotnet="Temporalio.Common.RetryPolicy" - /// dotnet-from="ProtoExtensions.FromRetryPolicyProto" - /// typescript-import="@temporalio/common" - type retry-policy = placeholder; - - /// @nexus.proto "temporal.api.taskqueue.v1.TaskQueue" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="str" - /// typescript="string" - /// dotnet="string" - /// dotnet-from="ProtoExtensions.FromTaskQueueProto" - /// dotnet-to="ProtoExtensions.ToTaskQueueProto" - type task-queue = placeholder; - - /// @nexus.proto "temporal.api.common.v1.Memo" typescript-import="@temporalio/proto" - /// @nexus.type python="collections.abc.Mapping[str, typing.Any]" typescript="Record" dotnet="IReadOnlyDictionary" dotnet-from="ProtoExtensions.FromMemoProto" - type memo = placeholder; - - /// @nexus.proto "temporal.api.common.v1.Header" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="collections.abc.Mapping[str, typing.Any]" - /// typescript="common.Headers" - /// go="map[string]any" - /// dotnet="IReadOnlyDictionary" - /// dotnet-from="ProtoExtensions.FromHeaderProto" - /// dotnet-to="ProtoExtensions.ToHeaderProto" - /// typescript-import="@temporalio/common" - type header = placeholder; - - /// @nexus.proto "temporal.api.common.v1.SearchAttributes" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="temporalio.common.TypedSearchAttributes" - /// typescript="common.TypedSearchAttributes" - /// dotnet="Temporalio.Common.SearchAttributeCollection" - /// dotnet-from="ProtoExtensions.FromSearchAttributesProto" - /// typescript-import="@temporalio/common" - type search-attributes = placeholder; - - /// @nexus.proto "temporal.api.common.v1.Priority" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="temporalio.common.Priority" - /// typescript="common.Priority" - /// dotnet="Temporalio.Common.Priority" - /// dotnet-from="ProtoExtensions.FromPriorityProto" - /// typescript-import="@temporalio/common" - type priority = placeholder; - - /// @nexus.proto "temporal.api.workflow.v1.VersioningOverride" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="temporalio.common.VersioningOverride" - /// typescript="common.VersioningOverride" - /// dotnet="Temporalio.Common.VersioningOverride" - /// dotnet-from="ProtoExtensions.FromVersioningOverrideProto" - /// typescript-import="@temporalio/common" - type versioning-override = placeholder; - - /// @nexus.proto "google.protobuf.Duration" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="datetime.timedelta" - /// typescript="common.Duration" - /// dotnet="System.TimeSpan" - /// dotnet-from="ProtoExtensions.FromDurationProto" - /// typescript-import="@temporalio/common" - type duration = placeholder; - - /// @nexus.proto "temporal.api.enums.v1.WorkflowIdReusePolicy" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="temporalio.common.WorkflowIDReusePolicy" - /// typescript="common.WorkflowIdReusePolicy" - /// dotnet="Temporalio.Api.Enums.V1.WorkflowIdReusePolicy" - /// typescript-import="@temporalio/common" - type workflow-id-reuse-policy = placeholder; - - /// @nexus.proto "temporal.api.enums.v1.WorkflowIdConflictPolicy" typescript-import="@temporalio/proto" - /// @nexus.type - /// python="temporalio.common.WorkflowIDConflictPolicy" - /// typescript="common.WorkflowIdConflictPolicy" - /// dotnet="Temporalio.Api.Enums.V1.WorkflowIdConflictPolicy" - /// typescript-import="@temporalio/common" - type workflow-id-conflict-policy = placeholder; - - /// @nexus.proto "temporal.api.sdk.v1.UserMetadata" typescript-import="@temporalio/proto" - /// @nexus.flatten-in-api - record user-metadata { - /// @nexus.doc "Single-line fixed summary for the workflow execution that may appear in UI and CLI. This can be in single-line Temporal Markdown format." - /// @nexus.proto-field "summary" - /// @nexus.flattened-type python="str" typescript="string" dotnet="string" - static-summary: option, - /// @nexus.doc "General fixed details for the workflow execution that may appear in UI and CLI. This can be in Temporal Markdown format and can span multiple lines. This value is fixed on the workflow execution and cannot be updated." - /// @nexus.proto-field "details" - /// @nexus.flattened-type python="str" typescript="string" dotnet="string" - static-details: option, - } -} diff --git a/temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml b/temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml deleted file mode 100644 index ebc74b68b..000000000 --- a/temporalio/nexus/system/wit/temporal-proto-models-nexusrpc.yaml +++ /dev/null @@ -1,19 +0,0 @@ -nexusrpc: 1.0.0 -services: - temporal.api.workflowservice.v1.WorkflowService: - operations: - SignalWithStartWorkflowExecution: - input: - $dotnetRef: Temporalio.Api.WorkflowService.V1.SignalWithStartWorkflowExecutionRequest - $goRef: go.temporal.io/api/workflowservice/v1.SignalWithStartWorkflowExecutionRequest - $javaRef: io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest - $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest - $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionRequest - $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionRequest' - output: - $dotnetRef: Temporalio.Api.WorkflowService.V1.SignalWithStartWorkflowExecutionResponse - $goRef: go.temporal.io/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse - $javaRef: io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse - $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse - $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionResponse - $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse' diff --git a/temporalio/nexus/system/wit/workflow-service.wit b/temporalio/nexus/system/wit/workflow-service.wit deleted file mode 100644 index 190aae88c..000000000 --- a/temporalio/nexus/system/wit/workflow-service.wit +++ /dev/null @@ -1,130 +0,0 @@ -package temporal:nexus@1.0.0; - -world system { - export workflow-service; -} - -/// @nexus.endpoint "__temporal_system" -/// @nexus.service-name "temporal.api.workflowservice.v1.WorkflowService" -/// @nexus.namespace dotnet="Temporalio.Workflows" -/// @nexus.operations-class dotnet="Workflow" -/// @nexus.delay-load-temporalio-workflow -/// @nexus.experimental -interface workflow-service { - use nexus:temporal-types/model@1.0.0.{ - duration, - header, - memo, - payloads, - placeholder, - priority, - retry-policy, - search-attributes, - signal-function, - task-queue, - user-metadata, - versioning-override, - workflow-function, - workflow-id-conflict-policy, - workflow-id-reuse-policy, - }; - - /// @nexus.doc "Request fields for signaling a workflow, starting it first if needed." - /// @nexus.experimental - /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest" typescript-import="@temporalio/proto" - record signal-with-start-workflow-request { - /// @nexus.doc - /// python="Workflow type name or callable identifying the workflow to start." - /// typescript="Workflow type name or workflow function identifying the workflow to start." - /// dotnet="Workflow type name or workflow expression identifying the workflow to start." - /// @nexus.proto-field "workflow_type" - workflow: workflow-function, - /// @nexus.doc "Unique identifier for the workflow execution." - /// @nexus.proto-field "workflow_id" - id: string, - /// @nexus.doc "Task queue to run the workflow on." - task-queue: task-queue, - /// @nexus.doc - /// python="Signal name or callable to send with the start request." - /// typescript="Signal name or signal definition to send with the start request." - /// dotnet="Signal name or signal expression to send with the start request." - /// @nexus.proto-field "signal_name" - signal: signal-function, - /// @nexus.doc "Total workflow execution timeout, including retries and continue-as-new." - /// @nexus.proto-field "workflow_execution_timeout" - execution-timeout: option, - /// @nexus.doc "Timeout of a single workflow run." - /// @nexus.proto-field "workflow_run_timeout" - run-timeout: option, - /// @nexus.doc "Timeout of a single workflow task." - /// @nexus.proto-field "workflow_task_timeout" - task-timeout: option, - /// @nexus.omit - identity: placeholder, - /// @nexus.omit - request-id: placeholder, - /// @nexus.doc "Behavior when a closed workflow with the same ID exists. Default is allow-duplicate." - /// @nexus.proto-field "workflow_id_reuse_policy" - /// @nexus.default "allow-duplicate" - id-reuse-policy: workflow-id-reuse-policy, - /// @nexus.doc "Behavior when a workflow is currently running with the same ID. Set to use-existing for idempotent deduplication on workflow ID. Cannot be set if id-reuse-policy is terminate-if-running." - /// @nexus.proto-field "workflow_id_conflict_policy" - id-conflict-policy: option, - /// @nexus.doc "Retry policy for the workflow." - retry-policy: option, - /// @nexus.doc "Cron schedule for recurring workflow executions. See https://docs.temporal.io/cron-job." - cron-schedule: option, - /// @nexus.doc "Memo for the workflow." - memo: option, - /// @nexus.doc "Typed search attributes for the workflow." - search-attributes: option, - /// @nexus.doc "Priority of the workflow execution." - priority: option, - /// @nexus.doc "Override for workflow versioning behavior." - versioning-override: option, - /// @nexus.doc "Amount of time to wait before starting the workflow. This does not work with cron-schedule." - /// @nexus.proto-field "workflow_start_delay" - start-delay: option, - user-metadata: option, - /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" - namespace: string, - /// @nexus.omit - control: placeholder, - /// @nexus.api-omit - /// @nexus.proto-field "header" - headers: option
, - /// @nexus.omit - links: placeholder, - /// @nexus.omit - time-skipping-config: placeholder, - } - - /// @nexus.experimental - /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse" typescript-import="@temporalio/proto" - record signal-with-start-workflow-response { - run-id: option, - started: option, - /// @nexus.omit - signal-link: placeholder, - /// @nexus.omit - first-execution-run-id: placeholder, - } - - /// @nexus.doc - /// "Signal a workflow, starting it first if needed." - /// returns="A workflow handle to the started workflow." - /// @nexus.output-transform - /// python-type="temporalio.workflow.ExternalWorkflowHandle[WorkflowResult]" - /// python="temporalio.workflow.get_external_workflow_handle(request.id, run_id=result.run_id)" - /// typescript-type="workflow.ExternalWorkflowHandle" - /// typescript="workflow.getExternalWorkflowHandle(request.id, result.runId ?? undefined)" - /// typescript-import="@temporalio/workflow" - /// dotnet-type="Temporalio.Workflows.ExternalWorkflowHandle" - /// dotnet="Temporalio.Workflows.Workflow.GetExternalWorkflowHandle(request.Id, result.RunId)" - /// @nexus.operation name="SignalWithStartWorkflowExecution" - /// @nexus.serialization-context python="signal_with_start_workflow_serialization_context" dotnet="WorkflowServiceSerializationContexts.SignalWithStartWorkflow" - /// @nexus.experimental - signal-with-start-workflow: func( - request: signal-with-start-workflow-request, - ) -> signal-with-start-workflow-response; -} From db9d72efb59cfb35c0dde72ba6903098c84100f6 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 31 Aug 2026 08:45:40 -0700 Subject: [PATCH 4/6] chore: update NexGen to 0.2.3 --- pyproject.toml | 5 + scripts/gen_nexus_system_api.py | 2 +- .../nexus/system/workflow_service/__init__.py | 2 +- .../workflow_service/_support/__init__.py | 2 +- .../_system_nexus_interceptor.py | 2 +- .../nexus/system/workflow_service/models.py | 126 ++++++++++-------- .../workflow_service/operations/__init__.py | 2 +- .../operations/signal_with_start_workflow.py | 2 +- .../nexus/system/workflow_service/services.py | 2 +- 9 files changed, 79 insertions(+), 66 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d05d2f8c6..ec621ea69 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -187,8 +187,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 0c410f009..b45616434 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -35,7 +35,7 @@ / "v1" / "request_response.proto" ) -NEX_GEN_VERSION = "0.2.2" +NEX_GEN_VERSION = "0.2.3" def nex_gen_command() -> list[str]: diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index 49f7e4c12..3d7dac503 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/__init__.py b/temporalio/nexus/system/workflow_service/_support/__init__.py index a07fc7273..f56c1cca7 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 nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py index 8f21b9bdb..2f5e54ee2 100644 --- a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py +++ b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index 5361e77b6..7e0c9f2eb 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. DO NOT EDIT! from __future__ import annotations @@ -46,9 +46,41 @@ ) +@dataclasses.dataclass(slots=True, kw_only=True) +class SignalWithStartWorkflowRequest: + """ + .. warning:: + This API is experimental and subject to change. + """ + + workflow: str | collections.abc.Callable[..., collections.abc.Awaitable[object]] + args: list[typing.Any] | None = None + id: str + task_queue: str + signal: str | collections.abc.Callable[..., None | collections.abc.Awaitable[None]] + signal_args: list[typing.Any] | None = None + execution_timeout: datetime.timedelta | None = None + run_timeout: datetime.timedelta | None = None + task_timeout: datetime.timedelta | None = None + id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ( + temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE + ) + id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = None + retry_policy: temporalio.common.RetryPolicy | None = None + cron_schedule: str | None = None + memo: collections.abc.Mapping[str, typing.Any] | None = None + search_attributes: temporalio.common.TypedSearchAttributes | None = None + priority: temporalio.common.Priority | None = None + 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) + headers: collections.abc.Mapping[str, typing.Any] | None = None + + class _SignalWithStartWorkflowRequestTransferTypeConverter( temporalio.converter.TransferTypeConverter[ - "SignalWithStartWorkflowRequest", + SignalWithStartWorkflowRequest, temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, ] ): @@ -63,8 +95,8 @@ class _SignalWithStartWorkflowRequestTransferTypeConverter( def from_transfer_type( self, value: temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, - type_hint: type["SignalWithStartWorkflowRequest"], - ) -> "SignalWithStartWorkflowRequest": + type_hint: type[SignalWithStartWorkflowRequest], + ) -> SignalWithStartWorkflowRequest: if not value.HasField("workflow_type"): raise ValueError( "missing required field SignalWithStartWorkflowRequest.workflow" @@ -142,7 +174,7 @@ def from_transfer_type( @typing_extensions.override def to_transfer_type( self, - value: "SignalWithStartWorkflowRequest", + 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(value.workflow)) @@ -200,44 +232,20 @@ def to_transfer_type( return message -@temporalio.converter.transfer_type_convertible( +_ = temporalio.converter.transfer_type_convertible( _SignalWithStartWorkflowRequestTransferTypeConverter -) -@dataclasses.dataclass(slots=True, kw_only=True) -class SignalWithStartWorkflowRequest: - """ - .. warning:: - This API is experimental and subject to change. - """ +)(SignalWithStartWorkflowRequest) - workflow: str | collections.abc.Callable[..., collections.abc.Awaitable[object]] - args: list[typing.Any] | None = None - id: str - task_queue: str - signal: str | collections.abc.Callable[..., None | collections.abc.Awaitable[None]] - signal_args: list[typing.Any] | None = None - execution_timeout: datetime.timedelta | None = None - run_timeout: datetime.timedelta | None = None - task_timeout: datetime.timedelta | None = None - id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ( - temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE - ) - id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = None - retry_policy: temporalio.common.RetryPolicy | None = None - cron_schedule: str | None = None - memo: collections.abc.Mapping[str, typing.Any] | None = None - search_attributes: temporalio.common.TypedSearchAttributes | None = None - priority: temporalio.common.Priority | None = None - 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) - headers: collections.abc.Mapping[str, typing.Any] | None = None + +@dataclasses.dataclass(slots=True) +class UserMetadata: + static_summary: typing.Any | None = None + static_details: typing.Any | None = None class _UserMetadataTransferTypeConverter( temporalio.converter.TransferTypeConverter[ - "UserMetadata", temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata + UserMetadata, temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata ] ): transfer_type: type[temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata] | None = ( @@ -248,8 +256,8 @@ class _UserMetadataTransferTypeConverter( def from_transfer_type( self, value: temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata, - type_hint: type["UserMetadata"], - ) -> "UserMetadata": + type_hint: type[UserMetadata], + ) -> UserMetadata: return UserMetadata( static_summary=payload_from_proto(value.summary) if value.HasField("summary") @@ -262,7 +270,7 @@ def from_transfer_type( @typing_extensions.override def to_transfer_type( self, - value: "UserMetadata", + value: UserMetadata, ) -> temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata: message = temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata() if value.static_summary is not None: @@ -272,16 +280,25 @@ def to_transfer_type( return message -@temporalio.converter.transfer_type_convertible(_UserMetadataTransferTypeConverter) +_ = temporalio.converter.transfer_type_convertible(_UserMetadataTransferTypeConverter)( + UserMetadata +) + + @dataclasses.dataclass(slots=True) -class UserMetadata: - static_summary: typing.Any | None = None - static_details: typing.Any | None = None +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", + SignalWithStartWorkflowResponse, temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, ] ): @@ -296,8 +313,8 @@ class _SignalWithStartWorkflowResponseTransferTypeConverter( def from_transfer_type( self, value: temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, - type_hint: type["SignalWithStartWorkflowResponse"], - ) -> "SignalWithStartWorkflowResponse": + 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, @@ -306,7 +323,7 @@ def from_transfer_type( @typing_extensions.override def to_transfer_type( self, - value: "SignalWithStartWorkflowResponse", + 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: @@ -316,15 +333,6 @@ def to_transfer_type( return message -@temporalio.converter.transfer_type_convertible( +_ = temporalio.converter.transfer_type_convertible( _SignalWithStartWorkflowResponseTransferTypeConverter -) -@dataclasses.dataclass(slots=True) -class SignalWithStartWorkflowResponse: - """ - .. warning:: - This API is experimental and subject to change. - """ - - run_id: str | None = None - started: bool | None = None +)(SignalWithStartWorkflowResponse) diff --git a/temporalio/nexus/system/workflow_service/operations/__init__.py b/temporalio/nexus/system/workflow_service/operations/__init__.py index 3c550f69c..e5abdafea 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 nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. 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 97eb8159a..c04b115b3 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 nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/services.py b/temporalio/nexus/system/workflow_service/services.py index 237565fff..9d46fe40a 100644 --- a/temporalio/nexus/system/workflow_service/services.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.2. DO NOT EDIT! +# Generated by nexgen v0.2.3. DO NOT EDIT! from __future__ import annotations From 5174c0840582464d56975726356a3f96846ae46e Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 31 Aug 2026 08:55:07 -0700 Subject: [PATCH 5/6] fix: update bridge for SDK Core --- temporalio/bridge/Cargo.lock | 30 ++++++++++++++++++------------ temporalio/bridge/src/envconfig.rs | 5 +++++ temporalio/bridge/src/worker.rs | 8 ++++---- 3 files changed, 27 insertions(+), 16 deletions(-) diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index 745d3b358..72f89d746 100644 --- a/temporalio/bridge/Cargo.lock +++ b/temporalio/bridge/Cargo.lock @@ -156,6 +156,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "bindgen" version = "0.72.1" @@ -539,7 +545,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -923,7 +929,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -1512,7 +1518,7 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8edd1efdd8ab23ba9cb9ace3d9987a72663d5d7c9f74fa00b51d6213645cf6c" dependencies = [ - "base64", + "base64 0.22.1", "serde", ] @@ -1934,7 +1940,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2068,7 +2074,7 @@ version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-core", @@ -2154,7 +2160,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2213,7 +2219,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2554,7 +2560,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2587,7 +2593,7 @@ dependencies = [ "anyhow", "async-trait", "backon", - "base64", + "base64 0.23.1", "bon", "bytes", "derive_more", @@ -2692,7 +2698,7 @@ name = "temporalio-protos" version = "0.7.0" dependencies = [ "anyhow", - "base64", + "base64 0.23.1", "derive_more", "http", "pbjson", @@ -2926,7 +2932,7 @@ checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" dependencies = [ "async-trait", "axum", - "base64", + "base64 0.22.1", "bytes", "flate2", "h2", @@ -3381,7 +3387,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/temporalio/bridge/src/envconfig.rs b/temporalio/bridge/src/envconfig.rs index 7ce2f4664..0b21d1987 100644 --- a/temporalio/bridge/src/envconfig.rs +++ b/temporalio/bridge/src/envconfig.rs @@ -19,6 +19,11 @@ fn data_source_to_dict(py: Python, ds: &DataSource) -> PyResult> { match ds { DataSource::Path(p) => dict.set_item("path", p)?, DataSource::Data(d) => dict.set_item("data", PyBytes::new(py, d))?, + _ => { + return Err(PyRuntimeError::new_err( + "unsupported configuration data source", + )) + } }; Ok(dict.into()) } diff --git a/temporalio/bridge/src/worker.rs b/temporalio/bridge/src/worker.rs index 321dc6560..e74698e94 100644 --- a/temporalio/bridge/src/worker.rs +++ b/temporalio/bridge/src/worker.rs @@ -900,10 +900,10 @@ fn convert_versioning_strategy( WorkerVersioningStrategy::DeploymentBased(options) => { temporalio_sdk_core::WorkerVersioningStrategy::WorkerDeploymentBased( temporalio_common::worker::WorkerDeploymentOptions::new( - temporalio_common::worker::WorkerDeploymentVersion { - deployment_name: options.version.deployment_name, - build_id: options.version.build_id, - }, + temporalio_common::worker::WorkerDeploymentVersion::builder() + .deployment_name(options.version.deployment_name) + .build_id(options.version.build_id) + .build(), ) .use_worker_versioning(options.use_worker_versioning) .maybe_default_versioning_behavior(if options.use_worker_versioning { From 980821276bcd1778f512cee980e622a8215d2508 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 31 Aug 2026 12:20:15 -0700 Subject: [PATCH 6/6] ci: use NexGen 0.2.3 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b671f00a3..1f5fae3d4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -238,7 +238,7 @@ jobs: - run: uv add --dev --python 3.10 "googleapis-common-protos==1.70.0" - run: uv add --python 3.10 "protobuf<4" - run: uv sync --all-extras - - run: cargo install --locked nexgen --version 0.2.2 --features advanced --force + - run: cargo install --locked nexgen --version 0.2.3 --features advanced --force - run: poe build-develop - run: poe gen-protos - name: Check generation unchanged