Use target serialization context for system Nexus signal-with-start

scripts/gen_nexus_system_api.py

@@ -35,15 +35,28 @@
     / "v1"
     / "request_response.proto"
 )
+NEX_GEN_VERSION = "0.2.1"
 
 
 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",
+                "nex-gen",
+                "--version",
+                NEX_GEN_VERSION,
+                "--features",
+                "advanced",
+                "--force",
+            ]
+        )
+    return ["nexgen"]
 
 
 def build_descriptor_set(descriptor_path: Path) -> None:
@@ -114,13 +127,10 @@ def generate_nexus_system_api() -> None:
         subprocess.check_call(
             [
                 *command,
-                "generate",
-                "--lang",
                 "python",
-                "--input",
                 str(wit_path),
-                "--input",
                 str(wit_deps_dir),
+                "--native-api",
                 "--support-file",
                 str(python_support_path),
                 "--descriptors",

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
             ):

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],
     )
 
 
@@ -177,6 +213,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 +235,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")

temporalio/bridge/_visitor.py

@@ -368,6 +368,18 @@ async def _visit_temporal_api_sdk_v1_UserMetadata(
         if o.HasField("details"):
             await self._visit_temporal_api_common_v1_Payload(fs, o.details)
 
+    async def _visit_temporal_api_sdk_v1_EventGroupMarker_Label(
+        self, fs: VisitorFunctions, o: Any
+    ):
+        if o.HasField("label"):
+            await self._visit_temporal_api_common_v1_Payload(fs, o.label)
+
+    async def _visit_temporal_api_sdk_v1_EventGroupMarker(
+        self, fs: VisitorFunctions, o: Any
+    ):
+        if o.HasField("label"):
+            await self._visit_temporal_api_sdk_v1_EventGroupMarker_Label(fs, o.label)
+
     async def _visit_coresdk_workflow_commands_ScheduleActivity(
         self, fs: VisitorFunctions, o: Any
     ):
@@ -479,6 +491,8 @@ async def _visit_coresdk_workflow_commands_WorkflowCommand(
     ):
         if o.HasField("user_metadata"):
             await self._visit_temporal_api_sdk_v1_UserMetadata(fs, o.user_metadata)
+        for v in o.event_group_markers:
+            await self._visit_temporal_api_sdk_v1_EventGroupMarker(fs, v)
         if o.HasField("schedule_activity"):
             await self._visit_coresdk_workflow_commands_ScheduleActivity(
                 fs, o.schedule_activity

temporalio/bridge/proto/workflow_activation/workflow_activation_pb2.pyi

@@ -480,6 +480,8 @@ class InitializeWorkflow(google.protobuf.message.Message):
     START_TIME_FIELD_NUMBER: builtins.int
     ROOT_WORKFLOW_FIELD_NUMBER: builtins.int
     PRIORITY_FIELD_NUMBER: builtins.int
+    ORIGINATING_EVENT_ID_FIELD_NUMBER: builtins.int
+    ORIGINAL_EXECUTION_RUN_ID_FIELD_NUMBER: builtins.int
     workflow_type: builtins.str
     """The identifier the lang-specific sdk uses to execute workflow code"""
     workflow_id: builtins.str
@@ -582,6 +584,13 @@ class InitializeWorkflow(google.protobuf.message.Message):
     @property
     def priority(self) -> temporalio.api.common.v1.message_pb2.Priority:
         """Priority of this workflow execution"""
+    originating_event_id: builtins.int
+    """Event ID of the `WORKFLOW_EXECUTION_STARTED` history event that triggered this job."""
+    original_execution_run_id: builtins.str
+    """The run id recorded on the `WORKFLOW_EXECUTION_STARTED` event. Unlike the execution's current
+    run id, this value is preserved across workflow resets. Mirrors the `original_execution_run_id`
+    field from `WorkflowExecutionStartedEventAttributes`.
+    """
     def __init__(
         self,
         *,
@@ -622,6 +631,8 @@ class InitializeWorkflow(google.protobuf.message.Message):
         root_workflow: temporalio.api.common.v1.message_pb2.WorkflowExecution
         | None = ...,
         priority: temporalio.api.common.v1.message_pb2.Priority | None = ...,
+        originating_event_id: builtins.int = ...,
+        original_execution_run_id: builtins.str = ...,
     ) -> None: ...
     def HasField(
         self,
@@ -683,6 +694,10 @@ class InitializeWorkflow(google.protobuf.message.Message):
             b"last_completion_result",
             "memo",
             b"memo",
+            "original_execution_run_id",
+            b"original_execution_run_id",
+            "originating_event_id",
+            b"originating_event_id",
             "parent_workflow_info",
             b"parent_workflow_info",
             "priority",
@@ -1098,6 +1113,7 @@ class SignalWorkflow(google.protobuf.message.Message):
     INPUT_FIELD_NUMBER: builtins.int
     IDENTITY_FIELD_NUMBER: builtins.int
     HEADERS_FIELD_NUMBER: builtins.int
+    ORIGINATING_EVENT_ID_FIELD_NUMBER: builtins.int
     signal_name: builtins.str
     @property
     def input(
@@ -1114,6 +1130,8 @@ class SignalWorkflow(google.protobuf.message.Message):
         builtins.str, temporalio.api.common.v1.message_pb2.Payload
     ]:
         """Headers attached to the signal"""
+    originating_event_id: builtins.int
+    """Event ID of the `WORKFLOW_EXECUTION_SIGNALED` history event that produced this job."""
     def __init__(
         self,
         *,
@@ -1125,6 +1143,7 @@ class SignalWorkflow(google.protobuf.message.Message):
             builtins.str, temporalio.api.common.v1.message_pb2.Payload
         ]
         | None = ...,
+        originating_event_id: builtins.int = ...,
     ) -> None: ...
     def ClearField(
         self,
@@ -1135,6 +1154,8 @@ class SignalWorkflow(google.protobuf.message.Message):
             b"identity",
             "input",
             b"input",
+            "originating_event_id",
+            b"originating_event_id",
             "signal_name",
             b"signal_name",
         ],