@@ -0,0 +1 @@
+Add experimental `versioning_override` to `workflow.start_child_workflow` and `workflow.execute_child_workflow` to override inherited parent versioning, and `common.OneTimeVersioningOverride` to route an execution to a deployment version until its first completed workflow task. Child overrides require Temporal Server 1.32 or newer; invalid child overrides raise `exceptions.InvalidVersioningOverrideError`.
@@ -1314,6 +1314,29 @@ def _to_proto(self) -> temporalio.api.workflow.v1.VersioningOverride:
         )
 
 
+@dataclass(frozen=True)
+class OneTimeVersioningOverride(VersioningOverride):
+    """Route workflow tasks to a specific deployment version until a task completes
+    on that version, then clear the override.
+
+    The worker's reported versioning behavior governs subsequent tasks. This does
+    not pin the workflow to the target version.
+
+    .. warning::
+        This API is experimental and requires Temporal Server 1.32 or newer.
+    """
+
+    target_version: WorkerDeploymentVersion
+
+    def _to_proto(self) -> temporalio.api.workflow.v1.VersioningOverride:
+        """Convert to proto representation."""
+        return temporalio.api.workflow.v1.VersioningOverride(
+            one_time=temporalio.api.workflow.v1.VersioningOverride.OneTimeOverride(
+                target_deployment_version=self.target_version._to_proto(),
+            ),
+        )
+
+
 # Should be set as the "arg" argument for _arg_or_args checks where the argument
 # is unset. This is different than None which is a legitimate argument.
 _arg_unset = object()
 
@@ -78,6 +78,16 @@ def __init__(
         self.first_run_id = first_run_id
 
 
+class InvalidVersioningOverrideError(FailureError):
+    """Thrown by a workflow when a child cannot start because its versioning
+    override is invalid.
+    """
+
+    def __init__(self) -> None:
+        """Initialize an invalid versioning override error."""
+        super().__init__("Invalid versioning override")
+
+
 class ActivityAlreadyStartedError(FailureError):
     """Thrown by a client when an activity execution has already started.
 
 
@@ -263,6 +263,16 @@ def versioning_override_from_proto(
                 build_id=version.build_id,
             )
         )
+    if proto.HasField("one_time") and proto.one_time.HasField(
+        "target_deployment_version"
+    ):
+        version = proto.one_time.target_deployment_version
+        return temporalio.common.OneTimeVersioningOverride(
+            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(
 
@@ -311,6 +311,12 @@ class StartChildWorkflowInput:
     versioning_intent: VersioningIntent | None
     static_summary: str | None
     static_details: str | None
+    versioning_override: temporalio.common.VersioningOverride | None = field(
+        default=None, kw_only=True
+    )
+    """.. warning::
+        Child workflow versioning overrides are experimental and may change without notice.
+    """
     event_groups: Sequence[temporalio.workflow.EventGroup] | None = field(
         default=None, kw_only=True
     )
 
@@ -987,6 +987,13 @@ def _apply_resolve_child_workflow_execution_start(
                         job.failed.workflow_id, job.failed.workflow_type
                     )
                 )
+            elif (
+                job.failed.cause
+                == temporalio.bridge.proto.child_workflow.StartChildWorkflowExecutionFailedCause.START_CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_INVALID_VERSIONING_OVERRIDE
+            ):
+                handle._resolve_failure(
+                    temporalio.exceptions.InvalidVersioningOverrideError()
+                )
             else:
                 handle._resolve_failure(
                     RuntimeError(f"Unknown child start fail cause: {job.failed.cause}")
@@ -1655,6 +1662,7 @@ async def workflow_start_child_workflow(
         static_details: str | None = None,
         event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
+        versioning_override: temporalio.common.VersioningOverride | None = None,
     ) -> temporalio.workflow.ChildWorkflowHandle[Any, Any]:
         # Use definition if callable
         name: str
@@ -1692,6 +1700,7 @@ async def workflow_start_child_workflow(
                 arg_types=arg_types,
                 ret_type=ret_type,
                 versioning_intent=versioning_intent,
+                versioning_override=versioning_override,
                 static_summary=static_summary,
                 static_details=static_details,
                 event_groups=event_groups,
@@ -3667,6 +3676,8 @@ def _apply_start_command(self) -> None:
         )
         if self._input.versioning_intent:
             v.versioning_intent = self._input.versioning_intent._to_proto()
+        if self._input.versioning_override:
+            v.versioning_override.CopyFrom(self._input.versioning_override._to_proto())
         if self._input.static_summary:
             command.user_metadata.summary.CopyFrom(
                 self._payload_converter.to_payload(self._input.static_summary)
 
@@ -463,6 +463,7 @@ async def workflow_start_child_workflow(
         static_details: str | None = None,
         event_groups: Sequence[EventGroup] | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
+        versioning_override: temporalio.common.VersioningOverride | None = None,
     ) -> ChildWorkflowHandle[Any, Any]: ...
 
     @abstractmethod
 
@@ -164,7 +164,8 @@ class ChildWorkflowConfig(TypedDict, total=False):
     """TypedDict of config that can be used for :py:func:`start_child_workflow`
     and :py:func:`execute_child_workflow`.
 
-    ``event_groups`` is experimental and may change without notice.
+    ``event_groups`` and ``versioning_override`` are experimental and may change
+    without notice.
     """
 
     id: str | None
@@ -182,6 +183,7 @@ class ChildWorkflowConfig(TypedDict, total=False):
         temporalio.common.SearchAttributes | temporalio.common.TypedSearchAttributes
     )
     versioning_intent: VersioningIntent | None
+    versioning_override: temporalio.common.VersioningOverride | None
     static_summary: str | None
     static_details: str | None
     event_groups: Sequence[EventGroup] | None
@@ -213,6 +215,7 @@ async def start_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ChildWorkflowHandle[SelfType, ReturnType]: ...
 
 
@@ -242,6 +245,7 @@ async def start_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ChildWorkflowHandle[SelfType, ReturnType]: ...
 
 
@@ -271,6 +275,7 @@ async def start_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ChildWorkflowHandle[SelfType, ReturnType]: ...
 
 
@@ -302,6 +307,7 @@ async def start_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ChildWorkflowHandle[Any, Any]: ...
 
 
@@ -331,6 +337,7 @@ async def start_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ChildWorkflowHandle[Any, Any]:
     """Start a child workflow and return its handle.
 
@@ -371,6 +378,9 @@ async def start_child_workflow(
             addition to those active in the current scope. See
             :py:func:`temporalio.workflow.create_event_group`.
         priority: Priority to use for this workflow.
+        versioning_override: Experimental override of the child workflow's versioning
+            behavior. Takes precedence over versioning inherited from the parent
+            workflow. Requires Temporal Server 1.32 or newer.
 
     Returns:
         A workflow handle to the started/existing workflow.
@@ -397,6 +407,7 @@ async def start_child_workflow(
         static_details=static_details,
         event_groups=event_groups,
         priority=priority,
+        versioning_override=versioning_override,
     )
 
 
@@ -425,6 +436,7 @@ async def execute_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ReturnType: ...
 
 
@@ -454,6 +466,7 @@ async def execute_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ReturnType: ...
 
 
@@ -483,6 +496,7 @@ async def execute_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> ReturnType: ...
 
 
@@ -514,6 +528,7 @@ async def execute_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> Any: ...
 
 
@@ -543,6 +558,7 @@ async def execute_child_workflow(
     static_details: str | None = None,
     event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
+    versioning_override: temporalio.common.VersioningOverride | None = None,
 ) -> Any:
     """Start a child workflow and wait for completion.
 
@@ -572,6 +588,7 @@ async def execute_child_workflow(
         static_details=static_details,
         event_groups=event_groups,
         priority=priority,
+        versioning_override=versioning_override,
     )
     return await handle
 
 
@@ -14,6 +14,7 @@
 import temporalio.api.common.v1
 import temporalio.api.failure.v1
 import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2
+import temporalio.common
 import temporalio.converter
 import temporalio.exceptions
 import temporalio.nexus.system as nexus_system
@@ -457,7 +458,23 @@ def _new_unmarked_system_nexus_request_payload() -> temporalio.api.common.v1.Pay
     return payload
 
 
-async def test_nexus_payload_serializer_decodes_system_input() -> None:
+@pytest.mark.parametrize(
+    "versioning_override",
+    [
+        None,
+        temporalio.common.PinnedVersioningOverride(
+            temporalio.common.WorkerDeploymentVersion("deployment", "build")
+        ),
+        temporalio.common.AutoUpgradeVersioningOverride(),
+        temporalio.common.OneTimeVersioningOverride(
+            temporalio.common.WorkerDeploymentVersion("deployment", "build")
+        ),
+    ],
+    ids=["absent", "pinned", "auto-upgrade", "one-time"],
+)
+async def test_nexus_payload_serializer_decodes_system_input(
+    versioning_override: temporalio.common.VersioningOverride | None,
+) -> None:
     """A marked system request is decoded into its generated Nexus model."""
     data_converter = temporalio.converter.default()
     request = workflow_service_models.SignalWithStartWorkflowRequest(
@@ -468,6 +485,7 @@ async def test_nexus_payload_serializer_decodes_system_input() -> None:
         signal="test-signal",
         namespace="target-namespace",
         headers={"test-header": "header-value"},
+        versioning_override=versioning_override,
     )
     payload = nexus_system._get_payload_converter(
         data_converter.payload_converter,
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	+Add experimental `versioning_override` to `workflow.start_child_workflow` and `workflow.execute_child_workflow` to override inherited parent versioning, and `common.OneTimeVersioningOverride` to route an execution to a deployment version until its first completed workflow task. Child overrides require Temporal Server 1.32 or newer; invalid child overrides raise `exceptions.InvalidVersioningOverrideError`.