Pin workflows across ContinueAsNew and implement Cross-TQ inheritance (#7776)

## What changed?
- Updated worker versioning inheritance logic for workflow chains
(parent-child and continue-as-new)
- Replaced deprecated pinned_deployment_version with new
inherited_pinned_version field in workflow execution
- Handled versioning inheritance for retry and cron workflows
- Allow cross-task-queue CaN and Child pinned version inheritance if new
task queue is in the inherited version

## Why?
- Improves versioning behavior consistency across workflow chains by
properly inheriting versioning info from parent/previous runs
- Removes deprecated fields and aligns with new versioning API design
- Fixes edge cases in versioning inheritance for retry and cron
workflows

Basically, we want `PINNED` behavior to carry through parent child
chains, and across the CaN boundary, so that we can roll out a new
behavior for Trampolining in the future, which will not be inherited
across the CaN boundary.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [ ] added new unit test(s)
- [x] added new functional test(s)

## Potential risks
Could break history :P

Author: carlydf

api/historyservice/v1/request_response.pb.go

@@ -15,12 +15,15 @@ import (
 	"go.temporal.io/api/serviceerror"
 	workflowpb "go.temporal.io/api/workflow/v1"
 	deploymentspb "go.temporal.io/server/api/deployment/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence/visibility/manager"
+	"go.temporal.io/server/common/resource"
 	"go.temporal.io/server/common/searchattribute"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
@@ -257,6 +260,64 @@ func MakeDirectiveForWorkflowTask(
 	return nil
 }
 
+type IsWFTaskQueueInVersionDetector = func(ctx context.Context, namespaceID, tq string, version *deploymentpb.WorkerDeploymentVersion) (bool, error)
+
+func GetIsWFTaskQueueInVersionDetector(matchingClient resource.MatchingClient) IsWFTaskQueueInVersionDetector {
+	return func(ctx context.Context,
+		namespaceID, tq string,
+		version *deploymentpb.WorkerDeploymentVersion) (bool, error) {
+		resp, err := matchingClient.GetTaskQueueUserData(ctx,
+			&matchingservice.GetTaskQueueUserDataRequest{
+				NamespaceId:   namespaceID,
+				TaskQueue:     tq,
+				TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
+			})
+		if err != nil {
+			return false, err
+		}
+		tqData, ok := resp.GetUserData().GetData().GetPerType()[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)]
+		if !ok {
+			// The TQ is unversioned
+			return false, nil
+		}
+		return HasDeploymentVersion(tqData.GetDeploymentData(), DeploymentVersionFromDeployment(DeploymentFromExternalDeploymentVersion(version))), nil
+	}
+}
+
+// [cleanup-wv-pre-release]
+func FindDeployment(deployments *persistencespb.DeploymentData, deployment *deploymentpb.Deployment) int {
+	for i, d := range deployments.GetDeployments() { //nolint:staticcheck // SA1019: worker versioning v0.30
+		if d.Deployment.Equal(deployment) {
+			return i
+		}
+	}
+	return -1
+}
+
+func FindDeploymentVersion(deployments *persistencespb.DeploymentData, v *deploymentspb.WorkerDeploymentVersion) int {
+	for i, vd := range deployments.GetVersions() {
+		if proto.Equal(v, vd.GetVersion()) {
+			return i
+		}
+	}
+	return -1
+}
+
+//nolint:staticcheck
+func HasDeploymentVersion(deployments *persistencespb.DeploymentData, v *deploymentspb.WorkerDeploymentVersion) bool {
+	for _, d := range deployments.GetDeployments() {
+		if d.Deployment.Equal(DeploymentFromDeploymentVersion(v)) {
+			return true
+		}
+	}
+	for _, vd := range deployments.GetVersions() {
+		if proto.Equal(v, vd.GetVersion()) {
+			return true
+		}
+	}
+	return false
+}
+
 // DeploymentVersionFromDeployment Temporary helper function to convert Deployment to
 // WorkerDeploymentVersion proto until we update code to use the new proto in all places.
 func DeploymentVersionFromDeployment(deployment *deploymentpb.Deployment) *deploymentspb.WorkerDeploymentVersion {
@@ -397,6 +458,17 @@ func OverrideIsPinned(override *workflowpb.VersioningOverride) bool {
 		override.GetPinned().GetBehavior() == workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED
 }
 
+func GetOverridePinnedVersion(override *workflowpb.VersioningOverride) *deploymentpb.WorkerDeploymentVersion {
+	if OverrideIsPinned(override) {
+		if v := override.GetPinned().GetVersion(); v != nil {
+			return v
+		} else if v := override.GetPinnedVersion(); v != "" { //nolint:staticcheck // SA1019: worker versioning v0.31
+			return ExternalWorkerDeploymentVersionFromStringV31(v)
+		}
+		return ExternalWorkerDeploymentVersionFromDeployment(override.GetDeployment()) //nolint:staticcheck // SA1019: worker versioning v0.30
+	}
+	return nil
+}
 func ExtractVersioningBehaviorFromOverride(override *workflowpb.VersioningOverride) enumspb.VersioningBehavior {
 	if override.GetAutoUpgrade() {
 		return enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE

api/workflow/v1/message.pb.go

@@ -59,7 +59,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.34.0
 	go.opentelemetry.io/otel/sdk/metric v1.34.0
 	go.opentelemetry.io/otel/trace v1.34.0
-	go.temporal.io/api v1.49.2-0.20250528164413-b08ef1e10ee0
+	go.temporal.io/api v1.49.2-0.20250530003406-6ac9b437f9db
 	go.temporal.io/sdk v1.34.0
 	go.temporal.io/version v0.3.0
 	go.uber.org/automaxprocs v1.6.0

common/worker_versioning/worker_versioning.go

@@ -385,8 +385,8 @@ go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC
 go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
 go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU/3i4=
 go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4=
-go.temporal.io/api v1.49.2-0.20250528164413-b08ef1e10ee0 h1:MNgNcZbq0Zu8XtFCMgNY2jzGVweA4JjoaR81xUbD6DE=
-go.temporal.io/api v1.49.2-0.20250528164413-b08ef1e10ee0/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
+go.temporal.io/api v1.49.2-0.20250530003406-6ac9b437f9db h1:RAkllf0vM3U0UBHBjq9iwi4JD36ATl/5gmQciZsrvho=
+go.temporal.io/api v1.49.2-0.20250530003406-6ac9b437f9db/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
 go.temporal.io/sdk v1.34.0 h1:VLg/h6ny7GvLFVoQPqz2NcC93V9yXboQwblkRvZ1cZE=
 go.temporal.io/sdk v1.34.0/go.mod h1:iE4U5vFrH3asOhqpBBphpj9zNtw8btp8+MSaf5A0D3w=
 go.temporal.io/version v0.3.0 h1:dMrei9l9NyHt8nG6EB8vAwDLLTwx2SvRyucCSumAiig=

go.mod

@@ -82,12 +82,15 @@ message StartWorkflowExecutionRequest {
     // For top-level workflows (ie., without parent), this field must be nil.
     temporal.server.api.workflow.v1.RootExecutionInfo root_execution_info = 11;
     // inherited build ID from parent/previous execution
+    // Deprecated. Use behavior, version, and task queue fields in `parent_execution_info`.
     string inherited_build_id = 12;
     // If set, takes precedence over the Versioning Behavior sent by the SDK on Workflow Task completion.
     // To unset the override after the workflow is running, use UpdateWorkflowExecutionOptions.
     temporal.api.workflow.v1.VersioningOverride versioning_override = 13;
     // If set, we verify the parent-child relationship before applying ID conflict policy WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING
     bool child_workflow_only = 14;
+    // If present, the new workflow should start on this version with pinned base behavior.
+    temporal.api.deployment.v1.WorkerDeploymentVersion inherited_pinned_version = 15;
 }
 
 message StartWorkflowExecutionResponse {

go.sum

@@ -5,7 +5,6 @@ package temporal.server.api.workflow.v1;
 option go_package = "go.temporal.io/server/api/workflow/v1;workflow";
 
 import "temporal/api/common/v1/message.proto";
-import "temporal/api/deployment/v1/message.proto";
 
 import "temporal/server/api/clock/v1/message.proto";
 
@@ -21,14 +20,9 @@ message ParentExecutionInfo {
     // first starts the child workflow, and the child workflow is starting on a Task Queue belonging
     // to the same Worker Deployment Version.
     // Not set in the subsequent execution if the child workflow continues-as-new.
-    // Deprecated. Replaced with `pinned_deployment_version`.
+    // Deprecated. Replaced with `inherited_pinned_version` in WorkflowExecutionStartedEventAttributes.
     string pinned_worker_deployment_version = 7;
-    // When present, child workflow starts as Pinned to this Worker Deployment Version.
-    // Set only if the parent execution is effectively Pinned to a Worker Deployment Version when it
-    // first starts the child workflow, and the child workflow is starting on a Task Queue belonging
-    // to the same Worker Deployment Version.
-    // Not set in the subsequent execution if the child workflow continues-as-new.
-    temporal.api.deployment.v1.WorkerDeploymentVersion pinned_deployment_version = 8;
+    reserved 8;
 }
 
 message RootExecutionInfo {

proto/internal/temporal/server/api/historyservice/v1/request_response.proto

@@ -58,6 +58,7 @@ type (
 		searchAttributesValidator      *searchattribute.Validator
 		persistenceVisibilityMgr       manager.VisibilityManager
 		commandHandlerRegistry         *workflow.CommandHandlerRegistry
+		matchingClient                 matchingservice.MatchingServiceClient
 	}
 )
 
@@ -69,6 +70,7 @@ func NewWorkflowTaskCompletedHandler(
 	searchAttributesValidator *searchattribute.Validator,
 	visibilityManager manager.VisibilityManager,
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
+	matchingClient matchingservice.MatchingServiceClient,
 ) *WorkflowTaskCompletedHandler {
 	return &WorkflowTaskCompletedHandler{
 		config:                     shardContext.GetConfig(),
@@ -90,6 +92,7 @@ func NewWorkflowTaskCompletedHandler(
 		searchAttributesValidator:      searchAttributesValidator,
 		persistenceVisibilityMgr:       visibilityManager,
 		commandHandlerRegistry:         commandHandlerRegistry,
+		matchingClient:                 matchingClient,
 	}
 }
 
@@ -375,6 +378,7 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 			handler.searchAttributesMapperProvider,
 			hasBufferedEventsOrMessages,
 			handler.commandHandlerRegistry,
+			handler.matchingClient,
 		)
 
 		if responseMutations, err = workflowTaskHandler.handleCommands(

proto/internal/temporal/server/api/workflow/v1/message.proto

@@ -125,7 +125,8 @@ func (s *WorkflowTaskCompletedHandlerSuite) SetupSubTest() {
 		nil,
 		nil,
 		nil,
-		api.NewWorkflowConsistencyChecker(s.mockShard, s.workflowCache))
+		api.NewWorkflowConsistencyChecker(s.mockShard, s.workflowCache),
+		nil)
 }
 
 func (s *WorkflowTaskCompletedHandlerSuite) TearDownTest() {

service/history/api/respondworkflowtaskcompleted/api.go

@@ -16,6 +16,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/collection"
@@ -31,6 +32,7 @@ import (
 	"go.temporal.io/server/common/protocol"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/tasktoken"
+	"go.temporal.io/server/common/worker_versioning"
 	"go.temporal.io/server/service/history/api"
 	"go.temporal.io/server/service/history/configs"
 	historyi "go.temporal.io/server/service/history/interfaces"
@@ -73,6 +75,7 @@ type (
 		shard                  historyi.ShardContext
 		tokenSerializer        *tasktoken.Serializer
 		commandHandlerRegistry *workflow.CommandHandlerRegistry
+		matchingClient         matchingservice.MatchingServiceClient
 	}
 
 	workflowTaskFailedCause struct {
@@ -111,6 +114,7 @@ func newWorkflowTaskCompletedHandler(
 	searchAttributesMapperProvider searchattribute.MapperProvider,
 	hasBufferedEventsOrMessages bool,
 	commandHandlerRegistry *workflow.CommandHandlerRegistry,
+	matchingClient matchingservice.MatchingServiceClient,
 ) *workflowTaskCompletedHandler {
 	return &workflowTaskCompletedHandler{
 		identity:                identity,
@@ -142,6 +146,7 @@ func newWorkflowTaskCompletedHandler(
 		shard:                  shard,
 		tokenSerializer:        tasktoken.NewSerializer(),
 		commandHandlerRegistry: commandHandlerRegistry,
+		matchingClient:         matchingClient,
 	}
 }
 
@@ -927,7 +932,7 @@ func (handler *workflowTaskCompletedHandler) handleCommandContinueAsNewWorkflow(
 	}
 
 	if handler.mutableState.GetAssignedBuildId() == "" {
-		// TODO: this is supported in new versioning [cleanup-old-wv]
+		// TODO(carlydf): this is supported in new versioning [cleanup-old-wv]
 		if attr.InheritBuildId && attr.TaskQueue.GetName() != "" && attr.TaskQueue.Name != handler.mutableState.GetExecutionInfo().TaskQueue {
 			err := serviceerror.NewInvalidArgument("ContinueAsNew with UseCompatibleVersion cannot run on different task queue.")
 			return nil, handler.failWorkflowTask(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_CONTINUE_AS_NEW_ATTRIBUTES, err)
@@ -986,6 +991,7 @@ func (handler *workflowTaskCompletedHandler) handleCommandContinueAsNewWorkflow(
 		handler.workflowTaskCompletedID,
 		parentNamespace,
 		attr,
+		worker_versioning.GetIsWFTaskQueueInVersionDetector(handler.matchingClient),
 	)
 	if err != nil {
 		return nil, err