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

scripts/nex_gen_support.py

@@ -12,6 +12,11 @@
 import temporalio.converter
 
 
+class SignalWithStartWorkflowRequest(typing.Protocol):
+    namespace: str
+    id: str
+
+
 def retry_policy_from_proto(
     proto: common_pb2.RetryPolicy,
 ) -> temporalio.common.RetryPolicy:
@@ -70,6 +75,15 @@ def workflow_namespace() -> str:
     return info().namespace
 
 
+def signal_with_start_workflow_serialization_context(
+    request: SignalWithStartWorkflowRequest,
+) -> 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:

temporalio/nexus/system/__init__.py

@@ -121,6 +121,20 @@ def _get_payload_converter(  # pyright: ignore[reportUnusedFunction]
     return _SystemNexusPayloadConverter(user_payload_converter)
 
 
+def _get_serialization_context(  # pyright: ignore[reportUnusedFunction]
+    service: str,
+    operation: str,
+    request: Any,
+) -> temporalio.converter.SerializationContext | None:
+    """Return the target serialization context for a system Nexus operation."""
+    from .workflow_service import __nexus_operation_registry__
+
+    operation_info = __nexus_operation_registry__.get((service, operation))
+    if operation_info is None or operation_info.serialization_context is None:
+        return None
+    return operation_info.serialization_context(request)
+
+
 __all__ = [
     "TEMPORAL_SYSTEM_ENDPOINT",
     "is_system_endpoint",

temporalio/nexus/system/workflow_service/__init__.py

@@ -2,17 +2,50 @@
 
 from __future__ import annotations
 
-from . import service as _service
+import collections.abc
+import typing
+
+import nexusrpc
+
+import temporalio.converter
+
+from . import services as _services
+from ._support import signal_with_start_workflow_serialization_context
 from .operations.signal_with_start_workflow import signal_with_start_workflow
 
 __all__ = [
     "signal_with_start_workflow",
 ]
 
 
+_InputT = typing.TypeVar("_InputT")
+_OutputT = typing.TypeVar("_OutputT")
+
+
+_SerializationContextFactory = collections.abc.Callable[
+    [_InputT], temporalio.converter.SerializationContext
+]
+
+
+class _NexusOperationInfo(typing.Generic[_InputT, _OutputT]):
+    def __init__(
+        self,
+        *,
+        operation: nexusrpc.Operation[_InputT, _OutputT],
+        serialization_context: _SerializationContextFactory[_InputT] | None = None,
+    ) -> None:
+        self.operation: nexusrpc.Operation[_InputT, _OutputT] = operation
+        self.serialization_context: _SerializationContextFactory[_InputT] | None = (
+            serialization_context
+        )
+
+
 __nexus_operation_registry__ = {
     (
         "temporal.api.workflowservice.v1.WorkflowService",
         "SignalWithStartWorkflowExecution",
-    ): _service.WorkflowService.signal_with_start_workflow,
+    ): _NexusOperationInfo(
+        operation=_services.WorkflowService.signal_with_start_workflow,
+        serialization_context=signal_with_start_workflow_serialization_context,
+    ),
 }

temporalio/nexus/system/workflow_service/_support/nex_gen_support.py

@@ -12,6 +12,11 @@
 import temporalio.converter
 
 
+class SignalWithStartWorkflowRequest(typing.Protocol):
+    namespace: str
+    id: str
+
+
 def retry_policy_from_proto(
     proto: common_pb2.RetryPolicy,
 ) -> temporalio.common.RetryPolicy:
@@ -70,6 +75,15 @@ def workflow_namespace() -> str:
     return info().namespace
 
 
+def signal_with_start_workflow_serialization_context(
+    request: SignalWithStartWorkflowRequest,
+) -> 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: