feat: add System Nexus operation interceptors

.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

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

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

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]:
@@ -131,6 +131,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",

temporalio/bridge/proto/workflow_commands/workflow_commands_pb2.pyi

@@ -620,6 +620,7 @@ class ScheduleLocalActivity(google.protobuf.message.Message):
     RETRY_POLICY_FIELD_NUMBER: builtins.int
     LOCAL_RETRY_THRESHOLD_FIELD_NUMBER: builtins.int
     CANCELLATION_TYPE_FIELD_NUMBER: builtins.int
+    INCLUDE_ARGUMENTS_INTO_MARKER_FIELD_NUMBER: builtins.int
     seq: builtins.int
     """Lang's incremental sequence number, used as the operation identifier"""
     activity_id: builtins.str
@@ -684,6 +685,11 @@ class ScheduleLocalActivity(google.protobuf.message.Message):
     confirmed. Lang should default this to `WAIT_CANCELLATION_COMPLETED`, even though proto
     will default to `TRY_CANCEL` automatically.
     """
+    include_arguments_into_marker: builtins.bool
+    """If set, the local activity arguments will be included in the resulting marker under the
+    `input` key. This is disabled by default to avoid increasing history size unless the lang
+    SDK explicitly chooses to expose it.
+    """
     def __init__(
         self,
         *,
@@ -706,6 +712,7 @@ class ScheduleLocalActivity(google.protobuf.message.Message):
         retry_policy: temporalio.api.common.v1.message_pb2.RetryPolicy | None = ...,
         local_retry_threshold: google.protobuf.duration_pb2.Duration | None = ...,
         cancellation_type: global___ActivityCancellationType.ValueType = ...,
+        include_arguments_into_marker: builtins.bool = ...,
     ) -> None: ...
     def HasField(
         self,
@@ -739,6 +746,8 @@ class ScheduleLocalActivity(google.protobuf.message.Message):
             b"cancellation_type",
             "headers",
             b"headers",
+            "include_arguments_into_marker",
+            b"include_arguments_into_marker",
             "local_retry_threshold",
             b"local_retry_threshold",
             "original_schedule_time",

temporalio/bridge/src/envconfig.rs

@@ -19,6 +19,11 @@ fn data_source_to_dict(py: Python, ds: &DataSource) -> PyResult<Py<PyAny>> {
     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())
 }

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 {

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,13 +689,26 @@ 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,
         *,
         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

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)

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
 

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