tasks/service/frontend/workflow_handler.go

package frontend

import (
	"cmp"
	"context"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/temporalio/sqlparser"
	batchpb "go.temporal.io/api/batch/v1"
	commonpb "go.temporal.io/api/common/v1"
	deploymentpb "go.temporal.io/api/deployment/v1"
	enumspb "go.temporal.io/api/enums/v1"
	filterpb "go.temporal.io/api/filter/v1"
	historypb "go.temporal.io/api/history/v1"
	querypb "go.temporal.io/api/query/v1"
	schedulepb "go.temporal.io/api/schedule/v1"
	"go.temporal.io/api/serviceerror"
	taskqueuepb "go.temporal.io/api/taskqueue/v1"
	updatepb "go.temporal.io/api/update/v1"
	workerpb "go.temporal.io/api/worker/v1"
	workflowpb "go.temporal.io/api/workflow/v1"
	"go.temporal.io/api/workflowservice/v1"
	batchspb "go.temporal.io/server/api/batch/v1"
	deploymentspb "go.temporal.io/server/api/deployment/v1"
	"go.temporal.io/server/api/historyservice/v1"
	"go.temporal.io/server/api/matchingservice/v1"
	schedulespb "go.temporal.io/server/api/schedule/v1"
	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
	"go.temporal.io/server/chasm"
	"go.temporal.io/server/chasm/lib/activity"
	"go.temporal.io/server/chasm/lib/callback"
	chasmnexus "go.temporal.io/server/chasm/lib/nexusoperation"
	chasmscheduler "go.temporal.io/server/chasm/lib/scheduler"
	schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
	"go.temporal.io/server/chasm/lib/workflow"
	"go.temporal.io/server/client/frontend"
	matchingclient "go.temporal.io/server/client/matching"
	"go.temporal.io/server/common"
	"go.temporal.io/server/common/archiver"
	"go.temporal.io/server/common/archiver/provider"
	"go.temporal.io/server/common/backoff"
	"go.temporal.io/server/common/clock"
	"go.temporal.io/server/common/cluster"
	"go.temporal.io/server/common/collection"
	"go.temporal.io/server/common/contextutil"
	"go.temporal.io/server/common/dynamicconfig"
	"go.temporal.io/server/common/enums"
	"go.temporal.io/server/common/failure"
	"go.temporal.io/server/common/headers"
	"go.temporal.io/server/common/log"
	"go.temporal.io/server/common/log/tag"
	"go.temporal.io/server/common/membership"
	"go.temporal.io/server/common/metrics"
	"go.temporal.io/server/common/namespace"
	"go.temporal.io/server/common/namespace/nsreplication"
	"go.temporal.io/server/common/payload"
	"go.temporal.io/server/common/payloads"
	"go.temporal.io/server/common/persistence"
	"go.temporal.io/server/common/persistence/serialization"
	"go.temporal.io/server/common/persistence/visibility/manager"
	"go.temporal.io/server/common/primitives"
	"go.temporal.io/server/common/primitives/timestamp"
	"go.temporal.io/server/common/priorities"
	"go.temporal.io/server/common/quotas"
	"go.temporal.io/server/common/retrypolicy"
	"go.temporal.io/server/common/rpc"
	"go.temporal.io/server/common/rpc/interceptor"
	"go.temporal.io/server/common/sdk"
	"go.temporal.io/server/common/searchattribute"
	"go.temporal.io/server/common/searchattribute/sadefs"
	"go.temporal.io/server/common/tasktoken"
	"go.temporal.io/server/common/tqid"
	"go.temporal.io/server/common/util"
	"go.temporal.io/server/common/worker_versioning"
	"go.temporal.io/server/service/history/api"
	"go.temporal.io/server/service/worker/batcher"
	"go.temporal.io/server/service/worker/dummy"
	"go.temporal.io/server/service/worker/scheduler"
	"go.temporal.io/server/service/worker/workerdeployment"
	"google.golang.org/grpc/health"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/protobuf/encoding/protowire"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/fieldmaskpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var _ Handler = (*WorkflowHandler)(nil)

var (
	minTime = time.Unix(0, 0).UTC()
	maxTime = time.Date(2100, 1, 1, 1, 0, 0, 0, time.UTC)

	// Tail room for context deadline to bail out from retry for long poll.
	longPollTailRoom  = time.Second
	errWaitForRefresh = serviceerror.NewDeadlineExceeded("waiting for schedule to refresh status of completed workflows")
)

const (
	maxReasonLength              = 1000 // Maximum length for the reason field in RateLimitUpdate configurations.
	defaultUserTerminateReason   = "terminated by user via frontend"
	defaultUserTerminateIdentity = "frontend-service"
)

type (
	// ActivityHandler is the activity frontend handler, aliased to avoid embedding name collision.
	ActivityHandler = activity.FrontendHandler
	// NexusOperationHandler is the nexus operation frontend handler, aliased to avoid embedding name collision.
	NexusOperationHandler = chasmnexus.FrontendHandler

	// WorkflowHandler - gRPC handler interface for workflowservice
	WorkflowHandler struct {
		workflowservice.UnsafeWorkflowServiceServer
		ActivityHandler
		NexusOperationHandler

		validator *workflow.RequestValidator

		status int32

		callbackValidator               callback.Validator
		tokenSerializer                 *tasktoken.Serializer
		config                          *Config
		versionChecker                  headers.VersionChecker
		namespaceHandler                *namespaceHandler
		getDefaultWorkflowRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
		visibilityMgr                   manager.VisibilityManager
		logger                          log.Logger
		throttledLogger                 log.Logger
		persistenceExecutionName        string
		clusterMetadataManager          persistence.ClusterMetadataManager
		clusterMetadata                 cluster.Metadata
		historyClient                   historyservice.HistoryServiceClient
		matchingClient                  matchingservice.MatchingServiceClient
		workerDeploymentClient          workerdeployment.Client
		schedulerClient                 schedulerpb.SchedulerServiceClient
		archiverProvider                provider.ArchiverProvider
		payloadSerializer               serialization.Serializer
		namespaceRegistry               namespace.Registry
		saMapperProvider                searchattribute.MapperProvider
		saProvider                      searchattribute.Provider
		saValidator                     *searchattribute.Validator
		archivalMetadata                archiver.ArchivalMetadata
		healthServer                    *health.Server
		overrides                       *Overrides
		membershipMonitor               membership.Monitor
		healthInterceptor               *interceptor.HealthInterceptor
		scheduleSpecBuilder             *scheduler.SpecBuilder
		outstandingPollers              collection.SyncMap[string, collection.SyncMap[string, context.CancelFunc]]
		httpEnabled                     bool
		registry                        *chasm.Registry
		workerDeploymentReadRateLimiter quotas.RequestRateLimiter
	}
)

func (wh *WorkflowHandler) CreateWorkerDeploymentVersion(
	ctx context.Context,
	request *workflowservice.CreateWorkerDeploymentVersionRequest,
) (_ *workflowservice.CreateWorkerDeploymentVersionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if request.GetDeploymentVersion().GetDeploymentName() == "" {
		return nil, serviceerror.NewInvalidArgument("deployment name cannot be empty")
	}

	if request.GetDeploymentVersion().GetBuildId() == "" {
		return nil, serviceerror.NewInvalidArgument("build ID cannot be empty")
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	requestID := request.RequestId
	if requestID == "" {
		requestID = uuid.NewString()
	}

	err = wh.workerDeploymentClient.CreateWorkerDeploymentVersion(
		ctx,
		namespaceEntry,
		request.GetDeploymentVersion().GetDeploymentName(),
		request.GetDeploymentVersion().GetBuildId(),
		request.Identity,
		requestID,
		request.GetComputeConfig(),
	)
	if err != nil {
		return nil, err
	}

	return &workflowservice.CreateWorkerDeploymentVersionResponse{}, nil
}

func (wh *WorkflowHandler) UpdateWorkerDeploymentVersionComputeConfig(
	ctx context.Context,
	request *workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest,
) (_ *workflowservice.UpdateWorkerDeploymentVersionComputeConfigResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if request.GetDeploymentVersion().GetDeploymentName() == "" {
		return nil, serviceerror.NewInvalidArgument("deployment name cannot be empty")
	}

	if request.GetDeploymentVersion().GetBuildId() == "" {
		return nil, serviceerror.NewInvalidArgument("build ID cannot be empty")
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	requestID := request.RequestId
	if requestID == "" {
		requestID = uuid.NewString()
	}

	err = wh.workerDeploymentClient.UpdateVersionComputeConfig(
		ctx,
		namespaceEntry,
		request.GetDeploymentVersion(),
		request.GetComputeConfigScalingGroups(),
		request.GetRemoveComputeConfigScalingGroups(),
		request.Identity,
		requestID,
	)
	if err != nil {
		return nil, err
	}

	return &workflowservice.UpdateWorkerDeploymentVersionComputeConfigResponse{}, nil
}

func (wh *WorkflowHandler) ValidateWorkerDeploymentVersionComputeConfig(
	ctx context.Context,
	request *workflowservice.ValidateWorkerDeploymentVersionComputeConfigRequest,
) (_ *workflowservice.ValidateWorkerDeploymentVersionComputeConfigResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	err = wh.workerDeploymentClient.ValidateComputeConfig(
		ctx,
		namespaceEntry,
		request.GetDeploymentVersion(),
		request.GetComputeConfigScalingGroups(),
		request.GetRemoveComputeConfigScalingGroups(),
		request.Identity,
	)
	if err != nil {
		return nil, err
	}

	return &workflowservice.ValidateWorkerDeploymentVersionComputeConfigResponse{}, nil
}

// NewWorkflowHandler creates a gRPC handler for workflowservice
func NewWorkflowHandler(
	callbackValidator callback.Validator,
	config *Config,
	namespaceReplicationQueue persistence.NamespaceReplicationQueue,
	visibilityMgr manager.VisibilityManager,
	logger log.Logger,
	throttledLogger log.Logger,
	persistenceExecutionName string,
	clusterMetadataManager persistence.ClusterMetadataManager,
	persistenceMetadataManager persistence.MetadataManager,
	historyClient historyservice.HistoryServiceClient,
	matchingClient matchingservice.MatchingServiceClient,
	workerDeploymentClient workerdeployment.Client,
	schedulerClient schedulerpb.SchedulerServiceClient,
	archiverProvider provider.ArchiverProvider,
	payloadSerializer serialization.Serializer,
	namespaceRegistry namespace.Registry,
	saMapperProvider searchattribute.MapperProvider,
	saProvider searchattribute.Provider,
	saValidator *searchattribute.Validator,
	clusterMetadata cluster.Metadata,
	archivalMetadata archiver.ArchivalMetadata,
	healthServer *health.Server,
	timeSource clock.TimeSource,
	membershipMonitor membership.Monitor,
	healthInterceptor *interceptor.HealthInterceptor,
	scheduleSpecBuilder *scheduler.SpecBuilder,
	httpEnabled bool,
	activityHandler activity.FrontendHandler,
	nexusOperationHandler chasmnexus.FrontendHandler,
	registry *chasm.Registry,
	workerDeploymentReadRateLimiter quotas.RequestRateLimiter,
	validator *workflow.RequestValidator,
) *WorkflowHandler {
	handler := &WorkflowHandler{
		ActivityHandler:       activityHandler,
		NexusOperationHandler: nexusOperationHandler,
		status:                common.DaemonStatusInitialized,
		callbackValidator:     callbackValidator,
		config:                config,
		tokenSerializer:       tasktoken.NewSerializer(),
		versionChecker:        headers.NewDefaultVersionChecker(),
		namespaceHandler: newNamespaceHandler(
			logger,
			persistenceMetadataManager,
			namespaceRegistry,
			clusterMetadata,
			nsreplication.NewReplicator(namespaceReplicationQueue, logger),
			archivalMetadata,
			archiverProvider,
			timeSource,
			config,
		),
		getDefaultWorkflowRetrySettings: config.DefaultWorkflowRetryPolicy,
		visibilityMgr:                   visibilityMgr,
		logger:                          logger,
		throttledLogger:                 throttledLogger,
		persistenceExecutionName:        persistenceExecutionName,
		clusterMetadataManager:          clusterMetadataManager,
		clusterMetadata:                 clusterMetadata,
		historyClient:                   historyClient,
		matchingClient:                  matchingClient,
		workerDeploymentClient:          workerDeploymentClient,
		schedulerClient:                 schedulerClient,
		archiverProvider:                archiverProvider,
		payloadSerializer:               payloadSerializer,
		namespaceRegistry:               namespaceRegistry,
		saProvider:                      saProvider,
		saMapperProvider:                saMapperProvider,
		saValidator:                     saValidator,
		archivalMetadata:                archivalMetadata,
		healthServer:                    healthServer,
		overrides:                       NewOverrides(),
		membershipMonitor:               membershipMonitor,
		healthInterceptor:               healthInterceptor,
		scheduleSpecBuilder:             scheduleSpecBuilder,
		outstandingPollers:              collection.NewSyncMap[string, collection.SyncMap[string, context.CancelFunc]](),
		httpEnabled:                     httpEnabled,
		registry:                        registry,
		workerDeploymentReadRateLimiter: workerDeploymentReadRateLimiter,
		validator:                       validator,
	}

	return handler
}

// Start starts the handler
func (wh *WorkflowHandler) Start() {
	if atomic.CompareAndSwapInt32(
		&wh.status,
		common.DaemonStatusInitialized,
		common.DaemonStatusStarted,
	) {
		// Start in NOT_SERVING state and switch to SERVING after membership is ready
		wh.healthServer.SetServingStatus(WorkflowServiceName, healthpb.HealthCheckResponse_NOT_SERVING)
		go func() {
			_ = wh.membershipMonitor.WaitUntilInitialized(context.Background())
			wh.healthServer.SetServingStatus(WorkflowServiceName, healthpb.HealthCheckResponse_SERVING)
			wh.healthInterceptor.SetHealthy(true)
			wh.logger.Info("Frontend is now healthy")
		}()

		wh.namespaceRegistry.RegisterStateChangeCallback(wh, func(ns *namespace.Namespace, deletedFromDb bool) {
			if deletedFromDb {
				return
			}

			if ns.IsGlobalNamespace() &&
				ns.ReplicationPolicy() == namespace.ReplicationPolicyMultiCluster &&
				!ns.ActiveInCluster(wh.clusterMetadata.GetCurrentClusterName()) {
				pollers, ok := wh.outstandingPollers.Get(ns.ID().String())
				if ok {
					for _, cancelFn := range pollers.PopAll() {
						cancelFn()
					}
				}
			}
		})
	}
}

// Stop stops the handler
func (wh *WorkflowHandler) Stop() {
	if atomic.CompareAndSwapInt32(
		&wh.status,
		common.DaemonStatusStarted,
		common.DaemonStatusStopped,
	) {
		wh.namespaceRegistry.UnregisterStateChangeCallback(wh)
		wh.healthServer.SetServingStatus(WorkflowServiceName, healthpb.HealthCheckResponse_NOT_SERVING)
		wh.healthInterceptor.SetHealthy(false)
	}
}

// GetConfig return config
func (wh *WorkflowHandler) GetConfig() *Config {
	return wh.config
}

// RegisterNamespace creates a new namespace which can be used as a container for all resources.  Namespace is a top level
// entity within Temporal, used as a container for all resources like workflow executions, task queues, etc.  Namespace
// acts as a sandbox and provides isolation for all resources within the namespace.  All resources belong to exactly one
// namespace.
func (wh *WorkflowHandler) RegisterNamespace(ctx context.Context, request *workflowservice.RegisterNamespaceRequest) (_ *workflowservice.RegisterNamespaceResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := wh.validateNamespace(request.GetNamespace()); err != nil {
		return nil, err
	}

	resp, err := wh.namespaceHandler.RegisterNamespace(ctx, request)
	if err != nil {
		return nil, err
	}

	return resp, nil
}

// DescribeNamespace returns the information and configuration for a registered namespace.
func (wh *WorkflowHandler) DescribeNamespace(ctx context.Context, request *workflowservice.DescribeNamespaceRequest) (_ *workflowservice.DescribeNamespaceResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	resp, err := wh.namespaceHandler.DescribeNamespace(ctx, request)
	if err != nil {
		return resp, err
	}
	return resp, err
}

// ListNamespaces returns the information and configuration for all namespaces.
func (wh *WorkflowHandler) ListNamespaces(ctx context.Context, request *workflowservice.ListNamespacesRequest) (_ *workflowservice.ListNamespacesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	resp, err := wh.namespaceHandler.ListNamespaces(ctx, request)
	if err != nil {
		return resp, err
	}
	return resp, err
}

// UpdateNamespace is used to update the information and configuration for a registered namespace.
func (wh *WorkflowHandler) UpdateNamespace(ctx context.Context, request *workflowservice.UpdateNamespaceRequest) (_ *workflowservice.UpdateNamespaceResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	resp, err := wh.namespaceHandler.UpdateNamespace(ctx, request)
	if err != nil {
		return resp, err
	}
	return resp, err
}

// DeprecateNamespace us used to update status of a registered namespace to DEPRECATED.  Once the namespace is deprecated
// it cannot be used to start new workflow executions.  Existing workflow executions will continue to run on
// deprecated namespaces.
// Deprecated.
func (wh *WorkflowHandler) DeprecateNamespace(ctx context.Context, request *workflowservice.DeprecateNamespaceRequest) (_ *workflowservice.DeprecateNamespaceResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	resp, err := wh.namespaceHandler.DeprecateNamespace(ctx, request)
	if err != nil {
		return nil, err
	}
	return resp, err
}

// StartWorkflowExecution starts a new workflow instance (a "workflow execution").  It will create the instance with
// 'WorkflowExecutionStarted' event in history and also schedule the first WorkflowTask for the worker to make the
// first workflow task for this instance.  It will return 'WorkflowExecutionAlreadyStartedError', if an instance already
// exists with same workflowId.
func (wh *WorkflowHandler) StartWorkflowExecution(
	ctx context.Context,
	request *workflowservice.StartWorkflowExecutionRequest,
) (_ *workflowservice.StartWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	var err error
	if request, err = wh.prepareStartWorkflowRequest(ctx, request); err != nil {
		return nil, err
	}

	wh.logger.Debug("Received StartWorkflowExecution.", tag.WorkflowID(request.GetWorkflowId()), tag.WorkflowType(request.GetWorkflowType().GetName()))

	namespaceName := namespace.Name(request.GetNamespace())

	wh.logger.Debug("Start workflow execution request namespace.", tag.WorkflowNamespace(namespaceName.String()))
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}
	wh.logger.Debug("Start workflow execution request namespaceID.", tag.WorkflowNamespaceID(namespaceID.String()))

	resp, err := wh.historyClient.StartWorkflowExecution(
		ctx,
		common.CreateHistoryStartWorkflowRequest(
			namespaceID.String(),
			request,
			nil,
			nil,
			time.Now().UTC(),
		),
	)
	if err != nil {
		return nil, err
	}
	return wh.convertToStartWorkflowExecutionResponse(resp, namespaceName)
}

func (wh *WorkflowHandler) convertToStartWorkflowExecutionResponse(
	resp *historyservice.StartWorkflowExecutionResponse,
	namespaceName namespace.Name,
) (*workflowservice.StartWorkflowExecutionResponse, error) {
	if resp.GetEagerWorkflowTask() != nil {
		if err := api.ProcessOutgoingSearchAttributes(
			wh.saProvider,
			wh.saMapperProvider,
			resp.GetEagerWorkflowTask().GetHistory().GetEvents(),
			namespaceName,
			wh.visibilityMgr,
		); err != nil {
			return nil, err
		}
	}

	return &workflowservice.StartWorkflowExecutionResponse{
		RunId:             resp.GetRunId(),
		Started:           resp.Started,
		EagerWorkflowTask: resp.GetEagerWorkflowTask(),
		Link:              resp.GetLink(),
		Status:            resp.GetStatus(),
	}, nil
}

// Validates the request and sets default values where they are missing.
func (wh *WorkflowHandler) prepareStartWorkflowRequest(
	ctx context.Context,
	request *workflowservice.StartWorkflowExecutionRequest,
) (*workflowservice.StartWorkflowExecutionRequest, error) {
	if request == nil {
		return nil, errRequestNotSet
	}

	// Apply defaults before validation; must be first for idempotency on internal retries.
	enums.SetDefaultWorkflowIDPolicies(
		&request.WorkflowIdReusePolicy,
		&request.WorkflowIdConflictPolicy,
		enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL,
	)

	if err := wh.validator.ValidateWorkflowID(request.GetWorkflowId()); err != nil {
		return nil, err
	}

	namespaceName := namespace.Name(request.GetNamespace())
	if err := wh.validator.ValidateRetryPolicy(request.GetNamespace(), request.RetryPolicy); err != nil {
		return nil, err
	}

	if err := wh.validator.ValidateWorkflowStartDelay(request.GetCronSchedule(), request.WorkflowStartDelay); err != nil {
		return nil, err
	}

	if err := backoff.ValidateSchedule(request.GetCronSchedule()); err != nil {
		return nil, err
	}

	if request.WorkflowType == nil || request.WorkflowType.GetName() == "" {
		return nil, errWorkflowTypeNotSet
	}

	if len(request.WorkflowType.GetName()) > wh.config.MaxIDLengthLimit() {
		return nil, errWorkflowTypeTooLong
	}

	if err := tqid.NormalizeAndValidateUserDefined(request.TaskQueue, "", "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	if err := wh.validator.ValidateWorkflowTimeouts(request); err != nil {
		return nil, err
	}

	if err := validateRequestId(&request.RequestId, wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	if err := wh.validator.ValidateWorkflowIDReusePolicy(
		request.WorkflowIdReusePolicy,
		request.WorkflowIdConflictPolicy); err != nil {
		return nil, err
	}

	if err := wh.validateOnConflictOptions(request.OnConflictOptions); err != nil {
		return nil, err
	}

	sa, err := wh.validator.UnaliasedSearchAttributesFrom(request.GetSearchAttributes(), request.GetNamespace())
	if err != nil {
		return nil, err
	}
	if sa != request.SearchAttributes {
		// Since unaliasedSearchAttributesFrom is not idempotent, we need to clone the request so that
		// in case of retries, the field is set to the original value.
		request = common.CloneProto(request)
		request.SearchAttributes = sa
	}

	if err := priorities.Validate(request.Priority); err != nil {
		return nil, err
	}

	if cbs := request.GetCompletionCallbacks(); len(cbs) > 0 {
		if err := wh.callbackValidator.Validate(ctx, namespaceName.String(), cbs); err != nil {
			return nil, err
		}
	}

	request.Links = dedupLinksFromCallbacks(request.GetLinks(), request.GetCompletionCallbacks())

	allLinks := make([]*commonpb.Link, 0, len(request.GetLinks())+len(request.GetCompletionCallbacks()))
	allLinks = append(allLinks, request.GetLinks()...)
	for _, cb := range request.GetCompletionCallbacks() {
		allLinks = append(allLinks, cb.GetLinks()...)
	}
	if err := wh.validator.ValidateLinks(namespaceName.String(), allLinks); err != nil {
		return nil, err
	}

	if err := wh.validateTimeSkippingConfig(request.GetTimeSkippingConfig(), namespaceName); err != nil {
		return nil, err
	}
	return request, nil
}

func (wh *WorkflowHandler) validateTimeSkippingConfig(
	timeSkippingConfig *workflowpb.TimeSkippingConfig,
	namespaceName namespace.Name,
) error {
	if timeSkippingConfig == nil {
		return nil
	}

	// if this feature is not enabled, we don't allow setting any related config
	if !wh.config.TimeSkippingEnabled(namespaceName.String()) {
		return serviceerror.NewUnimplementedf(
			"The Time-Skipping feature is not enabled for namespace %s",
			namespaceName,
		)
	}

	return nil
}

func (wh *WorkflowHandler) unaliasedSearchAttributesFrom(
	attributes *commonpb.SearchAttributes,
	namespaceName namespace.Name,
) (*commonpb.SearchAttributes, error) {
	sa, err := searchattribute.UnaliasFields(wh.saMapperProvider, attributes, namespaceName.String())
	if err != nil {
		return nil, err
	}

	if err = wh.validator.ValidateSearchAttributes(sa, namespaceName.String()); err != nil {
		return nil, err
	}
	return sa, nil
}

func (wh *WorkflowHandler) ExecuteMultiOperation(
	ctx context.Context,
	request *workflowservice.ExecuteMultiOperationRequest,
) (_ *workflowservice.ExecuteMultiOperationResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceName := namespace.Name(request.Namespace)
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	// as a temporary limitation, the only allowed list of operations is exactly [Start, Update]
	if len(request.Operations) != 2 {
		return nil, errMultiOpNotStartAndUpdate
	}
	if request.Operations[0].GetStartWorkflow() == nil {
		return nil, errMultiOpNotStartAndUpdate
	}
	if request.Operations[1].GetUpdateWorkflow() == nil {
		return nil, errMultiOpNotStartAndUpdate
	}

	metrics.EventBlobSize.With(wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String()))).Record(int64(request.Operations[1].GetUpdateWorkflow().GetRequest().GetInput().GetArgs().Size()), metrics.OperationTag("UpdateWorkflowExecution"))

	historyReq, err := wh.convertToHistoryMultiOperationRequest(ctx, namespaceID, request)
	if err != nil {
		return nil, err
	}

	historyResp, err := wh.historyClient.ExecuteMultiOperation(ctx, historyReq)
	if err != nil {
		var multiErr *serviceerror.MultiOperationExecution
		if errors.As(err, &multiErr) {
			// Tweak error message for end-users to match the feature name.
			// The per-operation errors are embedded inside the error and unpacked by the SDK.
			multiErr.Message = "Update-with-Start could not be executed."
		}
		return nil, err
	}

	response, err := wh.convertToMultiOperationResponse(historyResp, namespaceName)
	if err != nil {
		return nil, err
	}
	return response, nil
}

func (wh *WorkflowHandler) convertToHistoryMultiOperationRequest(
	ctx context.Context,
	namespaceID namespace.ID,
	request *workflowservice.ExecuteMultiOperationRequest,
) (*historyservice.ExecuteMultiOperationRequest, error) {
	var lastWorkflowID string
	ops := make([]*historyservice.ExecuteMultiOperationRequest_Operation, len(request.Operations))

	var hasError bool
	errs := make([]error, len(request.Operations))

	for i, op := range request.Operations {
		convertedOp, opWorkflowID, err := wh.convertToHistoryMultiOperationItem(ctx, namespaceID, namespace.Name(request.Namespace), op)
		if err != nil {
			hasError = true
		} else {
			// set to default in case the whole MultOp request
			err = errMultiOpAborted

			switch {
			case lastWorkflowID == "":
				lastWorkflowID = opWorkflowID
			case lastWorkflowID != opWorkflowID:
				err = errMultiOpWorkflowIdInconsistent
				hasError = true
			}
		}
		errs[i] = err
		ops[i] = convertedOp
	}

	if hasError {
		return nil, serviceerror.NewMultiOperationExecution("Update-with-Start could not be executed.", errs)
	}

	return &historyservice.ExecuteMultiOperationRequest{
		NamespaceId: namespaceID.String(),
		WorkflowId:  lastWorkflowID,
		Operations:  ops,
	}, nil
}

func (wh *WorkflowHandler) convertToHistoryMultiOperationItem(
	ctx context.Context,
	namespaceID namespace.ID,
	namespaceName namespace.Name,
	op *workflowservice.ExecuteMultiOperationRequest_Operation,
) (*historyservice.ExecuteMultiOperationRequest_Operation, string, error) {
	var workflowId string
	var opReq *historyservice.ExecuteMultiOperationRequest_Operation

	if startReq := op.GetStartWorkflow(); startReq != nil {
		if startReq.Namespace != "" && startReq.Namespace != namespaceName.String() {
			return nil, "", errMultiOpNamespaceMismatch
		}
		var err error
		if startReq, err = wh.prepareStartWorkflowRequest(ctx, startReq); err != nil {
			return nil, "", err
		}
		if len(startReq.CronSchedule) > 0 {
			return nil, "", errMultiOpStartCronSchedule
		}
		if startReq.RequestEagerExecution {
			return nil, "", errMultiOpEagerWorkflow
		}
		if timestamp.DurationValue(startReq.WorkflowStartDelay) > 0 {
			return nil, "", errMultiOpStartDelay
		}

		workflowId = startReq.WorkflowId
		opReq = &historyservice.ExecuteMultiOperationRequest_Operation{
			Operation: &historyservice.ExecuteMultiOperationRequest_Operation_StartWorkflow{
				StartWorkflow: common.CreateHistoryStartWorkflowRequest(
					namespaceID.String(),
					startReq,
					nil,
					nil,
					time.Now().UTC(),
				),
			},
		}
	} else if updateReq := op.GetUpdateWorkflow(); updateReq != nil {
		if updateReq.Namespace != "" && updateReq.Namespace != namespaceName.String() {
			return nil, "", errMultiOpNamespaceMismatch
		}
		if err := wh.prepareUpdateWorkflowRequest(updateReq); err != nil {
			return nil, "", err
		}
		if updateReq.FirstExecutionRunId != "" {
			return nil, "", errMultiOpUpdateFirstExecutionRunId
		}
		if updateReq.WorkflowExecution.RunId != "" {
			return nil, "", errMultiOpUpdateExecutionRunId
		}

		workflowId = updateReq.WorkflowExecution.WorkflowId
		opReq = &historyservice.ExecuteMultiOperationRequest_Operation{
			Operation: &historyservice.ExecuteMultiOperationRequest_Operation_UpdateWorkflow{
				UpdateWorkflow: &historyservice.UpdateWorkflowExecutionRequest{
					NamespaceId: namespaceID.String(),
					Request:     updateReq,
				},
			},
		}
	} else {
		return nil, "", serviceerror.NewInternalf("unsupported operation: %T", op.Operation)
	}

	return opReq, workflowId, nil
}

func (wh *WorkflowHandler) convertToMultiOperationResponse(
	historyResp *historyservice.ExecuteMultiOperationResponse,
	namespaceName namespace.Name,
) (*workflowservice.ExecuteMultiOperationResponse, error) {
	resp := &workflowservice.ExecuteMultiOperationResponse{
		Responses: make([]*workflowservice.ExecuteMultiOperationResponse_Response, len(historyResp.Responses)),
	}
	for i, op := range historyResp.Responses {
		var opResp *workflowservice.ExecuteMultiOperationResponse_Response
		if historyStartResp := op.GetStartWorkflow(); historyStartResp != nil {
			startResp, err := wh.convertToStartWorkflowExecutionResponse(historyStartResp, namespaceName)
			if err != nil {
				return nil, err
			}
			opResp = &workflowservice.ExecuteMultiOperationResponse_Response{
				Response: &workflowservice.ExecuteMultiOperationResponse_Response_StartWorkflow{
					StartWorkflow: startResp,
				},
			}
		} else if histUpdateResp := op.GetUpdateWorkflow(); histUpdateResp != nil {
			opResp = &workflowservice.ExecuteMultiOperationResponse_Response{
				Response: &workflowservice.ExecuteMultiOperationResponse_Response_UpdateWorkflow{
					UpdateWorkflow: histUpdateResp.GetResponse(),
				},
			}
		} else {
			return nil, serviceerror.NewInternalf("unexpected operation result: %T", op.Response)
		}
		resp.Responses[i] = opResp
	}
	return resp, nil
}

// GetWorkflowExecutionHistory returns the history of specified workflow execution.  It fails with 'EntityNotExistError' if specified workflow
// execution in unknown to the service.
func (wh *WorkflowHandler) GetWorkflowExecutionHistory(ctx context.Context, request *workflowservice.GetWorkflowExecutionHistoryRequest) (_ *workflowservice.GetWorkflowExecutionHistoryResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.Execution); err != nil {
		return nil, err
	}

	if request.GetMaximumPageSize() <= 0 {
		request.MaximumPageSize = int32(wh.config.HistoryMaxPageSize(request.GetNamespace()))
	}

	enums.SetDefaultHistoryEventFilterType(&request.HistoryEventFilterType)

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	// force limit page size if exceed
	if request.GetMaximumPageSize() > primitives.GetHistoryMaxPageSize {
		wh.throttledLogger.Warn("GetHistory page size is larger than threshold",
			tag.WorkflowID(request.Execution.GetWorkflowId()),
			tag.WorkflowRunID(request.Execution.GetRunId()),
			tag.WorkflowNamespaceID(namespaceID.String()), tag.WorkflowSize(int64(request.GetMaximumPageSize())))
		request.MaximumPageSize = primitives.GetHistoryMaxPageSize
	}

	if !request.GetSkipArchival() {
		enableArchivalRead := wh.archivalMetadata.GetHistoryConfig().ReadEnabled()
		historyArchived := wh.historyArchived(ctx, request, namespaceID)
		if enableArchivalRead && historyArchived {
			return wh.getArchivedHistory(ctx, request, namespaceID)
		}
	}

	response, err := wh.historyClient.GetWorkflowExecutionHistory(ctx,
		&historyservice.GetWorkflowExecutionHistoryRequest{
			NamespaceId: namespaceID.String(),
			Request:     request,
		})
	if err != nil {
		return nil, err
	}

	isCloseEventOnly := request.HistoryEventFilterType == enumspb.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT
	err = api.ProcessInternalRawHistory(
		ctx,
		wh.saProvider,
		wh.saMapperProvider,
		response,
		wh.visibilityMgr,
		wh.versionChecker,
		namespace.Name(request.GetNamespace()),
		isCloseEventOnly,
	)
	if err != nil {
		return nil, err
	}
	return response.Response, nil
}

// GetWorkflowExecutionHistory returns the history of specified workflow execution.  It fails with 'EntityNotExistError' if specified workflow
// execution in unknown to the service.
func (wh *WorkflowHandler) GetWorkflowExecutionHistoryReverse(ctx context.Context, request *workflowservice.GetWorkflowExecutionHistoryReverseRequest) (_ *workflowservice.GetWorkflowExecutionHistoryReverseResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.Execution); err != nil {
		return nil, err
	}

	if request.GetMaximumPageSize() <= 0 {
		request.MaximumPageSize = int32(wh.config.HistoryMaxPageSize(request.GetNamespace()))
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	// force limit page size if exceed
	if request.GetMaximumPageSize() > primitives.GetHistoryMaxPageSize {
		wh.throttledLogger.Warn("GetHistory page size is larger than threshold",
			tag.WorkflowID(request.Execution.GetWorkflowId()),
			tag.WorkflowRunID(request.Execution.GetRunId()),
			tag.WorkflowNamespaceID(namespaceID.String()), tag.WorkflowSize(int64(request.GetMaximumPageSize())))
		request.MaximumPageSize = primitives.GetHistoryMaxPageSize
	}

	response, err := wh.historyClient.GetWorkflowExecutionHistoryReverse(ctx,
		&historyservice.GetWorkflowExecutionHistoryReverseRequest{
			NamespaceId: namespaceID.String(),
			Request:     request,
		})
	if err != nil {
		return nil, err
	}
	return response.Response, nil
}

// PollWorkflowTaskQueue is called by application worker to process WorkflowTask from a specific task queue.  A
// WorkflowTask is dispatched to callers for active workflow executions, with pending workflow tasks.
// Application is then expected to call 'RespondWorkflowTaskCompleted' API when it is done processing the WorkflowTask.
// It will also create a 'WorkflowTaskStarted' event in the history for that session before handing off WorkflowTask to
// application worker.
func (wh *WorkflowHandler) PollWorkflowTaskQueue(ctx context.Context, request *workflowservice.PollWorkflowTaskQueueRequest) (_ *workflowservice.PollWorkflowTaskQueueResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)
	if request == nil {
		return nil, errRequestNotSet
	}

	wh.logger.Debug("Received PollWorkflowTaskQueue")
	if err := common.ValidateLongPollContextTimeout(
		ctx,
		"PollWorkflowTaskQueue",
		wh.throttledLogger,
	); err != nil {
		return nil, err
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	if err := wh.validateVersioningInfo(request.Namespace, request.WorkerVersionCapabilities, request.DeploymentOptions, request.TaskQueue); err != nil {
		return nil, err
	}

	if request.TaskQueue.GetKind() == enumspb.TASK_QUEUE_KIND_UNSPECIFIED {
		wh.logger.Warn("Unspecified task queue kind",
			tag.WorkflowTaskQueueName(request.TaskQueue.GetName()),
			tag.WorkflowNamespace(namespace.Name(request.GetNamespace()).String()),
		)
	}

	if err := tqid.NormalizeAndValidate(request.TaskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	callTime := time.Now().UTC()

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}
	namespaceID := namespaceEntry.ID()

	wh.logger.Debug("Poll workflow task queue.", tag.WorkflowNamespace(namespaceEntry.Name().String()), tag.WorkflowNamespaceID(namespaceID.String()))
	if err := wh.checkBadBinary(namespaceEntry, request.GetBinaryChecksum()); err != nil {
		return nil, err
	}

	if contextNearDeadline(ctx, longPollTailRoom) {
		return &workflowservice.PollWorkflowTaskQueueResponse{}, nil
	}

	pollerID := uuid.NewString()
	childCtx := wh.registerOutstandingPollContext(ctx, pollerID, namespaceID.String())
	defer wh.unregisterOutstandingPollContext(pollerID, namespaceID.String())

	matchingResp, err := wh.matchingClient.PollWorkflowTaskQueue(childCtx, &matchingservice.PollWorkflowTaskQueueRequest{
		NamespaceId: namespaceID.String(),
		PollerId:    pollerID,
		PollRequest: request,
	})
	if err != nil {
		contextWasCanceled := wh.cancelOutstandingPoll(childCtx, namespaceID, enumspb.TASK_QUEUE_TYPE_WORKFLOW, request.TaskQueue, pollerID)
		if contextWasCanceled {
			// Clear error as we don't want to report context cancellation error to count against our SLA.
			// It doesn't matter what to return here, client has already gone. But (nil,nil) is invalid gogo return pair.
			return &workflowservice.PollWorkflowTaskQueueResponse{}, nil
		}

		// These errors are expected from some versioning situations. We should not log them, it'd be too noisy.
		var newerBuild *serviceerror.NewerBuildExists      // expected when versioned poller is superceded
		var failedPrecond *serviceerror.FailedPrecondition // expected when user data is disabled
		if errors.As(err, &newerBuild) || errors.As(err, &failedPrecond) {
			return nil, err
		}

		// For all other errors log an error and return it back to client.
		ctxTimeout := "not-set"
		ctxDeadline, ok := childCtx.Deadline()
		if ok {
			ctxTimeout = ctxDeadline.Sub(callTime).String()
		}
		wh.logger.Error("Unable to call matching.PollWorkflowTaskQueue.",
			tag.WorkflowTaskQueueName(request.GetTaskQueue().GetName()),
			tag.Timeout(ctxTimeout),
			tag.Error(err))
		return nil, err
	}

	// Handle history from matching response:
	// 1. If History field is set, use it directly (already processed by history service or matching)
	// 2. If RawHistory is set, use RawHistory (raw bytes path, needs SA processing)
	//
	// When history.sendRawHistoryBytesToMatchingService is enabled, matching service passes raw bytes
	// through to RawHistory field. The matching client auto-deserializes the repeated bytes into
	// a History message via gRPC wire compatibility.
	history := matchingResp.History
	if matchingResp.RawHistory != nil {
		history = matchingResp.RawHistory
		// Process search attributes for raw history since it bypasses the normal processing path.
		if err := api.ProcessOutgoingSearchAttributes(
			wh.saProvider,
			wh.saMapperProvider,
			history.GetEvents(),
			namespaceEntry.Name(),
			wh.visibilityMgr,
		); err != nil {
			return nil, err
		}
	}

	return &workflowservice.PollWorkflowTaskQueueResponse{
		TaskToken:                  matchingResp.TaskToken,
		WorkflowExecution:          matchingResp.WorkflowExecution,
		WorkflowType:               matchingResp.WorkflowType,
		PreviousStartedEventId:     matchingResp.PreviousStartedEventId,
		StartedEventId:             matchingResp.StartedEventId,
		Query:                      matchingResp.Query,
		BacklogCountHint:           matchingResp.BacklogCountHint,
		Attempt:                    matchingResp.Attempt,
		History:                    history,
		NextPageToken:              matchingResp.NextPageToken,
		WorkflowExecutionTaskQueue: matchingResp.WorkflowExecutionTaskQueue,
		ScheduledTime:              matchingResp.ScheduledTime,
		StartedTime:                matchingResp.StartedTime,
		Queries:                    matchingResp.Queries,
		Messages:                   matchingResp.Messages,
		PollerScalingDecision:      matchingResp.PollerScalingDecision,
	}, nil
}

func contextNearDeadline(ctx context.Context, tailroom time.Duration) bool {
	if ctxDeadline, ok := ctx.Deadline(); ok {
		return time.Now().Add(tailroom).After(ctxDeadline)
	}
	return false
}

// RespondWorkflowTaskCompleted is called by application worker to complete a WorkflowTask handed as a result of
// 'PollWorkflowTaskQueue' API call.  Completing a WorkflowTask will result in new events for the workflow execution and
// potentially new ActivityTask being created for corresponding commands.  It will also create a WorkflowTaskCompleted
// event in the history for that session.  Use the 'taskToken' provided as response of PollWorkflowTaskQueue API call
// for completing the WorkflowTask.
// The response could contain a new workflow task if there is one or if the request asking for one.
func (wh *WorkflowHandler) RespondWorkflowTaskCompleted(
	ctx context.Context,
	request *workflowservice.RespondWorkflowTaskCompletedRequest,
) (_ *workflowservice.RespondWorkflowTaskCompletedResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	if err := wh.validateVersioningInfo(
		request.Namespace,
		request.WorkerVersionStamp,
		request.DeploymentOptions,
		request.StickyAttributes.GetWorkerTaskQueue(),
	); err != nil {
		return nil, err
	}

	wh.overrides.DisableEagerActivityDispatchForBuggyClients(ctx, request)

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	response, err := wh.historyClient.RespondWorkflowTaskCompleted(ctx,
		&historyservice.RespondWorkflowTaskCompletedRequest{
			NamespaceId:     namespaceID.String(),
			CompleteRequest: request,
		},
	)
	if err != nil {
		return nil, err
	}

	return &workflowservice.RespondWorkflowTaskCompletedResponse{
		WorkflowTask:        response.NewWorkflowTask,
		ActivityTasks:       response.ActivityTasks,
		ResetHistoryEventId: response.ResetHistoryEventId,
	}, nil
}

// RespondWorkflowTaskFailed is called by application worker to indicate failure.  This results in
// WorkflowTaskFailedEvent written to the history and a new WorkflowTask created.  This API can be used by client to
// either clear sticky taskqueue or report any panics during WorkflowTask processing.  Temporal will only append first
// WorkflowTaskFailed event to the history of workflow execution for consecutive failures.
func (wh *WorkflowHandler) RespondWorkflowTaskFailed(
	ctx context.Context,
	request *workflowservice.RespondWorkflowTaskFailedRequest,
) (_ *workflowservice.RespondWorkflowTaskFailedResponse, retError error) {

	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	taskToken, err := wh.tokenSerializer.Deserialize(request.TaskToken)
	if err != nil {
		return nil, errDeserializingToken
	}
	namespaceId := namespace.ID(taskToken.GetNamespaceId())
	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceId)
	if err != nil {
		return nil, err
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	if err := common.CheckEventBlobSizeLimit(
		request.GetFailure().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceId.String(),
		taskToken.GetWorkflowId(),
		taskToken.GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondWorkflowTaskFailed",
	); err != nil {
		serverFailure := failure.NewServerFailure(common.FailureReasonFailureExceedsLimit, true)
		serverFailure.Cause = failure.Truncate(request.Failure, sizeLimitWarn)
		request.Failure = serverFailure
	}

	if request.GetCause() == enumspb.WORKFLOW_TASK_FAILED_CAUSE_NON_DETERMINISTIC_ERROR {
		wh.logger.Info("Non-Deterministic Error",
			tag.WorkflowNamespaceID(taskToken.GetNamespaceId()),
			tag.WorkflowID(taskToken.GetWorkflowId()),
			tag.WorkflowRunID(taskToken.GetRunId()),
		)
		metrics.ServiceErrNonDeterministicCounter.With(wh.metricsScope(ctx)).Record(1)
	}

	_, err = wh.historyClient.RespondWorkflowTaskFailed(ctx, &historyservice.RespondWorkflowTaskFailedRequest{
		NamespaceId:   namespaceId.String(),
		FailedRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.RespondWorkflowTaskFailedResponse{}, nil
}

// PollActivityTaskQueue is called by application worker to process ActivityTask from a specific task queue.  ActivityTask
// is dispatched to callers whenever a ScheduleTask command is made for a workflow execution.
// Application is expected to call 'RespondActivityTaskCompleted' or 'RespondActivityTaskFailed' once it is done
// processing the task.
// Application also needs to call 'RecordActivityTaskHeartbeat' API within 'heartbeatTimeoutSeconds' interval to
// prevent the task from getting timed out.  An event 'ActivityTaskStarted' event is also written to workflow execution
// history before the ActivityTask is dispatched to application worker.
func (wh *WorkflowHandler) PollActivityTaskQueue(ctx context.Context, request *workflowservice.PollActivityTaskQueueRequest) (_ *workflowservice.PollActivityTaskQueueResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	callTime := time.Now().UTC()

	if request == nil {
		return nil, errRequestNotSet
	}

	wh.logger.Debug("Received PollActivityTaskQueue")
	if err := common.ValidateLongPollContextTimeout(
		ctx,
		"PollActivityTaskQueue",
		wh.throttledLogger,
	); err != nil {
		return nil, err
	}

	namespaceName := namespace.Name(request.GetNamespace())
	if err := tqid.NormalizeAndValidate(request.TaskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}
	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	if err := wh.validateVersioningInfo(request.Namespace, request.WorkerVersionCapabilities, request.DeploymentOptions, request.TaskQueue); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if contextNearDeadline(ctx, longPollTailRoom) {
		return &workflowservice.PollActivityTaskQueueResponse{}, nil
	}

	pollerID := uuid.NewString()
	childCtx := wh.registerOutstandingPollContext(ctx, pollerID, namespaceID.String())
	defer wh.unregisterOutstandingPollContext(pollerID, namespaceID.String())
	matchingResponse, err := wh.matchingClient.PollActivityTaskQueue(childCtx, &matchingservice.PollActivityTaskQueueRequest{
		NamespaceId: namespaceID.String(),
		PollerId:    pollerID,
		PollRequest: request,
	})
	if err != nil {
		contextWasCanceled := wh.cancelOutstandingPoll(childCtx, namespaceID, enumspb.TASK_QUEUE_TYPE_ACTIVITY, request.TaskQueue, pollerID)
		if contextWasCanceled {
			// Clear error as we don't want to report context cancellation error to count against our SLA.
			// It doesn't matter what to return here, client has already gone. But (nil,nil) is invalid gogo return pair.
			return &workflowservice.PollActivityTaskQueueResponse{}, nil
		}

		// These errors are expected from some versioning situations. We should not log them, it'd be too noisy.
		var newerBuild *serviceerror.NewerBuildExists      // expected when versioned poller is superceded
		var failedPrecond *serviceerror.FailedPrecondition // expected when user data is disabled
		if errors.As(err, &newerBuild) || errors.As(err, &failedPrecond) {
			return nil, err
		}

		// For all other errors log an error and return it back to client.
		ctxTimeout := "not-set"
		ctxDeadline, ok := childCtx.Deadline()
		if ok {
			ctxTimeout = ctxDeadline.Sub(callTime).String()
		}
		wh.logger.Error("Unable to call matching.PollActivityTaskQueue.",
			tag.WorkflowTaskQueueName(request.GetTaskQueue().GetName()),
			tag.Timeout(ctxTimeout),
			tag.Error(err))

		return nil, err
	}

	return &workflowservice.PollActivityTaskQueueResponse{
		TaskToken:                   matchingResponse.TaskToken,
		WorkflowExecution:           matchingResponse.WorkflowExecution,
		ActivityId:                  matchingResponse.ActivityId,
		ActivityType:                matchingResponse.ActivityType,
		ActivityRunId:               matchingResponse.ActivityRunId,
		Input:                       matchingResponse.Input,
		ScheduledTime:               matchingResponse.ScheduledTime,
		ScheduleToCloseTimeout:      matchingResponse.ScheduleToCloseTimeout,
		StartedTime:                 matchingResponse.StartedTime,
		StartToCloseTimeout:         matchingResponse.StartToCloseTimeout,
		HeartbeatTimeout:            matchingResponse.HeartbeatTimeout,
		Attempt:                     matchingResponse.Attempt,
		CurrentAttemptScheduledTime: matchingResponse.CurrentAttemptScheduledTime,
		HeartbeatDetails:            matchingResponse.HeartbeatDetails,
		WorkflowType:                matchingResponse.WorkflowType,
		WorkflowNamespace:           matchingResponse.WorkflowNamespace,
		Header:                      matchingResponse.Header,
		PollerScalingDecision:       matchingResponse.PollerScalingDecision,
		Priority:                    matchingResponse.Priority,
		RetryPolicy:                 matchingResponse.RetryPolicy,
	}, nil
}

// RecordActivityTaskHeartbeat is called by application worker while it is processing an ActivityTask.  If worker fails
// to heartbeat within 'heartbeatTimeoutSeconds' interval for the ActivityTask, then it will be marked as timedout and
// 'ActivityTaskTimedOut' event will be written to the workflow history.  Calling 'RecordActivityTaskHeartbeat' will
// fail with 'EntityNotExistsError' in such situations.  Use the 'taskToken' provided as response of
// PollActivityTaskQueue API call for heartbeating.
func (wh *WorkflowHandler) RecordActivityTaskHeartbeat(ctx context.Context, request *workflowservice.RecordActivityTaskHeartbeatRequest) (_ *workflowservice.RecordActivityTaskHeartbeatResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	wh.logger.Debug("Received RecordActivityTaskHeartbeat")
	taskToken, err := wh.tokenSerializer.Deserialize(request.TaskToken)
	if err != nil {
		return nil, errDeserializingToken
	}
	namespaceId := namespace.ID(taskToken.GetNamespaceId())
	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceId)
	if err != nil {
		return nil, err
	}
	namespaceName := namespaceEntry.Name().String()

	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
		return nil, activity.ErrStandaloneActivityDisabled
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	if err := common.CheckEventBlobSizeLimit(
		request.GetDetails().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceId.String(),
		taskToken.GetWorkflowId(),
		taskToken.GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RecordActivityTaskHeartbeat",
	); err != nil {
		// heartbeat details exceed size limit, we would fail the activity immediately with explicit error reason
		failRequest := &workflowservice.RespondActivityTaskFailedRequest{
			TaskToken: request.TaskToken,
			Failure:   failure.NewServerFailure(common.FailureReasonHeartbeatExceedsLimit, true),
			Identity:  request.Identity,
		}
		_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
			NamespaceId:   namespaceId.String(),
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, err
		}
		return &workflowservice.RecordActivityTaskHeartbeatResponse{CancelRequested: true}, nil
	}

	resp, err := wh.historyClient.RecordActivityTaskHeartbeat(ctx, &historyservice.RecordActivityTaskHeartbeatRequest{
		NamespaceId:      namespaceId.String(),
		HeartbeatRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.RecordActivityTaskHeartbeatResponse{
		CancelRequested: resp.GetCancelRequested(),
		ActivityPaused:  resp.GetActivityPaused(),
		ActivityReset:   resp.GetActivityReset(),
	}, nil
}

// RecordActivityTaskHeartbeatById is called by application worker while it is processing an ActivityTask.  If worker fails
// to heartbeat within 'heartbeatTimeoutSeconds' interval for the ActivityTask, then it will be marked as timedout and
// 'ActivityTaskTimedOut' event will be written to the workflow history.  Calling 'RecordActivityTaskHeartbeatById' will
// fail with 'EntityNotExistsError' in such situations.  Instead of using 'taskToken' like in RecordActivityTaskHeartbeat,
// use Namespace, WorkflowID and ActivityID
func (wh *WorkflowHandler) RecordActivityTaskHeartbeatById(ctx context.Context, request *workflowservice.RecordActivityTaskHeartbeatByIdRequest) (_ *workflowservice.RecordActivityTaskHeartbeatByIdResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	wh.logger.Debug("Received RecordActivityTaskHeartbeatById")
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}
	workflowID := request.GetWorkflowId()
	runID := request.GetRunId() // runID is optional so can be empty
	activityID := request.GetActivityId()

	if activityID == "" {
		return nil, errActivityIDNotSet
	}

	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
	// Else this should be a validation error.
	var componentRef []byte
	var attempt int32 = 1
	if workflowID == "" {
		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
			return nil, errWorkflowIDNotSet
		}

		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
			NamespaceID: namespaceID.String(),
			BusinessID:  activityID,
			RunID:       runID,
		})

		componentRef, err = ref.Serialize(wh.registry)
		if err != nil {
			return nil, err
		}
		attempt = activity.ByIDTokenAttempt
	}

	taskToken := tasktoken.NewActivityTaskToken(
		namespaceID.String(),
		workflowID,
		runID,
		common.EmptyEventID,
		activityID,
		"",
		attempt,
		nil,
		common.EmptyVersion,
		common.EmptyVersion,
		componentRef,
	)
	token, err := wh.tokenSerializer.Serialize(taskToken)
	if err != nil {
		return nil, err
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	if err := common.CheckEventBlobSizeLimit(
		request.GetDetails().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		taskToken.GetWorkflowId(),
		taskToken.GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RecordActivityTaskHeartbeatById",
	); err != nil {
		// heartbeat details exceed size limit, we would fail the activity immediately with explicit error reason
		failRequest := &workflowservice.RespondActivityTaskFailedRequest{
			TaskToken: token,
			Failure:   failure.NewServerFailure(common.FailureReasonHeartbeatExceedsLimit, true),
			Identity:  request.Identity,
		}
		_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
			NamespaceId:   namespaceID.String(),
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, err
		}
		return &workflowservice.RecordActivityTaskHeartbeatByIdResponse{CancelRequested: true}, nil
	}

	req := &workflowservice.RecordActivityTaskHeartbeatRequest{
		TaskToken: token,
		Details:   request.Details,
		Identity:  request.Identity,
	}

	resp, err := wh.historyClient.RecordActivityTaskHeartbeat(ctx, &historyservice.RecordActivityTaskHeartbeatRequest{
		NamespaceId:      namespaceID.String(),
		HeartbeatRequest: req,
	})
	if err != nil {
		return nil, err
	}
	return &workflowservice.RecordActivityTaskHeartbeatByIdResponse{
		CancelRequested: resp.GetCancelRequested(),
		ActivityPaused:  resp.GetActivityPaused(),
		ActivityReset:   resp.GetActivityReset(),
	}, nil
}

// RespondActivityTaskCompleted is called by application worker when it is done processing an ActivityTask.  It will
// result in a new 'ActivityTaskCompleted' event being written to the workflow history and a new WorkflowTask
// created for the workflow so new commands could be made.  Use the 'taskToken' provided as response of
// PollActivityTaskQueue API call for completion. It fails with 'NotFoundFailure' if the taskToken is not valid
// anymore due to activity timeout.
func (wh *WorkflowHandler) RespondActivityTaskCompleted(
	ctx context.Context,
	request *workflowservice.RespondActivityTaskCompletedRequest,
) (_ *workflowservice.RespondActivityTaskCompletedResponse, retError error) {

	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}
	taskToken, err := wh.tokenSerializer.Deserialize(request.TaskToken)
	if err != nil {
		return nil, errDeserializingToken
	}
	namespaceId := namespace.ID(taskToken.GetNamespaceId())
	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceId)
	if err != nil {
		return nil, err
	}
	namespaceName := namespaceEntry.Name().String()

	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
		return nil, activity.ErrStandaloneActivityDisabled
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)

	if err := common.CheckEventBlobSizeLimit(
		request.GetResult().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceId.String(),
		taskToken.GetWorkflowId(),
		taskToken.GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondActivityTaskCompleted",
	); err != nil {
		// result exceeds blob size limit, we would record it as failure
		failRequest := &workflowservice.RespondActivityTaskFailedRequest{
			TaskToken: request.TaskToken,
			Failure:   failure.NewServerFailure(common.FailureReasonCompleteResultExceedsLimit, true),
			Identity:  request.Identity,
		}
		_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
			NamespaceId:   namespaceId.String(),
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, err
		}
	} else {
		_, err = wh.historyClient.RespondActivityTaskCompleted(ctx, &historyservice.RespondActivityTaskCompletedRequest{
			NamespaceId:     namespaceId.String(),
			CompleteRequest: request,
		})
		if err != nil {
			return nil, err
		}
	}

	return &workflowservice.RespondActivityTaskCompletedResponse{}, nil
}

// RespondActivityTaskCompletedById is called by application worker when it is done processing an ActivityTask.
// It will result in a new 'ActivityTaskCompleted' event being written to the workflow history and a new WorkflowTask
// created for the workflow so new commands could be made.  Similar to RespondActivityTaskCompleted but use Namespace,
// WorkflowId and ActivityId instead of 'taskToken' for completion. It fails with 'NotFoundFailure'
// if the these Ids are not valid anymore due to activity timeout.
func (wh *WorkflowHandler) RespondActivityTaskCompletedById(ctx context.Context, request *workflowservice.RespondActivityTaskCompletedByIdRequest) (_ *workflowservice.RespondActivityTaskCompletedByIdResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}
	workflowID := request.GetWorkflowId()
	runID := request.GetRunId() // runID is optional so can be empty
	activityID := request.GetActivityId()

	if activityID == "" {
		return nil, errActivityIDNotSet
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
	// Else this should be a validation error.
	var componentRef []byte
	var attempt int32 = 1
	if workflowID == "" {
		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
			return nil, errWorkflowIDNotSet
		}

		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
			NamespaceID: namespaceID.String(),
			BusinessID:  activityID,
			RunID:       runID,
		})

		componentRef, err = ref.Serialize(wh.registry)
		if err != nil {
			return nil, err
		}
		attempt = activity.ByIDTokenAttempt
	}

	taskToken := tasktoken.NewActivityTaskToken(
		namespaceID.String(),
		workflowID,
		runID,
		common.EmptyEventID,
		activityID,
		"",
		attempt,
		nil,
		common.EmptyVersion,
		common.EmptyVersion,
		componentRef,
	)
	token, err := wh.tokenSerializer.Serialize(taskToken)
	if err != nil {
		return nil, err
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	if err := common.CheckEventBlobSizeLimit(
		request.GetResult().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		taskToken.GetWorkflowId(),
		runID,
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondActivityTaskCompletedById",
	); err != nil {
		// result exceeds blob size limit, we would record it as failure
		failRequest := &workflowservice.RespondActivityTaskFailedRequest{
			TaskToken: token,
			Failure:   failure.NewServerFailure(common.FailureReasonCompleteResultExceedsLimit, true),
			Identity:  request.Identity,
		}
		_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
			NamespaceId:   namespaceID.String(),
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, err
		}
	} else {
		req := &workflowservice.RespondActivityTaskCompletedRequest{
			TaskToken: token,
			Result:    request.Result,
			Identity:  request.Identity,
		}

		_, err = wh.historyClient.RespondActivityTaskCompleted(ctx, &historyservice.RespondActivityTaskCompletedRequest{
			NamespaceId:     namespaceID.String(),
			CompleteRequest: req,
		})
		if err != nil {
			return nil, err
		}
	}

	return &workflowservice.RespondActivityTaskCompletedByIdResponse{}, nil
}

// RespondActivityTaskFailed is called by application worker when it is done processing an ActivityTask.  It will
// result in a new 'ActivityTaskFailed' event being written to the workflow history and a new WorkflowTask
// created for the workflow instance so new commands could be made.  Use the 'taskToken' provided as response of
// PollActivityTaskQueue API call for completion. It fails with 'EntityNotExistsError' if the taskToken is not valid
// anymore due to activity timeout.
func (wh *WorkflowHandler) RespondActivityTaskFailed(
	ctx context.Context,
	request *workflowservice.RespondActivityTaskFailedRequest,
) (_ *workflowservice.RespondActivityTaskFailedResponse, retError error) {

	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	taskToken, err := wh.tokenSerializer.Deserialize(request.TaskToken)
	if err != nil {
		return nil, errDeserializingToken
	}
	namespaceID := namespace.ID(taskToken.GetNamespaceId())
	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}
	namespaceName := namespaceEntry.Name().String()

	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
		return nil, activity.ErrStandaloneActivityDisabled
	}

	if request.GetFailure() != nil && request.GetFailure().GetApplicationFailureInfo() == nil {
		return nil, errFailureMustHaveApplicationFailureInfo
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)

	response := workflowservice.RespondActivityTaskFailedResponse{}

	if request.GetLastHeartbeatDetails() != nil {
		if err := common.CheckEventBlobSizeLimit(
			request.GetLastHeartbeatDetails().Size(),
			sizeLimitWarn,
			sizeLimitError,
			namespaceID.String(),
			taskToken.GetWorkflowId(),
			taskToken.GetRunId(),
			wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
			wh.throttledLogger,
			"RespondActivityTaskFailed",
		); err != nil {
			// heartbeat details exceed size limit, we would fail the activity immediately with explicit error reason
			response.Failures = append(response.Failures, failure.NewServerFailure(common.FailureReasonHeartbeatExceedsLimit, true))

			// do not send heartbeat to history service
			request.LastHeartbeatDetails = nil
		}
	}

	if err := common.CheckEventBlobSizeLimit(
		request.GetFailure().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		taskToken.GetWorkflowId(),
		taskToken.GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondActivityTaskFailed",
	); err != nil {
		serverFailure := failure.NewServerFailure(common.FailureReasonFailureExceedsLimit, true)
		serverFailure.Cause = failure.Truncate(request.Failure, sizeLimitWarn)
		request.Failure = serverFailure

		response.Failures = append(response.Failures, serverFailure)
	}

	_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
		NamespaceId:   namespaceID.String(),
		FailedRequest: request,
	})
	if err != nil {
		return nil, err
	}
	return &response, nil
}

// RespondActivityTaskFailedById is called by application worker when it is done processing an ActivityTask.
// It will result in a new 'ActivityTaskFailed' event being written to the workflow history and a new WorkflowTask
// created for the workflow instance so new commands could be made.  Similar to RespondActivityTaskFailed but use
// Namespace, WorkflowID and ActivityID instead of 'taskToken' for completion. It fails with 'EntityNotExistsError'
// if the these IDs are not valid anymore due to activity timeout.
func (wh *WorkflowHandler) RespondActivityTaskFailedById(ctx context.Context, request *workflowservice.RespondActivityTaskFailedByIdRequest) (_ *workflowservice.RespondActivityTaskFailedByIdResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}
	workflowID := request.GetWorkflowId()
	runID := request.GetRunId() // runID is optional so can be empty
	activityID := request.GetActivityId()

	if activityID == "" {
		return nil, errActivityIDNotSet
	}
	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
	// Else this should be a validation error.
	var componentRef []byte
	var attempt int32 = 1
	if workflowID == "" {
		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
			return nil, errWorkflowIDNotSet
		}

		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
			NamespaceID: namespaceID.String(),
			BusinessID:  activityID,
			RunID:       runID,
		})

		componentRef, err = ref.Serialize(wh.registry)
		if err != nil {
			return nil, err
		}
		attempt = activity.ByIDTokenAttempt
	}

	taskToken := tasktoken.NewActivityTaskToken(
		namespaceID.String(),
		workflowID,
		runID,
		common.EmptyEventID,
		activityID,
		"",
		attempt,
		nil,
		common.EmptyVersion,
		common.EmptyVersion,
		componentRef,
	)
	token, err := wh.tokenSerializer.Serialize(taskToken)
	if err != nil {
		return nil, err
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	response := workflowservice.RespondActivityTaskFailedByIdResponse{}

	if request.GetLastHeartbeatDetails() != nil {
		if err := common.CheckEventBlobSizeLimit(
			request.GetLastHeartbeatDetails().Size(),
			sizeLimitWarn,
			sizeLimitError,
			namespaceID.String(),
			taskToken.GetWorkflowId(),
			runID,
			wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
			wh.throttledLogger,
			"RespondActivityTaskFailedById",
		); err != nil {
			// heartbeat details exceed size limit, we would fail the activity immediately with explicit error reason
			response.Failures = append(response.Failures, failure.NewServerFailure(common.FailureReasonHeartbeatExceedsLimit, true))

			// do not send heartbeat to history service
			request.LastHeartbeatDetails = nil
		}
	}

	if err := common.CheckEventBlobSizeLimit(
		request.GetFailure().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		taskToken.GetWorkflowId(),
		runID,
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondActivityTaskFailedById",
	); err != nil {
		serverFailure := failure.NewServerFailure(common.FailureReasonFailureExceedsLimit, true)
		serverFailure.Cause = failure.Truncate(request.Failure, sizeLimitWarn)
		request.Failure = serverFailure

		response.Failures = append(response.Failures, serverFailure)
	}

	req := &workflowservice.RespondActivityTaskFailedRequest{
		TaskToken: token,
		Failure:   request.GetFailure(),
		Identity:  request.Identity,
	}

	_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
		NamespaceId:   namespaceID.String(),
		FailedRequest: req,
	})
	if err != nil {
		return nil, err
	}
	return &response, nil
}

// RespondActivityTaskCanceled is called by application worker when it is successfully canceled an ActivityTask.  It will
// result in a new 'ActivityTaskCanceled' event being written to the workflow history and a new WorkflowTask
// created for the workflow instance so new commands could be made.  Use the 'taskToken' provided as response of
// PollActivityTaskQueue API call for completion. It fails with 'EntityNotExistsError' if the taskToken is not valid
// anymore due to activity timeout.
func (wh *WorkflowHandler) RespondActivityTaskCanceled(ctx context.Context, request *workflowservice.RespondActivityTaskCanceledRequest) (_ *workflowservice.RespondActivityTaskCanceledResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	taskToken, err := wh.tokenSerializer.Deserialize(request.TaskToken)
	if err != nil {
		return nil, errDeserializingToken
	}
	namespaceID := namespace.ID(taskToken.GetNamespaceId())
	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}
	namespaceName := namespaceEntry.Name().String()

	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
		return nil, activity.ErrStandaloneActivityDisabled
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)

	if err := common.CheckEventBlobSizeLimit(
		request.GetDetails().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		taskToken.GetWorkflowId(),
		taskToken.GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondActivityTaskCanceled",
	); err != nil {
		// details exceeds blob size limit, we would record it as failure
		failRequest := &workflowservice.RespondActivityTaskFailedRequest{
			TaskToken: request.TaskToken,
			Failure:   failure.NewServerFailure(common.FailureReasonCancelDetailsExceedsLimit, true),
			Identity:  request.Identity,
		}
		_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
			NamespaceId:   taskToken.GetNamespaceId(),
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, err
		}
	} else {
		_, err = wh.historyClient.RespondActivityTaskCanceled(ctx, &historyservice.RespondActivityTaskCanceledRequest{
			NamespaceId:   taskToken.GetNamespaceId(),
			CancelRequest: request,
		})
		if err != nil {
			return nil, err
		}
	}

	return &workflowservice.RespondActivityTaskCanceledResponse{}, nil
}

// RespondActivityTaskCanceledById is called by application worker when it is successfully canceled an ActivityTask.
// It will result in a new 'ActivityTaskCanceled' event being written to the workflow history and a new WorkflowTask
// created for the workflow instance so new commands could be made.  Similar to RespondActivityTaskCanceled but use
// Namespace, WorkflowID and ActivityID instead of 'taskToken' for completion. It fails with 'EntityNotExistsError'
// if the these IDs are not valid anymore due to activity timeout.
func (wh *WorkflowHandler) RespondActivityTaskCanceledById(ctx context.Context, request *workflowservice.RespondActivityTaskCanceledByIdRequest) (_ *workflowservice.RespondActivityTaskCanceledByIdResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}
	workflowID := request.GetWorkflowId()
	runID := request.GetRunId() // runID is optional so can be empty
	activityID := request.GetActivityId()

	if activityID == "" {
		return nil, errActivityIDNotSet
	}
	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
	// Else this should be a validation error.
	var componentRef []byte
	var attempt int32 = 1
	if workflowID == "" {
		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
			return nil, errWorkflowIDNotSet
		}

		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
			NamespaceID: namespaceID.String(),
			BusinessID:  activityID,
			RunID:       runID,
		})

		componentRef, err = ref.Serialize(wh.registry)
		if err != nil {
			return nil, err
		}
		attempt = activity.ByIDTokenAttempt
	}

	taskToken := tasktoken.NewActivityTaskToken(
		namespaceID.String(),
		workflowID,
		runID,
		common.EmptyEventID,
		activityID,
		"",
		attempt,
		nil,
		common.EmptyVersion,
		common.EmptyVersion,
		componentRef,
	)
	token, err := wh.tokenSerializer.Serialize(taskToken)
	if err != nil {
		return nil, err
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	if err := common.CheckEventBlobSizeLimit(
		request.GetDetails().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		taskToken.GetWorkflowId(),
		runID,
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondActivityTaskCanceledById",
	); err != nil {
		// details exceeds blob size limit, we would record it as failure
		failRequest := &workflowservice.RespondActivityTaskFailedRequest{
			TaskToken: token,
			Failure:   failure.NewServerFailure(common.FailureReasonCancelDetailsExceedsLimit, true),
			Identity:  request.Identity,
		}
		_, err = wh.historyClient.RespondActivityTaskFailed(ctx, &historyservice.RespondActivityTaskFailedRequest{
			NamespaceId:   namespaceID.String(),
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, err
		}
	} else {
		req := &workflowservice.RespondActivityTaskCanceledRequest{
			TaskToken: token,
			Details:   request.Details,
			Identity:  request.Identity,
		}

		_, err = wh.historyClient.RespondActivityTaskCanceled(ctx, &historyservice.RespondActivityTaskCanceledRequest{
			NamespaceId:   namespaceID.String(),
			CancelRequest: req,
		})
		if err != nil {
			return nil, err
		}
	}

	return &workflowservice.RespondActivityTaskCanceledByIdResponse{}, nil
}

// RequestCancelWorkflowExecution is called by application worker when it wants to request cancellation of a workflow instance.
// It will result in a new 'WorkflowExecutionCancelRequested' event being written to the workflow history and a new WorkflowTask
// created for the workflow instance so new commands could be made. It returns success if requested workflow already closed.
// It fails with 'NotFound' if the requested workflow doesn't exist.
func (wh *WorkflowHandler) RequestCancelWorkflowExecution(ctx context.Context, request *workflowservice.RequestCancelWorkflowExecutionRequest) (_ *workflowservice.RequestCancelWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.WorkflowExecution); err != nil {
		return nil, err
	}

	if err := wh.validator.ValidateLinks(request.GetNamespace(), request.GetLinks()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.RequestCancelWorkflowExecution(ctx, &historyservice.RequestCancelWorkflowExecutionRequest{
		NamespaceId:   namespaceID.String(),
		CancelRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.RequestCancelWorkflowExecutionResponse{}, nil
}

// SignalWorkflowExecution is used to send a signal event to running workflow execution. This results in
// a WorkflowExecutionSignaled event recorded in the history and a workflow task being created for the execution.
func (wh *WorkflowHandler) SignalWorkflowExecution(ctx context.Context, request *workflowservice.SignalWorkflowExecutionRequest) (_ *workflowservice.SignalWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.WorkflowExecution); err != nil {
		return nil, err
	}

	if request.GetSignalName() == "" {
		return nil, errSignalNameNotSet
	}

	if len(request.GetSignalName()) > wh.config.MaxIDLengthLimit() {
		return nil, errSignalNameTooLong
	}

	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
		return nil, errRequestIDTooLong
	}

	if err := wh.validator.ValidateLinks(request.GetNamespace(), request.GetLinks()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(request.GetNamespace())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(request.GetNamespace())
	if err := common.CheckEventBlobSizeLimit(
		request.GetInput().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		request.GetWorkflowExecution().GetWorkflowId(),
		request.GetWorkflowExecution().GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"SignalWorkflowExecution",
	); err != nil {
		return nil, err
	}

	resp, err := wh.historyClient.SignalWorkflowExecution(ctx, &historyservice.SignalWorkflowExecutionRequest{
		NamespaceId:   namespaceID.String(),
		SignalRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.SignalWorkflowExecutionResponse{
		Link: resp.GetLink(),
	}, nil
}

// SignalWithStartWorkflowExecution is used to ensure sending signal to a workflow.
// If the workflow is running, this results in WorkflowExecutionSignaled event being recorded in the history
// and a workflow task being created for the execution.
// If the workflow is not running or not found, this results in WorkflowExecutionStarted and WorkflowExecutionSignaled
// events being recorded in history, and a workflow task being created for the execution
func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(ctx context.Context, request *workflowservice.SignalWithStartWorkflowExecutionRequest) (_ *workflowservice.SignalWithStartWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := backoff.ValidateSchedule(request.GetCronSchedule()); err != nil {
		return nil, err
	}

	// The validator will modify the request proto if there are any search attributes to validate.
	// To avoid modifying the caller's object, which may be reused on retry, we clone the request here if needed.
	if len(request.GetSearchAttributes().GetIndexedFields()) > 0 {
		request = common.CloneProto(request)
	}

	if err := wh.validator.ValidateSignalWithStartRequest(request); err != nil {
		return nil, err
	}

	if err := wh.validator.ValidateLinks(request.GetNamespace(), request.GetLinks()); err != nil {
		return nil, err
	}

	namespaceName := namespace.Name(request.GetNamespace())
	if err := wh.validateTimeSkippingConfig(request.GetTimeSkippingConfig(), namespaceName); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	resp, err := wh.historyClient.SignalWithStartWorkflowExecution(ctx, &historyservice.SignalWithStartWorkflowExecutionRequest{
		NamespaceId:            namespaceID.String(),
		SignalWithStartRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.SignalWithStartWorkflowExecutionResponse{
		RunId:      resp.GetRunId(),
		Started:    resp.Started,
		SignalLink: resp.GetSignalLink(),
	}, nil
}

// ResetWorkflowExecution reset an existing workflow execution to WorkflowTaskCompleted event(exclusive).
// And it will immediately terminating the current execution instance.
func (wh *WorkflowHandler) ResetWorkflowExecution(ctx context.Context, request *workflowservice.ResetWorkflowExecutionRequest) (_ *workflowservice.ResetWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetRequestId() == "" {
		return nil, errRequestIDNotSet
	}
	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
		return nil, errRequestIDTooLong
	}

	if err := validateExecution(request.WorkflowExecution); err != nil {
		return nil, err
	}

	enums.SetDefaultResetReapplyType(&request.ResetReapplyType)
	if _, validType := enumspb.ResetReapplyType_name[int32(request.GetResetReapplyType())]; !validType {
		return nil, serviceerror.NewInternalf("unknown reset reapply type: %v", request.GetResetReapplyType())
	}

	for _, postOp := range request.GetPostResetOperations() {
		if updateOpts := postOp.GetUpdateWorkflowOptions(); updateOpts != nil {
			if err := wh.validateTimeSkippingConfig(
				updateOpts.GetWorkflowExecutionOptions().GetTimeSkippingConfig(),
				namespace.Name(request.GetNamespace()),
			); err != nil {
				return nil, err
			}
		}
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	resp, err := wh.historyClient.ResetWorkflowExecution(ctx, &historyservice.ResetWorkflowExecutionRequest{
		NamespaceId:  namespaceID.String(),
		ResetRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.ResetWorkflowExecutionResponse{RunId: resp.GetRunId()}, nil
}

// TerminateWorkflowExecution terminates an existing workflow execution by recording WorkflowExecutionTerminated event
// in the history and immediately terminating the execution instance.
func (wh *WorkflowHandler) TerminateWorkflowExecution(ctx context.Context, request *workflowservice.TerminateWorkflowExecutionRequest) (_ *workflowservice.TerminateWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.WorkflowExecution); err != nil {
		return nil, err
	}

	if err := wh.validator.ValidateLinks(request.GetNamespace(), request.GetLinks()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	// Set default values for user-initiated terminate requests to help distinguish from system-initiated ones
	if request.GetReason() == "" {
		request.Reason = defaultUserTerminateReason
	}
	if request.GetIdentity() == "" {
		request.Identity = defaultUserTerminateIdentity
	}

	_, err = wh.historyClient.TerminateWorkflowExecution(ctx, &historyservice.TerminateWorkflowExecutionRequest{
		NamespaceId:      namespaceID.String(),
		TerminateRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.TerminateWorkflowExecutionResponse{}, nil
}

// DeleteWorkflowExecution deletes a closed workflow execution asynchronously (workflow must be completed or terminated before).
// This method is EXPERIMENTAL and may be changed or removed in a later release.
func (wh *WorkflowHandler) DeleteWorkflowExecution(ctx context.Context, request *workflowservice.DeleteWorkflowExecutionRequest) (_ *workflowservice.DeleteWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.WorkflowExecution); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.DeleteWorkflowExecution(ctx, &historyservice.DeleteWorkflowExecutionRequest{
		NamespaceId:        namespaceID.String(),
		WorkflowExecution:  request.GetWorkflowExecution(),
		ClosedWorkflowOnly: false,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.DeleteWorkflowExecutionResponse{}, nil
}

// ListOpenWorkflowExecutions is a visibility API to list the open executions in a specific namespace.
func (wh *WorkflowHandler) ListOpenWorkflowExecutions(ctx context.Context, request *workflowservice.ListOpenWorkflowExecutionsRequest) (_ *workflowservice.ListOpenWorkflowExecutionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	maxPageSize := int32(wh.config.VisibilityMaxPageSize(request.GetNamespace()))
	if request.GetMaximumPageSize() <= 0 || request.GetMaximumPageSize() > maxPageSize {
		request.MaximumPageSize = maxPageSize
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if request.StartTimeFilter == nil {
		request.StartTimeFilter = &filterpb.StartTimeFilter{}
	}

	earliestTime := request.StartTimeFilter.GetEarliestTime()
	latestTime := request.StartTimeFilter.GetLatestTime()
	query := []string{}

	query = append(query, fmt.Sprintf(
		"%s = '%s'",
		sadefs.ExecutionStatus,
		enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
	))

	if earliestTime != nil && !earliestTime.AsTime().IsZero() &&
		latestTime != nil && !latestTime.AsTime().IsZero() {
		if earliestTime.AsTime().After(latestTime.AsTime()) {
			return nil, errEarliestTimeIsGreaterThanLatestTime
		}
		query = append(query, fmt.Sprintf(
			"%s BETWEEN '%s' AND '%s'",
			sadefs.StartTime,
			earliestTime.AsTime().Format(time.RFC3339Nano),
			latestTime.AsTime().Format(time.RFC3339Nano),
		))
	} else if earliestTime != nil && !earliestTime.AsTime().IsZero() {
		query = append(query, fmt.Sprintf(
			"%s >= '%s'",
			sadefs.StartTime,
			earliestTime.AsTime().Format(time.RFC3339Nano),
		))
	} else if latestTime != nil && !latestTime.AsTime().IsZero() {
		query = append(query, fmt.Sprintf(
			"%s <= '%s'",
			sadefs.StartTime,
			latestTime.AsTime().Format(time.RFC3339Nano),
		))
	}

	if request.GetExecutionFilter() != nil {
		if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
			return nil, errListNotAllowed
		}
		query = append(query, fmt.Sprintf(
			"%s = '%s'",
			sadefs.WorkflowID,
			request.GetExecutionFilter().GetWorkflowId()))

		wh.logger.Debug("List open workflow with filter",
			tag.WorkflowNamespace(request.GetNamespace()), tag.WorkflowListWorkflowFilterByID)
	} else if request.GetTypeFilter() != nil {
		if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
			return nil, errListNotAllowed
		}
		query = append(query, fmt.Sprintf(
			"%s = '%s'",
			sadefs.WorkflowType,
			request.GetTypeFilter().GetName()))

		wh.logger.Debug("List open workflow with filter",
			tag.WorkflowNamespace(request.GetNamespace()), tag.WorkflowListWorkflowFilterByType)
	}

	baseReq := &manager.ListWorkflowExecutionsRequestV2{
		NamespaceID:   namespaceID,
		Namespace:     namespaceName,
		PageSize:      int(request.GetMaximumPageSize()),
		NextPageToken: request.NextPageToken,
		Query:         strings.Join(query, " AND "),
	}
	persistenceResp, err := wh.visibilityMgr.ListWorkflowExecutions(ctx, baseReq)

	if err != nil {
		return nil, err
	}

	return &workflowservice.ListOpenWorkflowExecutionsResponse{
		Executions:    persistenceResp.Executions,
		NextPageToken: persistenceResp.NextPageToken,
	}, nil
}

// ListClosedWorkflowExecutions is a visibility API to list the closed executions in a specific namespace.
func (wh *WorkflowHandler) ListClosedWorkflowExecutions(ctx context.Context, request *workflowservice.ListClosedWorkflowExecutionsRequest) (_ *workflowservice.ListClosedWorkflowExecutionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	maxPageSize := int32(wh.config.VisibilityMaxPageSize(request.GetNamespace()))
	if request.GetMaximumPageSize() <= 0 || request.GetMaximumPageSize() > maxPageSize {
		request.MaximumPageSize = maxPageSize
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if request.StartTimeFilter == nil {
		request.StartTimeFilter = &filterpb.StartTimeFilter{}
	}

	earliestTime := request.StartTimeFilter.GetEarliestTime()
	latestTime := request.StartTimeFilter.GetLatestTime()
	query := []string{}

	query = append(query, fmt.Sprintf(
		"%s != '%s'",
		sadefs.ExecutionStatus,
		enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
	))

	if earliestTime != nil && !earliestTime.AsTime().IsZero() &&
		latestTime != nil && !latestTime.AsTime().IsZero() {
		if earliestTime.AsTime().After(latestTime.AsTime()) {
			return nil, errEarliestTimeIsGreaterThanLatestTime
		}
		query = append(query, fmt.Sprintf(
			"%s BETWEEN '%s' AND '%s'",
			sadefs.CloseTime,
			earliestTime.AsTime().Format(time.RFC3339Nano),
			latestTime.AsTime().Format(time.RFC3339Nano),
		))
	} else if earliestTime != nil && !earliestTime.AsTime().IsZero() {
		query = append(query, fmt.Sprintf(
			"%s >= '%s'",
			sadefs.CloseTime,
			earliestTime.AsTime().Format(time.RFC3339Nano),
		))
	} else if latestTime != nil && !latestTime.AsTime().IsZero() {
		query = append(query, fmt.Sprintf(
			"%s <= '%s'",
			sadefs.CloseTime,
			latestTime.AsTime().Format(time.RFC3339Nano),
		))
	}

	if request.GetExecutionFilter() != nil {
		if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
			return nil, errListNotAllowed
		}
		query = append(query, fmt.Sprintf(
			"%s = '%s'",
			sadefs.WorkflowID,
			request.GetExecutionFilter().GetWorkflowId()))

		wh.logger.Debug("List closed workflow with filter",
			tag.WorkflowNamespace(request.GetNamespace()), tag.WorkflowListWorkflowFilterByID)
	} else if request.GetTypeFilter() != nil {
		if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
			return nil, errListNotAllowed
		}
		query = append(query, fmt.Sprintf(
			"%s = '%s'",
			sadefs.WorkflowType,
			request.GetTypeFilter().GetName()))

		wh.logger.Debug("List closed workflow with filter",
			tag.WorkflowNamespace(request.GetNamespace()), tag.WorkflowListWorkflowFilterByType)
	} else if request.GetStatusFilter() != nil {
		if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
			return nil, errListNotAllowed
		}
		if request.GetStatusFilter().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED || request.GetStatusFilter().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
			return nil, errStatusFilterMustBeNotRunning
		}
		query = append(query, fmt.Sprintf(
			"%s = '%s'",
			sadefs.ExecutionStatus,
			request.GetStatusFilter().GetStatus()))

		wh.logger.Debug("List closed workflow with filter",
			tag.WorkflowNamespace(request.GetNamespace()), tag.WorkflowListWorkflowFilterByStatus)
	}

	baseReq := &manager.ListWorkflowExecutionsRequestV2{
		NamespaceID:   namespaceID,
		Namespace:     namespaceName,
		PageSize:      int(request.GetMaximumPageSize()),
		NextPageToken: request.NextPageToken,
		Query:         strings.Join(query, " AND "),
	}
	persistenceResp, err := wh.visibilityMgr.ListWorkflowExecutions(ctx, baseReq)

	if err != nil {
		return nil, err
	}

	return &workflowservice.ListClosedWorkflowExecutionsResponse{
		Executions:    persistenceResp.Executions,
		NextPageToken: persistenceResp.NextPageToken,
	}, nil
}

// ListWorkflowExecutions is a visibility API to list workflow executions in a specific namespace.
func (wh *WorkflowHandler) ListWorkflowExecutions(ctx context.Context, request *workflowservice.ListWorkflowExecutionsRequest) (_ *workflowservice.ListWorkflowExecutionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	maxPageSize := int32(wh.config.VisibilityMaxPageSize(request.GetNamespace()))
	if request.GetPageSize() <= 0 || request.GetPageSize() > maxPageSize {
		request.PageSize = maxPageSize
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	metrics.VisibilityListWorkflowsQueryLength.With(wh.metricsScope(ctx)).Record(int64(len(request.GetQuery())))

	req := &manager.ListWorkflowExecutionsRequestV2{
		NamespaceID:   namespaceID,
		Namespace:     namespaceName,
		PageSize:      int(request.GetPageSize()),
		NextPageToken: request.NextPageToken,
		Query:         request.GetQuery(),
	}
	persistenceResp, err := wh.visibilityMgr.ListWorkflowExecutions(ctx, req)
	if err != nil {
		return nil, err
	}

	return &workflowservice.ListWorkflowExecutionsResponse{
		Executions:    persistenceResp.Executions,
		NextPageToken: persistenceResp.NextPageToken,
	}, nil
}

// ListArchivedWorkflowExecutions is a visibility API to list archived workflow executions in a specific namespace.
func (wh *WorkflowHandler) ListArchivedWorkflowExecutions(ctx context.Context, request *workflowservice.ListArchivedWorkflowExecutionsRequest) (_ *workflowservice.ListArchivedWorkflowExecutionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	maxPageSize := int32(wh.config.VisibilityArchivalQueryMaxPageSize())
	if request.GetPageSize() <= 0 {
		request.PageSize = maxPageSize
	} else if request.GetPageSize() > maxPageSize {
		return nil, serviceerror.NewInvalidArgumentf(errPageSizeTooBigMessage, maxPageSize)
	}

	if !wh.archivalMetadata.GetVisibilityConfig().ClusterConfiguredForArchival() {
		return nil, errClusterIsNotConfiguredForVisibilityArchival
	}

	if !wh.archivalMetadata.GetVisibilityConfig().ReadEnabled() {
		return nil, errClusterIsNotConfiguredForReadingArchivalVisibility
	}

	entry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	if entry.VisibilityArchivalState().State != enumspb.ARCHIVAL_STATE_ENABLED {
		return nil, errNamespaceIsNotConfiguredForVisibilityArchival
	}

	URI, err := archiver.NewURI(entry.VisibilityArchivalState().URI)
	if err != nil {
		return nil, err
	}

	visibilityArchiver, err := wh.archiverProvider.GetVisibilityArchiver(URI.Scheme())
	if err != nil {
		return nil, err
	}

	archiverRequest := &archiver.QueryVisibilityRequest{
		NamespaceID:   entry.ID().String(),
		PageSize:      int(request.GetPageSize()),
		NextPageToken: request.NextPageToken,
		Query:         request.GetQuery(),
	}

	searchAttributes, err := wh.saProvider.GetSearchAttributes(wh.visibilityMgr.GetIndexName(), false)
	if err != nil {
		return nil, serviceerror.NewUnavailablef(errUnableToGetSearchAttributesMessage, err)
	}

	archiverResponse, err := visibilityArchiver.Query(
		ctx,
		URI,
		archiverRequest,
		searchAttributes)
	if err != nil {
		return nil, err
	}

	// special handling of ExecutionTime for cron or retry
	for _, execution := range archiverResponse.Executions {
		if execution.ExecutionTime == nil || execution.ExecutionTime.AsTime().IsZero() {
			execution.ExecutionTime = execution.GetStartTime()
		}
	}

	return &workflowservice.ListArchivedWorkflowExecutionsResponse{
		Executions:    archiverResponse.Executions,
		NextPageToken: archiverResponse.NextPageToken,
	}, nil
}

// ScanWorkflowExecutions _was_ a Visibility API to list large amount of workflow executions in a specific namespace without order.
// It has since been deprecated in favor of `ListWorkflowExecutions` and rewritten to use `ListWorkflowExecutions` internally.
// Deprecated: Use `ListWorkflowExecutions`
func (wh *WorkflowHandler) ScanWorkflowExecutions(ctx context.Context, request *workflowservice.ScanWorkflowExecutionsRequest) (_ *workflowservice.ScanWorkflowExecutionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	listResp, err := wh.ListWorkflowExecutions(ctx, &workflowservice.ListWorkflowExecutionsRequest{
		Namespace:     request.Namespace,
		PageSize:      request.PageSize,
		NextPageToken: request.NextPageToken,
		Query:         request.Query,
	})
	if err != nil {
		return nil, err
	}
	resp := &workflowservice.ScanWorkflowExecutionsResponse{
		Executions:    listResp.Executions,
		NextPageToken: listResp.NextPageToken,
	}
	return resp, nil
}

// CountWorkflowExecutions is a visibility API to count of workflow executions in a specific namespace.
func (wh *WorkflowHandler) CountWorkflowExecutions(ctx context.Context, request *workflowservice.CountWorkflowExecutionsRequest) (_ *workflowservice.CountWorkflowExecutionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	req := &manager.CountWorkflowExecutionsRequest{
		NamespaceID: namespaceID,
		Namespace:   namespaceName,
		Query:       request.GetQuery(),
	}
	persistenceResp, err := wh.visibilityMgr.CountWorkflowExecutions(ctx, req)
	if err != nil {
		return nil, err
	}

	resp := &workflowservice.CountWorkflowExecutionsResponse{
		Count:  persistenceResp.Count,
		Groups: persistenceResp.Groups,
	}
	return resp, nil
}

// GetSearchAttributes is a visibility API to get all legal keys that could be used in list APIs
func (wh *WorkflowHandler) GetSearchAttributes(ctx context.Context, _ *workflowservice.GetSearchAttributesRequest) (_ *workflowservice.GetSearchAttributesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	searchAttributes, err := wh.saProvider.GetSearchAttributes(wh.visibilityMgr.GetIndexName(), false)
	if err != nil {
		return nil, serviceerror.NewUnavailablef(errUnableToGetSearchAttributesMessage, err)
	}
	resp := &workflowservice.GetSearchAttributesResponse{
		Keys: searchAttributes.All(),
	}
	return resp, nil
}

// RespondQueryTaskCompleted is called by application worker to complete a QueryTask (which is a WorkflowTask for query)
// as a result of 'PollWorkflowTaskQueue' API call. Completing a QueryTask will unblock the client call to 'QueryWorkflow'
// API and return the query result to client as a response to 'QueryWorkflow' API call.
func (wh *WorkflowHandler) RespondQueryTaskCompleted(
	ctx context.Context,
	request *workflowservice.RespondQueryTaskCompletedRequest,
) (_ *workflowservice.RespondQueryTaskCompletedResponse, retError error) {

	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	queryTaskToken, err := wh.tokenSerializer.DeserializeQueryTaskToken(request.TaskToken)
	if err != nil {
		return nil, errDeserializingToken
	}
	if queryTaskToken.GetTaskQueue() == "" || queryTaskToken.GetTaskId() == "" {
		return nil, errInvalidTaskToken
	}
	namespaceId := namespace.ID(queryTaskToken.GetNamespaceId())
	namespaceEntry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceId)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())

	if err := common.CheckEventBlobSizeLimit(
		request.GetQueryResult().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceId.String(),
		"",
		"",
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"RespondQueryTaskCompleted",
	); err != nil {
		request = &workflowservice.RespondQueryTaskCompletedRequest{
			TaskToken:     request.TaskToken,
			CompletedType: enumspb.QUERY_RESULT_TYPE_FAILED,
			QueryResult:   nil,
			ErrorMessage:  err.Error(),
		}
	}

	matchingRequest := &matchingservice.RespondQueryTaskCompletedRequest{
		NamespaceId: namespaceId.String(),
		TaskQueue: &taskqueuepb.TaskQueue{
			Name: queryTaskToken.GetTaskQueue(),
			Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
		},
		TaskId:           queryTaskToken.GetTaskId(),
		CompletedRequest: request,
	}

	_, err = wh.matchingClient.RespondQueryTaskCompleted(ctx, matchingRequest)
	if err != nil {
		return nil, err
	}
	return &workflowservice.RespondQueryTaskCompletedResponse{}, nil
}

// ResetStickyTaskQueue resets the sticky taskqueue related information in mutable state of a given workflow.
// Things cleared are:
// 1. StickyTaskQueue
// 2. StickyScheduleToStartTimeout
func (wh *WorkflowHandler) ResetStickyTaskQueue(ctx context.Context, request *workflowservice.ResetStickyTaskQueueRequest) (_ *workflowservice.ResetStickyTaskQueueResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.Execution); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.ResetStickyTaskQueue(ctx, &historyservice.ResetStickyTaskQueueRequest{
		NamespaceId: namespaceID.String(),
		Execution:   request.Execution,
	})
	if err != nil {
		return nil, err
	}
	return &workflowservice.ResetStickyTaskQueueResponse{}, nil
}

func (wh *WorkflowHandler) ShutdownWorker(ctx context.Context, request *workflowservice.ShutdownWorkerRequest) (_ *workflowservice.ShutdownWorkerResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	// Run all shutdown operations concurrently to minimize latency.
	var waitGroup sync.WaitGroup

	// Cancel outstanding polls (best-effort)
	if wh.config.EnableCancelWorkerPollsOnShutdown(request.GetNamespace()) {
		waitGroup.Go(func() {
			wh.cancelOutstandingWorkerPolls(ctx, namespaceID.String(), request)
		})
	}

	// Record final heartbeat (best-effort)
	if request.WorkerHeartbeat != nil && wh.config.WorkerHeartbeatsEnabled(request.GetNamespace()) {
		waitGroup.Go(func() {
			_, err := wh.matchingClient.RecordWorkerHeartbeat(ctx, &matchingservice.RecordWorkerHeartbeatRequest{
				NamespaceId: namespaceID.String(),
				HeartbeartRequest: &workflowservice.RecordWorkerHeartbeatRequest{
					Namespace:       request.Namespace,
					Identity:        request.Identity,
					WorkerHeartbeat: []*workerpb.WorkerHeartbeat{request.WorkerHeartbeat},
				},
			})
			if err != nil {
				wh.logger.Error("Failed to record worker heartbeat during shutdown.",
					tag.WorkflowTaskQueueName(request.WorkerHeartbeat.GetTaskQueue()),
					tag.Error(err))
			}
		})
	}

	// Unload sticky task queue (required - error fails shutdown)
	// TODO: update poller info to indicate poller was shut down (pass identity/reason along)
	_, unloadError := wh.matchingClient.ForceUnloadTaskQueuePartition(ctx, &matchingservice.ForceUnloadTaskQueuePartitionRequest{
		NamespaceId: namespaceID.String(),
		TaskQueuePartition: &taskqueuespb.TaskQueuePartition{
			TaskQueue:     request.GetStickyTaskQueue(),
			TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, // sticky task queues are always workflow queues
		},
	})

	waitGroup.Wait()

	if unloadError != nil {
		return nil, unloadError
	}

	return &workflowservice.ShutdownWorkerResponse{}, nil
}

// cancelOutstandingWorkerPolls fans out poll cancellation to all partitions of the task queue.
// This is a best-effort operation - errors are logged but don't fail the shutdown.
func (wh *WorkflowHandler) cancelOutstandingWorkerPolls(
	ctx context.Context,
	namespaceID string,
	request *workflowservice.ShutdownWorkerRequest,
) {
	workerInstanceKey := request.GetWorkerInstanceKey()
	taskQueueName := request.GetTaskQueue()
	if workerInstanceKey == "" || taskQueueName == "" {
		return
	}

	namespaceName := request.GetNamespace()

	// Use task queue types from request, or default to both workflow and activity
	taskTypes := request.GetTaskQueueTypes()
	if len(taskTypes) == 0 {
		taskTypes = []enumspb.TaskQueueType{
			enumspb.TASK_QUEUE_TYPE_WORKFLOW,
			enumspb.TASK_QUEUE_TYPE_ACTIVITY,
		}
	}

	tqFamily, err := tqid.NewTaskQueueFamily(namespaceID, taskQueueName)
	if err != nil {
		wh.logger.Warn("Invalid task queue name for poll cancellation.",
			tag.WorkflowTaskQueueName(taskQueueName),
			tag.Error(err))
		return
	}

	// Deduplicate partitions by destination matching host. The partition is only used for
	// routing; the matching engine cancels all pollers for the workerInstanceKey on that host
	// regardless of partition. Sending one RPC per host instead of one per partition reduces
	// RPCs from numPartitions*taskTypes to numHosts*taskTypes.
	routingClient, ok := wh.matchingClient.(matchingclient.RoutingClient)
	if !ok {
		routingClient = nil
	}

	var waitGroup sync.WaitGroup
	var totalCancelled atomic.Int32
	var failedPartitions atomic.Int32

	for _, taskType := range taskTypes {
		numPartitions := wh.config.NumTaskQueueReadPartitions(namespaceName, taskQueueName, taskType)
		if numPartitions < 1 {
			numPartitions = 1
		}

		tq := tqFamily.TaskQueue(taskType)
		// Skip partitions that route to an already-visited matching host.
		seenHosts := make(map[string]bool)
		for partitionID := range numPartitions {
			partition := tq.NormalPartition(partitionID)

			if routingClient != nil {
				host, err := routingClient.Route(partition)
				if err != nil {
					wh.logger.Warn("Failed to resolve matching host for poll cancellation dedup, sending RPC anyway.",
						tag.WorkflowNamespaceID(namespaceID),
						tag.WorkflowTaskQueueName(partition.RpcName()),
						tag.Error(err))
				} else if seenHosts[host] {
					continue
				} else {
					seenHosts[host] = true
				}
			}

			waitGroup.Go(func() {
				resp, err := wh.matchingClient.CancelOutstandingWorkerPolls(ctx, &matchingservice.CancelOutstandingWorkerPollsRequest{
					NamespaceId: namespaceID,
					TaskQueue: &taskqueuepb.TaskQueue{
						Name: partition.RpcName(),
						Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
					},
					TaskQueueType:     taskType,
					WorkerInstanceKey: workerInstanceKey,
					WorkerIdentity:    request.GetIdentity(),
				})
				if err != nil {
					failedPartitions.Add(1)
					wh.logger.Warn("Failed to cancel outstanding polls for worker.",
						tag.WorkflowTaskQueueName(partition.RpcName()),
						tag.String("worker-instance-key", workerInstanceKey),
						tag.Error(err))
				} else {
					totalCancelled.Add(resp.CancelledCount)
				}
			})
		}
	}
	waitGroup.Wait()

	if totalCancelled.Load() > 0 || failedPartitions.Load() > 0 {
		wh.logger.Info("Cancelled outstanding polls for worker shutdown.",
			tag.String("worker-instance-key", workerInstanceKey),
			tag.NewInt32("cancelled-count", totalCancelled.Load()),
			tag.NewInt32("failed-partitions", failedPartitions.Load()))
	}
}

// QueryWorkflow returns query result for a specified workflow execution
func (wh *WorkflowHandler) QueryWorkflow(ctx context.Context, request *workflowservice.QueryWorkflowRequest) (_ *workflowservice.QueryWorkflowResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if wh.config.DisallowQuery(request.GetNamespace()) {
		return nil, errQueryDisallowedForNamespace
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	if err := validateExecution(request.Execution); err != nil {
		return nil, err
	}

	if request.Query == nil {
		return nil, errQueryNotSet
	}

	if request.Query.GetQueryType() == "" {
		return nil, errQueryTypeNotSet
	}

	enums.SetDefaultQueryRejectCondition(&request.QueryRejectCondition)

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	metricsHandler := wh.metricsScope(ctx).WithTags(metrics.HeaderCallsiteTag("QueryWorkflow"))
	metrics.HeaderSize.With(metricsHandler).Record(int64(request.GetQuery().GetHeader().Size()))

	sizeLimitError := wh.config.BlobSizeLimitError(request.GetNamespace())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(request.GetNamespace())

	if err := common.CheckEventBlobSizeLimit(
		request.GetQuery().GetQueryArgs().Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		request.GetExecution().GetWorkflowId(),
		request.GetExecution().GetRunId(),
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"QueryWorkflow"); err != nil {
		return nil, err
	}

	req := &historyservice.QueryWorkflowRequest{
		NamespaceId: namespaceID.String(),
		Request:     request,
	}
	hResponse, err := wh.historyClient.QueryWorkflow(ctx, req)
	if err != nil {
		if common.IsContextDeadlineExceededErr(err) {
			return nil, serviceerror.NewDeadlineExceeded("query timed out before a worker could process it")
		}
		return nil, err
	}
	return hResponse.GetResponse(), nil
}

// DescribeWorkflowExecution returns information about the specified workflow execution.
func (wh *WorkflowHandler) DescribeWorkflowExecution(ctx context.Context, request *workflowservice.DescribeWorkflowExecutionRequest) (_ *workflowservice.DescribeWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	if err := validateExecution(request.Execution); err != nil {
		return nil, err
	}

	response, err := wh.historyClient.DescribeWorkflowExecution(ctx, &historyservice.DescribeWorkflowExecutionRequest{
		NamespaceId: namespaceID.String(),
		Request:     request,
	})

	if err != nil {
		return nil, err
	}

	if response.GetWorkflowExecutionInfo().GetSearchAttributes() != nil {
		saTypeMap, err := wh.saProvider.GetSearchAttributes(wh.visibilityMgr.GetIndexName(), false)
		if err != nil {
			return nil, serviceerror.NewUnavailablef(errUnableToGetSearchAttributesMessage, err)
		}
		searchattribute.ApplyTypeMap(response.GetWorkflowExecutionInfo().GetSearchAttributes(), saTypeMap)
		aliasedSas, err := searchattribute.AliasFields(wh.saMapperProvider, response.GetWorkflowExecutionInfo().GetSearchAttributes(), request.GetNamespace())
		if err != nil {
			return nil, err
		}
		if aliasedSas != response.GetWorkflowExecutionInfo().GetSearchAttributes() {
			response.GetWorkflowExecutionInfo().SearchAttributes = aliasedSas
		}
	}

	response.GetWorkflowExecutionInfo().VersioningInfo = worker_versioning.AddV31VersioningInfoToV32(response.GetWorkflowExecutionInfo().GetVersioningInfo())

	return &workflowservice.DescribeWorkflowExecutionResponse{
		ExecutionConfig:        response.GetExecutionConfig(),
		WorkflowExecutionInfo:  response.GetWorkflowExecutionInfo(),
		PendingActivities:      response.GetPendingActivities(),
		PendingChildren:        response.GetPendingChildren(),
		PendingWorkflowTask:    response.GetPendingWorkflowTask(),
		Callbacks:              response.GetCallbacks(),
		PendingNexusOperations: response.GetPendingNexusOperations(),
		WorkflowExtendedInfo:   response.GetWorkflowExtendedInfo(),
	}, nil
}

// DescribeTaskQueue returns information about the target taskqueue, right now this API returns the
// pollers which polled this taskqueue in last few minutes.
func (wh *WorkflowHandler) DescribeTaskQueue(ctx context.Context, request *workflowservice.DescribeTaskQueueRequest) (_ *workflowservice.DescribeTaskQueueResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if request.ApiMode == enumspb.DESCRIBE_TASK_QUEUE_MODE_UNSPECIFIED {
		err = tqid.NormalizeAndValidatePartition(request.TaskQueue, "", wh.config.MaxIDLengthLimit())
	} else {
		err = tqid.NormalizeAndValidate(request.TaskQueue, "", wh.config.MaxIDLengthLimit())
	}
	if err != nil {
		return nil, err
	}

	if request.TaskQueueType == enumspb.TASK_QUEUE_TYPE_UNSPECIFIED || request.ApiMode == enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED {
		request.TaskQueueType = enumspb.TASK_QUEUE_TYPE_WORKFLOW
	}

	if len(request.TaskQueueTypes) == 0 {
		request.TaskQueueTypes = []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW, enumspb.TASK_QUEUE_TYPE_ACTIVITY}
	}

	if request.GetReportTaskReachability() &&
		len(request.GetVersions().GetBuildIds()) > wh.config.ReachabilityQueryBuildIdLimit() {
		return nil, serviceerror.NewInvalidArgumentf(
			"Too many build ids queried at once with ReportTaskReachability==true, limit: %d", wh.config.ReachabilityQueryBuildIdLimit())
	}

	if request.ApiMode == enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED {
		if request.TaskQueue.Kind == enumspb.TASK_QUEUE_KIND_STICKY {
			return nil, errUseEnhancedDescribeOnStickyQueue
		}
		if partition, err := tqid.PartitionFromProto(request.TaskQueue, namespaceID.String(), enumspb.TASK_QUEUE_TYPE_WORKFLOW); err != nil {
			return nil, errTaskQueuePartitionInvalid
		} else if !partition.IsRoot() {
			return nil, errUseEnhancedDescribeOnNonRootQueue
		}
	}

	matchingResponse, err := wh.matchingClient.DescribeTaskQueue(ctx, &matchingservice.DescribeTaskQueueRequest{
		NamespaceId: namespaceID.String(),
		DescRequest: request,
	})
	if err != nil {
		return nil, err
	}

	resp := matchingResponse.DescResponse
	// Manually parse unknown fields to handle proto incompatibility.
	// TODO: remove this after 1.24.0-m3
	if resp == nil {
		resp = &workflowservice.DescribeTaskQueueResponse{}
		unknown := []byte(matchingResponse.ProtoReflect().GetUnknown())
		for len(unknown) > 0 {
			num, typ, n := protowire.ConsumeTag(unknown)
			if n < 0 {
				break
			}
			unknown = unknown[n:]
			if typ != protowire.BytesType {
				break
			}
			msg, n := protowire.ConsumeBytes(unknown)
			if n < 0 {
				break
			}
			unknown = unknown[n:]
			switch num {
			case 1:
				// msg is either a temporal.api.workflowservice.v1.DescribeTaskQueueResponse (new) or repeated temporal.api.taskqueue.v1.PollerInfo (old)
				// try DescribeTaskQueueResponse first
				var dtqr workflowservice.DescribeTaskQueueResponse
				var pi taskqueuepb.PollerInfo
				if err := proto.Unmarshal(msg, &dtqr); err == nil {
					// merge this into the response, to avoid losing data in case this was a spurious success
					proto.Merge(resp, &dtqr)
				} else if err := proto.Unmarshal(msg, &pi); err == nil {
					resp.Pollers = append(resp.Pollers, &pi)
				}
			case 2:
				// msg should be a temporal.api.taskqueue.v1.TaskQueueStatus
				var tqstatus taskqueuepb.TaskQueueStatus
				if err := proto.Unmarshal(msg, &tqstatus); err == nil {
					resp.TaskQueueStatus = &tqstatus
				}
			}
		}
	}
	return resp, nil
}

// GetClusterInfo return information about Temporal deployment.
func (wh *WorkflowHandler) GetClusterInfo(ctx context.Context, _ *workflowservice.GetClusterInfoRequest) (_ *workflowservice.GetClusterInfoResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	metadata, err := wh.clusterMetadataManager.GetCurrentClusterMetadata(ctx)
	if err != nil {
		return nil, err
	}

	return &workflowservice.GetClusterInfoResponse{
		SupportedClients:         headers.SupportedClients,
		ServerVersion:            headers.ServerVersion,
		ClusterId:                metadata.ClusterId,
		VersionInfo:              metadata.VersionInfo,
		ClusterName:              metadata.ClusterName,
		HistoryShardCount:        metadata.HistoryShardCount,
		PersistenceStore:         wh.persistenceExecutionName,
		VisibilityStore:          strings.Join(wh.visibilityMgr.GetStoreNames(), ","),
		FailoverVersionIncrement: metadata.FailoverVersionIncrement,
		InitialFailoverVersion:   metadata.InitialFailoverVersion,
	}, nil
}

// GetSystemInfo returns information about the Temporal system.
func (wh *WorkflowHandler) GetSystemInfo(ctx context.Context, request *workflowservice.GetSystemInfoRequest) (_ *workflowservice.GetSystemInfoResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	return &workflowservice.GetSystemInfoResponse{
		ServerVersion: headers.ServerVersion,
		// Capabilities should be added as needed. In many cases, capabilities are
		// hardcoded boolean true values since older servers will respond with a
		// form of this message without the field which is implied false.
		Capabilities: &workflowservice.GetSystemInfoResponse_Capabilities{
			SignalAndQueryHeader:            true,
			InternalErrorDifferentiation:    true,
			ActivityFailureIncludeHeartbeat: true,
			SupportsSchedules:               true,
			EncodedFailureAttributes:        true,
			UpsertMemo:                      true,
			EagerWorkflowStart:              true,
			SdkMetadata:                     true,
			BuildIdBasedVersioning:          true,
			CountGroupByExecutionStatus:     true,
			Nexus:                           wh.httpEnabled,
			ServerScaledDeployments:         true,
		},
	}, nil
}

// ListTaskQueuePartitions returns all the partition and host for a task queue.
func (wh *WorkflowHandler) ListTaskQueuePartitions(ctx context.Context, request *workflowservice.ListTaskQueuePartitionsRequest) (_ *workflowservice.ListTaskQueuePartitionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceName := namespace.Name(request.GetNamespace())
	if err := tqid.NormalizeAndValidate(request.TaskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	matchingResponse, err := wh.matchingClient.ListTaskQueuePartitions(ctx, &matchingservice.ListTaskQueuePartitionsRequest{
		NamespaceId: namespaceID.String(),
		Namespace:   request.GetNamespace(),
		TaskQueue:   request.TaskQueue,
	})

	if matchingResponse == nil {
		return nil, err
	}

	return &workflowservice.ListTaskQueuePartitionsResponse{
		ActivityTaskQueuePartitions: matchingResponse.ActivityTaskQueuePartitions,
		WorkflowTaskQueuePartitions: matchingResponse.WorkflowTaskQueuePartitions,
	}, err
}

func (wh *WorkflowHandler) createScheduleCHASM(
	ctx context.Context,
	request *workflowservice.CreateScheduleRequest,
) (_ *workflowservice.CreateScheduleResponse, retError error) {
	namespaceName := namespace.Name(request.Namespace)
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	// Search attribute validation happens as part of unaliasing on the V1 codepath,
	// must be done explicitly here (even though we aren't using the unaliased
	// attributes).
	if _, err = wh.validator.UnaliasedSearchAttributesFrom(request.GetSearchAttributes(), request.Namespace); err != nil {
		return nil, err
	}

	if err := wh.validateSchedulePayloadSize(
		ctx, request.GetNamespace(), request.ScheduleId,
		request.GetSchedule(), request.GetMemo(), "CreateSchedule",
	); err != nil {
		return nil, err
	}

	// Phase 1: Write sentinel to V1 key space (dummy workflow) to prevent a
	// concurrent V1 CreateSchedule from succeeding for the same schedule ID.
	if wh.scheduleSentinelsEnabled(request.Namespace) {
		if err := wh.writeSchedulerWorkflowSentinel(ctx, namespaceID.String(), request); err != nil {
			var alreadyStartedErr *serviceerror.WorkflowExecutionAlreadyStarted
			if !errors.As(err, &alreadyStartedErr) {
				return nil, err
			}
			// V1 key is occupied. Check if it's a sentinel (proceed) or real scheduler (fail).
			isReal, checkErr := wh.isRealSchedulerInV1KeySpace(ctx, namespaceID.String(), request.Namespace, request.ScheduleId)
			if checkErr != nil {
				return nil, checkErr
			}
			if isReal {
				return nil, serviceerror.NewWorkflowExecutionAlreadyStarted(
					fmt.Sprintf("schedule %q is already registered", request.ScheduleId), "", "")
			}
		}
	}

	// Phase 2: Write real CHASM scheduler.
	res, err := wh.schedulerClient.CreateSchedule(ctx, &schedulerpb.CreateScheduleRequest{
		NamespaceId:     namespaceID.String(),
		FrontendRequest: request,
	})
	if err != nil {
		// The CHASM handler returns NotFound when the key is occupied by a
		// sentinel (written by a V1-path node that raced us). Translate to
		// WorkflowExecutionAlreadyStarted so the SDK can map it to
		// ErrScheduleAlreadyRunning.
		if isSchedulerErrorLegacyRoutable(err) {
			wh.logger.Warn("CreateSchedule race detected: sentinel found at CHASM key",
				tag.ScheduleID(request.ScheduleId))
			return nil, serviceerror.NewWorkflowExecutionAlreadyStarted(
				fmt.Sprintf("schedule %q: concurrent creation detected", request.ScheduleId), "", "")
		}
		var alreadyExistsErr *serviceerror.AlreadyExists
		if errors.As(err, &alreadyExistsErr) {
			return nil, serviceerror.NewWorkflowExecutionAlreadyStarted(alreadyExistsErr.Message, "", "")
		}
		return nil, err
	}
	return res.GetFrontendResponse(), nil
}

func (wh *WorkflowHandler) createScheduleWorkflow(
	ctx context.Context,
	request *workflowservice.CreateScheduleRequest,
) (_ *workflowservice.CreateScheduleResponse, retError error) {
	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId

	namespaceName := namespace.Name(request.Namespace)
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	// Phase 1: Write sentinel to CHASM key space to prevent a concurrent CHASM
	// CreateSchedule from succeeding for the same schedule ID.
	if wh.scheduleSentinelsEnabled(request.Namespace) {
		if err := wh.writeSchedulerCHASMSentinel(ctx, namespaceID.String(), namespaceName.String(), request.ScheduleId); err != nil {
			// Translate AlreadyExists (from CHASM handler) to
			// WorkflowExecutionAlreadyStarted for SDK compatibility.
			var alreadyExistsErr *serviceerror.AlreadyExists
			if errors.As(err, &alreadyExistsErr) {
				return nil, serviceerror.NewWorkflowExecutionAlreadyStarted(alreadyExistsErr.Message, "", "")
			}
			// Ignore unimplemented to avoid issues with mixed brain testing.
			//
			// We wouldn't hit this condition in prod, as we wouldn't migrate with the fleet
			// halfway deployed to the target version.
			var unimplErr *serviceerror.Unimplemented
			if !errors.As(err, &unimplErr) {
				return nil, err
			}
		}
	}

	// Phase 2: Write real V1 scheduler workflow.

	// Add namespace division before unaliasing search attributes.
	searchattribute.AddSearchAttribute(&request.SearchAttributes, sadefs.TemporalNamespaceDivision, payload.EncodeString(scheduler.NamespaceDivision))

	sa, err := wh.validator.UnaliasedSearchAttributesFrom(request.GetSearchAttributes(), request.Namespace)
	if err != nil {
		return nil, err
	}

	if startWorkflow := request.GetSchedule().GetAction().GetStartWorkflow(); startWorkflow != nil {
		metricsHandler := wh.metricsScope(ctx).WithTags(metrics.HeaderCallsiteTag("CreateSchedule"))
		metrics.HeaderSize.With(metricsHandler).Record(int64(startWorkflow.GetHeader().Size()))
	}

	// size limits will be validated on history. note that the start workflow request is
	// embedded in the schedule, which is in the scheduler input. so if the scheduler itself
	// doesn't exceed the limit, the started workflows should be safe as well.

	// Set up input to scheduler workflow
	input := &schedulespb.StartScheduleArgs{
		Schedule:     request.Schedule,
		InitialPatch: request.InitialPatch,
		State: &schedulespb.InternalState{
			Namespace:     namespaceName.String(),
			NamespaceId:   namespaceID.String(),
			ScheduleId:    request.ScheduleId,
			ConflictToken: scheduler.InitialConflictToken,
		},
	}
	inputPayloads, err := sdk.PreferProtoDataConverter.ToPayloads(input)
	if err != nil {
		return nil, err
	}
	// Add initial memo for list schedules
	wh.addInitialScheduleMemo(request, input)
	// Create StartWorkflowExecutionRequest
	startReq := &workflowservice.StartWorkflowExecutionRequest{
		Namespace:                request.Namespace,
		WorkflowId:               workflowID,
		WorkflowType:             &commonpb.WorkflowType{Name: scheduler.WorkflowType},
		TaskQueue:                &taskqueuepb.TaskQueue{Name: primitives.PerNSWorkerTaskQueue},
		Input:                    inputPayloads,
		Identity:                 request.Identity,
		RequestId:                request.RequestId,
		WorkflowIdReusePolicy:    enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
		WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL,
		Memo:                     request.Memo,
		SearchAttributes:         sa,
		Priority:                 &commonpb.Priority{}, // ie default priority
	}
	_, err = wh.historyClient.StartWorkflowExecution(
		ctx,
		common.CreateHistoryStartWorkflowRequest(
			namespaceID.String(),
			startReq,
			nil,
			nil,
			time.Now().UTC(),
		),
	)

	if err != nil {
		var alreadyStartedErr *serviceerror.WorkflowExecutionAlreadyStarted
		if errors.As(err, &alreadyStartedErr) {
			// V1 key is occupied. Check if it's a sentinel (race) or real scheduler.
			isReal, checkErr := wh.isRealSchedulerInV1KeySpace(ctx, namespaceID.String(), request.Namespace, request.ScheduleId)
			if checkErr != nil {
				return nil, checkErr
			}
			if !isReal {
				// A dummy workflow exists — a CHASM-path node raced us.
				wh.logger.Warn("CreateSchedule race detected: sentinel found at V1 key",
					tag.ScheduleID(request.ScheduleId))
				return nil, serviceerror.NewWorkflowExecutionAlreadyStarted(
					fmt.Sprintf("schedule %q: concurrent creation detected", request.ScheduleId), "", "")
			}
			return nil, serviceerror.NewWorkflowExecutionAlreadyStarted(
				fmt.Sprintf("schedule %q is already registered", request.ScheduleId), "", "")
		}
		return nil, err
	}
	token := make([]byte, 8)
	binary.BigEndian.PutUint64(token, scheduler.InitialConflictToken)
	return &workflowservice.CreateScheduleResponse{
		ConflictToken: token,
	}, nil
}

// writeSchedulerWorkflowSentinel starts a dummy workflow to reserve the V1
// workflow ID space for the given schedule ID. The dummy workflow sleeps for a
// fixed duration and then completes, releasing the reservation.
func (wh *WorkflowHandler) writeSchedulerWorkflowSentinel(
	ctx context.Context,
	namespaceID string,
	request *workflowservice.CreateScheduleRequest,
) error {
	startReq := &workflowservice.StartWorkflowExecutionRequest{
		Namespace:                request.Namespace,
		WorkflowId:               scheduler.WorkflowIDPrefix + request.ScheduleId,
		WorkflowType:             &commonpb.WorkflowType{Name: dummy.DummyWFTypeName},
		TaskQueue:                &taskqueuepb.TaskQueue{Name: primitives.PerNSWorkerTaskQueue},
		Identity:                 request.Identity,
		RequestId:                request.RequestId,
		WorkflowIdReusePolicy:    enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
		WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL,
		// Set TemporalNamespaceDivision so the dummy is hidden from
		// ListWorkflowExecutions. Use a distinct value so it doesn't match the
		// ListSchedules query (which filters on scheduler.NamespaceDivision).
		SearchAttributes: &commonpb.SearchAttributes{
			IndexedFields: map[string]*commonpb.Payload{
				sadefs.TemporalNamespaceDivision: payload.EncodeString("TemporalSchedulerSentinel"),
			},
		},
	}
	_, err := wh.historyClient.StartWorkflowExecution(
		ctx,
		common.CreateHistoryStartWorkflowRequest(namespaceID, startReq, nil, nil, time.Now().UTC()),
	)
	return err
}

// writeSchedulerCHASMSentinel creates a CHASM sentinel to reserve the CHASM key
// space for the given schedule ID. The sentinel auto-closes after a fixed
// duration via the idle task mechanism.
func (wh *WorkflowHandler) writeSchedulerCHASMSentinel(
	ctx context.Context,
	namespaceID, namespaceName, scheduleID string,
) error {
	_, err := wh.schedulerClient.CreateSentinel(ctx, &schedulerpb.CreateSentinelRequest{
		NamespaceId: namespaceID,
		Namespace:   namespaceName,
		ScheduleId:  scheduleID,
	})
	return err
}

// isRealSchedulerInV1KeySpace checks if a running V1 scheduler workflow (not a
// dummy sentinel) exists for the given schedule ID.
func (wh *WorkflowHandler) isRealSchedulerInV1KeySpace(
	ctx context.Context,
	namespaceID, namespaceName, scheduleID string,
) (bool, error) {
	workflowID := scheduler.WorkflowIDPrefix + scheduleID
	descResp, err := wh.historyClient.DescribeWorkflowExecution(ctx, &historyservice.DescribeWorkflowExecutionRequest{
		NamespaceId: namespaceID,
		Request: &workflowservice.DescribeWorkflowExecutionRequest{
			Namespace: namespaceName,
			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID},
		},
	})
	if err != nil {
		var notFoundErr *serviceerror.NotFound
		if errors.As(err, &notFoundErr) {
			return false, nil
		}
		return false, err
	}

	info := descResp.GetWorkflowExecutionInfo()
	if info.GetStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
		return false, nil
	}
	return info.GetType().GetName() != dummy.DummyWFTypeName, nil
}

func (wh *WorkflowHandler) CreateSchedule(
	ctx context.Context,
	request *workflowservice.CreateScheduleRequest,
) (_ *workflowservice.CreateScheduleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId
	// We apply this validation to both V1 and V2 schedules, even though CHASM
	// schedules don't need the workflow ID prefix, so that we can roll back to V1 and
	// not overrun the limit.
	if err := wh.validator.ValidateWorkflowID(workflowID); err != nil {
		return nil, err
	}

	if request.GetRequestId() == "" {
		return nil, errRequestIDNotSet
	}

	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
		return nil, errRequestIDTooLong
	}

	namespaceName := namespace.Name(request.Namespace)

	useChasmScheduler := wh.chasmSchedulerCreationEnabled(ctx, namespaceName.String())
	wh.logger.Debug("Received CreateSchedule",
		tag.ScheduleID(request.ScheduleId),
		tag.WorkflowNamespace(namespaceName.String()),
		tag.Bool("chasm-enabled", useChasmScheduler))

	if request.Schedule == nil {
		request.Schedule = &schedulepb.Schedule{}
	}
	err := wh.canonicalizeScheduleSpec(request.Schedule)
	if err != nil {
		return nil, err
	}

	if err = wh.validateStartWorkflowArgsForSchedule(namespaceName, request.GetSchedule().GetAction().GetStartWorkflow()); err != nil {
		return nil, err
	}

	if startWorkflow := request.GetSchedule().GetAction().GetStartWorkflow(); startWorkflow != nil {
		metricsHandler := wh.metricsScope(ctx).WithTags(metrics.HeaderCallsiteTag("CreateSchedule"))
		metrics.HeaderSize.With(metricsHandler).Record(int64(startWorkflow.GetHeader().Size()))
	}

	if useChasmScheduler {
		return wh.createScheduleCHASM(ctx, request)
	}
	return wh.createScheduleWorkflow(ctx, request)
}

// chasmSchedulerCreationEnabled returns true when CreateSchedule should create on
// the CHASM scheduler.
func (wh *WorkflowHandler) chasmSchedulerCreationEnabled(ctx context.Context, namespaceName string) bool {
	return (headers.IsExperimentRequested(ctx, ChasmSchedulerExperiment) &&
		wh.config.IsExperimentAllowed(ChasmSchedulerExperiment, namespaceName)) ||
		wh.config.EnableCHASMSchedulerCreation(namespaceName)
}

// chasmSchedulerEnabled returns true when schedule RPCs should route to CHASM
// first. Handlers must be capable of falling back to V1 codepaths for schedules
// that haven't been migrated to CHASM.
func (wh *WorkflowHandler) chasmSchedulerEnabled(ctx context.Context, namespaceName string) bool {
	return wh.chasmSchedulerCreationEnabled(ctx, namespaceName) ||
		wh.config.EnableCHASMSchedulerRouting(namespaceName)
}

func (wh *WorkflowHandler) scheduleSentinelsEnabled(namespaceName string) bool {
	return wh.config.EnableChasm(namespaceName) && wh.config.EnableCHASMSchedulerSentinels(namespaceName)
}

// isSchedulerErrorLegacyRoutable returns true if the error from the CHASM scheduler
// indicates that the request should be routed to the legacy (V1) scheduler stack.
// This accounts for three situations:
//   - NotFound: the CHASM stack doesn't have a schedule for that ID
//   - NotFound (sentinel): the key at that ID is a sentinel value (reserving the ID
//     for the V1 stack)
//   - ErrClosed: the CHASM schedule was migrated to V1 and marked closed; the
//     request should be retried against the workflow-backed stack.
func isSchedulerErrorLegacyRoutable(err error) bool {
	var notFoundErr *serviceerror.NotFound
	if errors.As(err, &notFoundErr) {
		return true
	}
	var failedPreconditionErr *serviceerror.FailedPrecondition
	if errors.As(err, &failedPreconditionErr) {
		return failedPreconditionErr.Message == chasmscheduler.ErrClosed.(*serviceerror.FailedPrecondition).Message
	}
	return false
}

// Validates inner start workflow request. Note that this can mutate search attributes if present.
func (wh *WorkflowHandler) validateStartWorkflowArgsForSchedule(
	namespaceName namespace.Name,
	startWorkflow *workflowpb.NewWorkflowExecutionInfo,
) error {
	if startWorkflow == nil {
		return nil
	}

	if err := wh.validator.ValidateWorkflowID(startWorkflow.WorkflowId + scheduler.AppendedTimestampForValidation); err != nil {
		return err
	}

	if startWorkflow.WorkflowType == nil || startWorkflow.WorkflowType.GetName() == "" {
		return errWorkflowTypeNotSet
	}

	if len(startWorkflow.WorkflowType.GetName()) > wh.config.MaxIDLengthLimit() {
		return errWorkflowTypeTooLong
	}

	// user cannot start internal workflows through schedule
	if err := tqid.NormalizeAndValidateUserDefined(startWorkflow.TaskQueue, "", "", wh.config.MaxIDLengthLimit()); err != nil {
		return err
	}

	if err := wh.validator.ValidateWorkflowTimeouts(&workflowservice.StartWorkflowExecutionRequest{
		WorkflowExecutionTimeout: startWorkflow.WorkflowExecutionTimeout,
		WorkflowRunTimeout:       startWorkflow.WorkflowRunTimeout,
		WorkflowTaskTimeout:      startWorkflow.WorkflowTaskTimeout,
	}); err != nil {
		return err
	}

	if len(startWorkflow.CronSchedule) > 0 {
		return errCronNotAllowed
	}

	if startWorkflow.WorkflowIdReusePolicy != enumspb.WORKFLOW_ID_REUSE_POLICY_UNSPECIFIED &&
		startWorkflow.WorkflowIdReusePolicy != enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE {
		return errIDReusePolicyNotAllowed
	}

	// Unalias startWorkflow search attributes only for validation.
	// Keep aliases in the request, because the request will be
	// sent back to frontend to start workflows, which will unalias at that point.
	unaliasedStartWorkflowSas, err := searchattribute.UnaliasFields(wh.saMapperProvider, startWorkflow.GetSearchAttributes(), namespaceName.String())
	if err != nil {
		return err
	}
	return wh.validator.ValidateSearchAttributes(unaliasedStartWorkflowSas, namespaceName.String())
}

// [cleanup-wv-pre-release]
func (wh *WorkflowHandler) DescribeDeployment(ctx context.Context, request *workflowservice.DescribeDeploymentRequest) (_ *workflowservice.DescribeDeploymentResponse, retError error) {
	return nil, serviceerror.NewUnimplemented("Deployments are deprecated and no longer supported, use Worker Deployments instead")
}

// [cleanup-wv-pre-release]
func (wh *WorkflowHandler) GetCurrentDeployment(ctx context.Context, request *workflowservice.GetCurrentDeploymentRequest) (_ *workflowservice.GetCurrentDeploymentResponse, retError error) {
	return nil, serviceerror.NewUnimplemented("Deployments are deprecated and no longer supported, use Worker Deployments instead")
}

// [cleanup-wv-pre-release]
func (wh *WorkflowHandler) ListDeployments(
	ctx context.Context,
	request *workflowservice.ListDeploymentsRequest,
) (_ *workflowservice.ListDeploymentsResponse, retError error) {
	return nil, serviceerror.NewUnimplemented("Deployments are deprecated and no longer supported, use Worker Deployments instead")
}

// [cleanup-wv-pre-release]
func (wh *WorkflowHandler) GetDeploymentReachability(
	ctx context.Context,
	request *workflowservice.GetDeploymentReachabilityRequest,
) (_ *workflowservice.GetDeploymentReachabilityResponse, retError error) {
	return nil, serviceerror.NewUnimplemented("Deployments are deprecated and no longer supported, use Worker Deployments instead")
}

// [cleanup-wv-pre-release]
func (wh *WorkflowHandler) SetCurrentDeployment(ctx context.Context, request *workflowservice.SetCurrentDeploymentRequest) (_ *workflowservice.SetCurrentDeploymentResponse, retError error) {
	return nil, serviceerror.NewUnimplemented("Deployments are deprecated and no longer supported, use Worker Deployments instead")
}

// Versioning-3 Public-Preview API's
func (wh *WorkflowHandler) DescribeWorkerDeploymentVersion(ctx context.Context, request *workflowservice.DescribeWorkerDeploymentVersionRequest) (_ *workflowservice.DescribeWorkerDeploymentVersionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	if err := wh.checkWorkerDeploymentReadRateLimit(ctx, request.GetNamespace(), "DescribeWorkerDeploymentVersion"); err != nil {
		return nil, err
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	versionStr := request.GetVersion()
	if versionStr == "" {
		if request.GetDeploymentVersion() == nil {
			return nil, serviceerror.NewInvalidArgument("deployment version cannot be empty")
		}
		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(request.GetDeploymentVersion())
	}

	versionInfo, vtq, err := wh.workerDeploymentClient.DescribeVersion(ctx, namespaceEntry, versionStr, request.ReportTaskQueueStats)
	if err != nil {
		return nil, err
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	versionInfo.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(versionInfo.Version)
	return &workflowservice.DescribeWorkerDeploymentVersionResponse{
		WorkerDeploymentVersionInfo: versionInfo,
		VersionTaskQueues:           vtq,
	}, nil
}

func (wh *WorkflowHandler) SetWorkerDeploymentCurrentVersion(ctx context.Context, request *workflowservice.SetWorkerDeploymentCurrentVersionRequest) (_ *workflowservice.SetWorkerDeploymentCurrentVersionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	if request.GetDeploymentName() == "" {
		return nil, serviceerror.NewInvalidArgument("deployment name cannot be empty")
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	versionStr := request.GetVersion()
	if versionStr == "" {
		var v *deploymentspb.WorkerDeploymentVersion
		if request.GetBuildId() != "" { // versioned
			v = &deploymentspb.WorkerDeploymentVersion{
				DeploymentName: request.GetDeploymentName(),
				BuildId:        request.GetBuildId(),
			}
		}
		versionStr = worker_versioning.WorkerDeploymentVersionToStringV31(v)
	}

	resp, err := wh.workerDeploymentClient.SetCurrentVersion(ctx, namespaceEntry, request.DeploymentName, versionStr, request.Identity, request.IgnoreMissingTaskQueues, request.GetConflictToken(), request.GetAllowNoPollers())
	if err != nil {
		return nil, err
	}

	return &workflowservice.SetWorkerDeploymentCurrentVersionResponse{
		ConflictToken:             resp.ConflictToken,
		PreviousVersion:           resp.PreviousVersion,
		PreviousDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(resp.PreviousVersion),
	}, nil
}

func (wh *WorkflowHandler) SetWorkerDeploymentRampingVersion(ctx context.Context, request *workflowservice.SetWorkerDeploymentRampingVersionRequest) (_ *workflowservice.SetWorkerDeploymentRampingVersionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	versionStr := request.GetVersion()
	if versionStr == "" {
		// If v0.31 user is trying to unset the ramp, let them do it until we update the deployment manager.
		// We know it's unsetting the ramp if Build ID is "" and percentage is 0.

		// This is a v0.32 user trying to ramp to unversioned.
		if request.GetBuildId() == "" && request.GetPercentage() > 0 {
			versionStr = worker_versioning.UnversionedVersionId
		}

		// This is a v0.32 user trying to ramp up a version. We don't care what percentage it is.
		if request.GetBuildId() != "" {
			versionStr = worker_versioning.WorkerDeploymentVersionToStringV31(&deploymentspb.WorkerDeploymentVersion{
				DeploymentName: request.GetDeploymentName(),
				BuildId:        request.GetBuildId(),
			})
		}
	}

	if versionStr == worker_versioning.UnversionedVersionId && request.GetPercentage() == 0 && request.GetBuildId() == "" {
		// GoSDK SDK seems to be sending __unversioned__ as the version string in case of unset ramp. empty that so the workflow is not confused.
		versionStr = ""
	}

	if request.GetPercentage() < 0 || request.GetPercentage() > 100 {
		return nil, serviceerror.NewInvalidArgument("Percentage must be between 0 and 100 (inclusive)")
	}

	resp, err := wh.workerDeploymentClient.SetRampingVersion(ctx, namespaceEntry, request.DeploymentName, versionStr, request.GetPercentage(), request.GetIdentity(), request.IgnoreMissingTaskQueues, request.GetConflictToken(), request.GetAllowNoPollers())
	if err != nil {
		return nil, err
	}

	return &workflowservice.SetWorkerDeploymentRampingVersionResponse{
		ConflictToken:             resp.ConflictToken,
		PreviousVersion:           resp.PreviousVersion,
		PreviousPercentage:        resp.PreviousPercentage,
		PreviousDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(resp.PreviousVersion),
	}, nil
}

func (wh *WorkflowHandler) ListWorkerDeployments(ctx context.Context, request *workflowservice.ListWorkerDeploymentsRequest) (_ *workflowservice.ListWorkerDeploymentsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	if wh.config.DisableListVisibilityByFilter(namespaceEntry.Name().String()) {
		return nil, errListNotAllowed
	}

	maxPageSize := int32(wh.config.VisibilityMaxPageSize(request.GetNamespace()))
	if request.GetPageSize() <= 0 || request.GetPageSize() > maxPageSize {
		request.PageSize = maxPageSize
	}

	resp, nextPageToken, err := wh.workerDeploymentClient.ListWorkerDeployments(ctx, namespaceEntry, int(request.PageSize), request.NextPageToken)
	if err != nil {
		return nil, err
	}

	workerDeployments := make([]*workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary, len(resp))
	for i, d := range resp {
		workerDeployments[i] = &workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary{
			Name:                  d.Name,
			CreateTime:            d.CreateTime,
			RoutingConfig:         d.RoutingConfig,
			LatestVersionSummary:  d.LatestVersionSummary,
			RampingVersionSummary: d.RampingVersionSummary,
			CurrentVersionSummary: d.CurrentVersionSummary,
		}
	}

	return &workflowservice.ListWorkerDeploymentsResponse{
		WorkerDeployments: workerDeployments,
		NextPageToken:     nextPageToken,
	}, nil
}

func (wh *WorkflowHandler) DescribeWorkerDeployment(ctx context.Context, request *workflowservice.DescribeWorkerDeploymentRequest) (_ *workflowservice.DescribeWorkerDeploymentResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	if err := wh.checkWorkerDeploymentReadRateLimit(ctx, request.GetNamespace(), "DescribeWorkerDeployment"); err != nil {
		return nil, err
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	workerDeploymentInfo, cT, err := wh.workerDeploymentClient.DescribeWorkerDeployment(ctx, namespaceEntry, request.DeploymentName)
	if err != nil {
		return nil, err
	}

	for _, vs := range workerDeploymentInfo.VersionSummaries {
		//nolint:staticcheck // SA1019: worker versioning v0.31
		vs.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(vs.Version)
	}
	return &workflowservice.DescribeWorkerDeploymentResponse{
		WorkerDeploymentInfo: workerDeploymentInfo,
		ConflictToken:        cT,
	}, nil
}

func (wh *WorkflowHandler) CreateWorkerDeployment(ctx context.Context, request *workflowservice.CreateWorkerDeploymentRequest) (_ *workflowservice.CreateWorkerDeploymentResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if request.DeploymentName == "" {
		return nil, serviceerror.NewInvalidArgument("deployment name cannot be empty")
	}

	if !wh.config.EnableDeploymentVersions(request.Namespace) {
		return nil, errDeploymentVersionsNotAllowed
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	// Generate request ID if not provided (for idempotency)
	requestID := request.RequestId
	if requestID == "" {
		requestID = uuid.NewString()
	}

	conflictToken, err := wh.workerDeploymentClient.CreateWorkerDeployment(
		ctx,
		namespaceEntry,
		request.DeploymentName,
		request.Identity,
		requestID,
	)
	if err != nil {
		return nil, err
	}

	return &workflowservice.CreateWorkerDeploymentResponse{
		ConflictToken: conflictToken,
	}, nil
}

func (wh *WorkflowHandler) SetWorkerDeploymentManager(ctx context.Context, request *workflowservice.SetWorkerDeploymentManagerRequest) (_ *workflowservice.SetWorkerDeploymentManagerResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if request.GetIdentity() == "" {
		return nil, serviceerror.NewInvalidArgument("Identity is required")
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	return wh.workerDeploymentClient.SetManager(ctx, namespaceEntry, request)
}

func (wh *WorkflowHandler) DeleteWorkerDeployment(ctx context.Context, request *workflowservice.DeleteWorkerDeploymentRequest) (_ *workflowservice.DeleteWorkerDeploymentResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	err = wh.workerDeploymentClient.DeleteWorkerDeployment(ctx, namespaceEntry, request.DeploymentName, request.Identity)
	if err != nil {
		return nil, err
	}

	return &workflowservice.DeleteWorkerDeploymentResponse{}, nil
}

func (wh *WorkflowHandler) DeleteWorkerDeploymentVersion(ctx context.Context, request *workflowservice.DeleteWorkerDeploymentVersionRequest) (_ *workflowservice.DeleteWorkerDeploymentVersionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	versionStr := request.GetVersion()
	if request.GetDeploymentVersion() != nil {
		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(request.GetDeploymentVersion())
	}

	err = wh.workerDeploymentClient.DeleteWorkerDeploymentVersion(ctx, namespaceEntry, versionStr, request.SkipDrainage, request.Identity)
	if err != nil {
		return nil, err
	}

	return &workflowservice.DeleteWorkerDeploymentVersionResponse{}, nil
}

func (wh *WorkflowHandler) UpdateWorkerDeploymentVersionMetadata(ctx context.Context, request *workflowservice.UpdateWorkerDeploymentVersionMetadataRequest) (_ *workflowservice.UpdateWorkerDeploymentVersionMetadataResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if request.RemoveEntries == nil && request.UpsertEntries == nil {
		return nil, serviceerror.NewInvalidArgument("At least one of remove_entries or upsert_entries must be provided")
	}

	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	version := request.GetDeploymentVersion()
	if version == nil {
		//nolint:staticcheck // SA1019: worker versioning v0.31
		version = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(request.GetVersion())
	}

	updatedMetadata, err := wh.workerDeploymentClient.UpdateVersionMetadata(ctx, namespaceEntry, version, request.UpsertEntries, request.RemoveEntries, request.Identity)
	if err != nil {
		return nil, err
	}

	return &workflowservice.UpdateWorkerDeploymentVersionMetadataResponse{
		Metadata: updatedMetadata,
	}, nil
}

// Returns the schedule description and current state of an existing schedule.
func (wh *WorkflowHandler) DescribeSchedule(ctx context.Context, request *workflowservice.DescribeScheduleRequest) (_ *workflowservice.DescribeScheduleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	// Prefer CHASM scheduler if enabled.
	if wh.chasmSchedulerEnabled(ctx, request.Namespace) {
		resp, err := wh.describeScheduleCHASM(ctx, request)
		if err == nil {
			return resp, nil
		}
		if !isSchedulerErrorLegacyRoutable(err) {
			return nil, err
		}
	}
	return wh.describeScheduleWorkflow(ctx, request)
}

func (wh *WorkflowHandler) describeScheduleCHASM(ctx context.Context, request *workflowservice.DescribeScheduleRequest) (*workflowservice.DescribeScheduleResponse, error) {
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	res, err := wh.schedulerClient.DescribeSchedule(ctx, &schedulerpb.DescribeScheduleRequest{
		FrontendRequest: request,
		NamespaceId:     namespaceID.String(),
	})
	return res.GetFrontendResponse(), err
}

func (wh *WorkflowHandler) describeScheduleWorkflow(ctx context.Context, request *workflowservice.DescribeScheduleRequest) (*workflowservice.DescribeScheduleResponse, error) {
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId
	execution := &commonpb.WorkflowExecution{WorkflowId: workflowID}

	// first describe to get memo and search attributes
	describeResponse, err := wh.historyClient.DescribeWorkflowExecution(ctx, &historyservice.DescribeWorkflowExecutionRequest{
		NamespaceId: namespaceID.String(),
		Request: &workflowservice.DescribeWorkflowExecutionRequest{
			Namespace: request.Namespace,
			Execution: execution,
		},
	})
	if err != nil {
		// TODO: rewrite "workflow" in error messages to "schedule"
		return nil, err
	}

	executionInfo := describeResponse.GetWorkflowExecutionInfo()
	if executionInfo.GetStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
		// only treat running schedules as existing
		return nil, serviceerror.NewNotFound("schedule not found")
	}
	if executionInfo.GetType().GetName() == dummy.DummyWFTypeName {
		// This is a sentinel workflow, not a real scheduler.
		return nil, serviceerror.NewNotFound("schedule not found")
	}

	// map search attributes
	if sas := executionInfo.GetSearchAttributes(); sas != nil {
		saTypeMap, err := wh.saProvider.GetSearchAttributes(wh.visibilityMgr.GetIndexName(), false)
		if err != nil {
			return nil, serviceerror.NewUnavailablef(errUnableToGetSearchAttributesMessage, err)
		}
		searchattribute.ApplyTypeMap(sas, saTypeMap)
		aliasedSas, err := searchattribute.AliasFields(wh.saMapperProvider, sas, request.GetNamespace())
		if err != nil {
			return nil, err
		}
		if aliasedSas != sas {
			executionInfo.SearchAttributes = aliasedSas
		}
	}

	// then query to get current state from the workflow itself
	req := &historyservice.QueryWorkflowRequest{
		NamespaceId: namespaceID.String(),
		Request: &workflowservice.QueryWorkflowRequest{
			Namespace: request.Namespace,
			Execution: execution,
			Query:     &querypb.WorkflowQuery{QueryType: scheduler.QueryNameDescribe},
		},
	}
	res, err := wh.historyClient.QueryWorkflow(ctx, req)
	if err != nil {
		return nil, err
	}

	var queryResponse schedulespb.DescribeResponse
	err = payloads.Decode(res.GetResponse().GetQueryResult(), &queryResponse)
	if err != nil {
		return nil, err
	}

	err = wh.annotateSearchAttributesOfScheduledWorkflow(&queryResponse, request.GetNamespace())
	if err != nil {
		return nil, serviceerror.NewInternalf("describe schedule: %v", err)
	}
	// Search attributes in the Action are already in external ("aliased") form. Do not alias them here.

	// for all running workflows started by the schedule, we should check that they're still running
	origLen := len(queryResponse.Info.RunningWorkflows)
	queryResponse.Info.RunningWorkflows = util.FilterSlice(queryResponse.Info.RunningWorkflows, func(ex *commonpb.WorkflowExecution) bool {
		// we'll usually have just zero or one of these so we can just do them sequentially
		msResponse, err := wh.historyClient.GetMutableState(ctx, &historyservice.GetMutableStateRequest{
			NamespaceId: namespaceID.String(),
			// Note: do not send runid here so that we always get the latest one
			Execution: &commonpb.WorkflowExecution{WorkflowId: ex.WorkflowId},
		})
		if err != nil {
			// if it's not found, it's certainly not running, so return false. if we got
			// another error, we don't know the state so assume it's still running.
			return !common.IsNotFoundError(err)
		}
		// return true if it is still running and is part of the chain the schedule started
		return msResponse.WorkflowStatus == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING &&
			msResponse.FirstExecutionRunId == ex.RunId
	})

	if len(queryResponse.Info.RunningWorkflows) < origLen {
		// we noticed some "running workflows" aren't running anymore. poke the workflow to
		// refresh, but don't wait for the state to change. ignore errors.
		go func() {
			disconnectedCtx := headers.SetCallerInfo(context.Background(), headers.NewBackgroundHighCallerInfo(request.Namespace))
			_, _ = wh.historyClient.SignalWorkflowExecution(disconnectedCtx, &historyservice.SignalWorkflowExecutionRequest{
				NamespaceId: namespaceID.String(),
				SignalRequest: &workflowservice.SignalWorkflowExecutionRequest{
					Namespace:         request.Namespace,
					WorkflowExecution: execution,
					SignalName:        scheduler.SignalNameRefresh,
					Identity:          "internal refresh from describe request",
					RequestId:         uuid.NewString(),
				},
			})
		}()
	}

	token := make([]byte, 8)
	binary.BigEndian.PutUint64(token, uint64(queryResponse.ConflictToken))

	searchAttributes := describeResponse.GetWorkflowExecutionInfo().GetSearchAttributes()
	searchAttributes = wh.cleanScheduleSearchAttributes(searchAttributes)

	memo := describeResponse.GetWorkflowExecutionInfo().GetMemo()
	memo = wh.cleanScheduleMemo(memo)

	scheduler.CleanSpec(queryResponse.Schedule.Spec)

	return &workflowservice.DescribeScheduleResponse{
		Schedule:         queryResponse.Schedule,
		Info:             queryResponse.Info,
		Memo:             memo,
		SearchAttributes: searchAttributes,
		ConflictToken:    token,
	}, nil
}

func (wh *WorkflowHandler) annotateSearchAttributesOfScheduledWorkflow(
	queryResponse *schedulespb.DescribeResponse,
	nsName string,
) error {
	ei := wh.getScheduledWorkflowExecutionInfoFrom(queryResponse)
	if ei == nil {
		return nil
	}
	annotatedAttributes, err := wh.annotateSearchAttributes(ei.GetSearchAttributes(), nsName)
	if err != nil {
		return fmt.Errorf("annotate search attributes: %w", err)
	}
	ei.SearchAttributes = annotatedAttributes
	return nil
}

func (wh *WorkflowHandler) getScheduledWorkflowExecutionInfoFrom(
	queryResponse *schedulespb.DescribeResponse,
) *workflowpb.NewWorkflowExecutionInfo {
	action := queryResponse.GetSchedule().GetAction().GetAction()
	startWorkflowAction, ok := action.(*schedulepb.ScheduleAction_StartWorkflow)
	if !ok {
		return nil
	}
	return startWorkflowAction.StartWorkflow
}

func (wh *WorkflowHandler) annotateSearchAttributes(
	searchAttributes *commonpb.SearchAttributes,
	nsName string,
) (*commonpb.SearchAttributes, error) {
	unaliasedSearchAttrs, err := searchattribute.UnaliasFields(
		wh.saMapperProvider,
		searchAttributes,
		nsName,
	)
	if err != nil {
		return nil, fmt.Errorf("create annotations: %w", err)
	}
	saTypeMap, err := wh.saProvider.GetSearchAttributes(wh.visibilityMgr.GetIndexName(), false)
	if err != nil {
		return nil, fmt.Errorf("create annotations: %w", err)
	}
	searchattribute.ApplyTypeMap(unaliasedSearchAttrs, saTypeMap)
	annotatedAttributes, err := searchattribute.AliasFields(
		wh.saMapperProvider,
		unaliasedSearchAttrs,
		nsName,
	)
	if err != nil {
		return nil, fmt.Errorf("create annotations: %w", err)
	}
	return annotatedAttributes, nil
}

// Changes the configuration or state of an existing schedule.
func (wh *WorkflowHandler) UpdateSchedule(
	ctx context.Context,
	request *workflowservice.UpdateScheduleRequest,
) (_ *workflowservice.UpdateScheduleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
		return nil, errRequestIDTooLong
	}

	namespaceName := namespace.Name(request.GetNamespace())

	if err := wh.validateStartWorkflowArgsForSchedule(
		namespaceName,
		request.GetSchedule().GetAction().GetStartWorkflow(),
	); err != nil {
		return nil, err
	}

	if request.Schedule == nil {
		request.Schedule = &schedulepb.Schedule{}
	}
	err := wh.canonicalizeScheduleSpec(request.Schedule)
	if err != nil {
		return nil, err
	}

	// Both V1 and V2 use unaliasedSearchAttributesFrom for validation, without using
	// the result. V1 uses UpsertSearchAttributes which expects aliased names, and V2
	// lets CHASM handle all visibility aliasing.
	if _, err = wh.validator.UnaliasedSearchAttributesFrom(request.GetSearchAttributes(), request.GetNamespace()); err != nil {
		return nil, err
	}

	if startWorkflow := request.GetSchedule().GetAction().GetStartWorkflow(); startWorkflow != nil {
		metricsHandler := wh.metricsScope(ctx).WithTags(metrics.HeaderCallsiteTag("UpdateSchedule"))
		metrics.HeaderSize.With(metricsHandler).Record(int64(startWorkflow.GetHeader().Size()))
	}

	if err := wh.validateSchedulePayloadSize(
		ctx, request.GetNamespace(), request.ScheduleId,
		request.GetSchedule(), request.GetMemo(), "UpdateSchedule",
	); err != nil {
		return nil, err
	}

	if wh.chasmSchedulerEnabled(ctx, request.Namespace) {
		res, err := wh.updateScheduleCHASM(ctx, request)
		if err == nil {
			return res, nil
		}
		if !isSchedulerErrorLegacyRoutable(err) {
			return nil, err
		}
	}

	// Reject memo updates for V1 schedules.
	if request.GetMemo() != nil {
		return nil, serviceerror.NewFailedPrecondition("memo updates are not supported on workflow-backed schedules")
	}

	return wh.updateScheduleWorkflow(ctx, request)
}

func (wh *WorkflowHandler) updateScheduleCHASM(
	ctx context.Context,
	request *workflowservice.UpdateScheduleRequest,
) (*workflowservice.UpdateScheduleResponse, error) {
	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	res, err := wh.schedulerClient.UpdateSchedule(ctx, &schedulerpb.UpdateScheduleRequest{
		FrontendRequest: request,
		NamespaceId:     namespaceID.String(),
	})
	if err != nil {
		return nil, err
	}
	return res.GetFrontendResponse(), err
}

func (wh *WorkflowHandler) updateScheduleWorkflow(
	ctx context.Context,
	request *workflowservice.UpdateScheduleRequest,
) (*workflowservice.UpdateScheduleResponse, error) {
	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	input := &schedulespb.FullUpdateRequest{
		Schedule:         request.Schedule,
		SearchAttributes: request.SearchAttributes,
	}
	if len(request.ConflictToken) >= 8 {
		input.ConflictToken = int64(binary.BigEndian.Uint64(request.ConflictToken))
	}
	inputPayloads, err := sdk.PreferProtoDataConverter.ToPayloads(input)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(request.GetNamespace())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(request.GetNamespace())
	if err := common.CheckEventBlobSizeLimit(
		inputPayloads.Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		workflowID,
		"", // don't have runid yet
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"UpdateSchedule",
	); err != nil {
		return nil, err
	}

	_, err = wh.historyClient.SignalWorkflowExecution(ctx, &historyservice.SignalWorkflowExecutionRequest{
		NamespaceId: namespaceID.String(),
		SignalRequest: &workflowservice.SignalWorkflowExecutionRequest{
			Namespace:         request.Namespace,
			WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: workflowID},
			SignalName:        scheduler.SignalNameUpdate,
			Input:             inputPayloads,
			Identity:          request.Identity,
			RequestId:         request.RequestId,
		},
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.UpdateScheduleResponse{}, nil
}

// Makes a specific change to a schedule or triggers an immediate action.
func (wh *WorkflowHandler) PatchSchedule(
	ctx context.Context,
	request *workflowservice.PatchScheduleRequest,
) (_ *workflowservice.PatchScheduleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
		return nil, errRequestIDTooLong
	}

	if len(request.Patch.Pause) > common.ScheduleNotesSizeLimit ||
		len(request.Patch.Unpause) > common.ScheduleNotesSizeLimit {
		return nil, errNotesTooLong
	}

	if trigger := request.Patch.GetTriggerImmediately(); trigger != nil && trigger.ScheduledTime == nil {
		trigger.ScheduledTime = timestamppb.Now()
	}

	inputPayloads, err := sdk.PreferProtoDataConverter.ToPayloads(request.Patch)
	if err != nil {
		return nil, err
	}

	// workflowID is included in size calculations with the V1 prefix, since V2 may
	// have to roll back.
	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(request.GetNamespace())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(request.GetNamespace())
	if err := common.CheckEventBlobSizeLimit(
		inputPayloads.Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		workflowID,
		"", // don't have runid yet
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"PatchSchedule",
	); err != nil {
		return nil, err
	}

	if wh.chasmSchedulerEnabled(ctx, request.Namespace) {
		res, err := wh.patchScheduleCHASM(ctx, request)
		if err == nil {
			return res, nil
		}
		if !isSchedulerErrorLegacyRoutable(err) {
			return nil, err
		}
	}

	return wh.patchScheduleWorkflow(ctx, request, inputPayloads)
}

func (wh *WorkflowHandler) patchScheduleWorkflow(
	ctx context.Context,
	request *workflowservice.PatchScheduleRequest,
	inputPayloads *commonpb.Payloads,
) (_ *workflowservice.PatchScheduleResponse, retError error) {
	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.SignalWorkflowExecution(ctx, &historyservice.SignalWorkflowExecutionRequest{
		NamespaceId: namespaceID.String(),
		SignalRequest: &workflowservice.SignalWorkflowExecutionRequest{
			Namespace:         request.Namespace,
			WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: workflowID},
			SignalName:        scheduler.SignalNamePatch,
			Input:             inputPayloads,
			Identity:          request.Identity,
			RequestId:         request.RequestId,
		},
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.PatchScheduleResponse{}, nil
}

func (wh *WorkflowHandler) patchScheduleCHASM(
	ctx context.Context,
	request *workflowservice.PatchScheduleRequest,
) (_ *workflowservice.PatchScheduleResponse, retError error) {
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	res, err := wh.schedulerClient.PatchSchedule(ctx, &schedulerpb.PatchScheduleRequest{
		NamespaceId:     namespaceID.String(),
		FrontendRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return res.GetFrontendResponse(), nil
}

// Lists matching times within a range.
func (wh *WorkflowHandler) ListScheduleMatchingTimes(ctx context.Context, request *workflowservice.ListScheduleMatchingTimesRequest) (_ *workflowservice.ListScheduleMatchingTimesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	// Prefer CHASM scheduler if enabled.
	if wh.chasmSchedulerEnabled(ctx, request.Namespace) {
		resp, err := wh.listScheduleMatchingTimesCHASM(ctx, request)
		if err == nil {
			return resp, nil
		}
		if !isSchedulerErrorLegacyRoutable(err) {
			return nil, err
		}
	}
	return wh.listScheduleMatchingTimesWorkflow(ctx, request)
}

func (wh *WorkflowHandler) listScheduleMatchingTimesCHASM(ctx context.Context, request *workflowservice.ListScheduleMatchingTimesRequest) (*workflowservice.ListScheduleMatchingTimesResponse, error) {
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	res, err := wh.schedulerClient.ListScheduleMatchingTimes(ctx, &schedulerpb.ListScheduleMatchingTimesRequest{
		FrontendRequest: request,
		NamespaceId:     namespaceID.String(),
	})
	return res.GetFrontendResponse(), err
}

func (wh *WorkflowHandler) listScheduleMatchingTimesWorkflow(ctx context.Context, request *workflowservice.ListScheduleMatchingTimesRequest) (*workflowservice.ListScheduleMatchingTimesResponse, error) {
	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	queryPayload, err := sdk.PreferProtoDataConverter.ToPayloads(request)
	if err != nil {
		return nil, err
	}

	sizeLimitError := wh.config.BlobSizeLimitError(request.GetNamespace())
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(request.GetNamespace())
	if err := common.CheckEventBlobSizeLimit(
		queryPayload.Size(),
		sizeLimitWarn,
		sizeLimitError,
		namespaceID.String(),
		workflowID,
		"",
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		"ListScheduleMatchingTimes"); err != nil {
		return nil, err
	}

	req := &historyservice.QueryWorkflowRequest{
		NamespaceId: namespaceID.String(),
		Request: &workflowservice.QueryWorkflowRequest{
			Namespace: request.Namespace,
			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID},
			Query: &querypb.WorkflowQuery{
				QueryType: scheduler.QueryNameListMatchingTimes,
				QueryArgs: queryPayload,
			},
		},
	}
	res, err := wh.historyClient.QueryWorkflow(ctx, req)
	if err != nil {
		return nil, err
	}

	var response workflowservice.ListScheduleMatchingTimesResponse
	err = payloads.Decode(res.GetResponse().GetQueryResult(), &response)
	if err != nil {
		return nil, err
	}

	return &response, nil
}

// Deletes a schedule, removing it from the system.
func (wh *WorkflowHandler) DeleteSchedule(ctx context.Context, request *workflowservice.DeleteScheduleRequest) (_ *workflowservice.DeleteScheduleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	// Always attempt deletion in both stacks. A schedule may exist in either or
	// both during dual-stack migration (and a V1 sentinel may linger after a
	// CHASM-only create). Surface an error only when neither stack succeeded.
	//
	// Each path gets its own metadata context so that downstream gRPC trailers
	// (propagated by TrailerToContextMetadataInterceptor) don't clobber each
	// other. After both paths complete, the winning side's metadata is copied
	// into the original context for upstream consumers (e.g. metering).
	chasmEnabled := wh.chasmSchedulerEnabled(ctx, request.Namespace)

	chasmCtx := contextutil.WithMetadataContext(ctx)
	var chasmErr error
	if chasmEnabled {
		_, chasmErr = wh.deleteScheduleCHASM(chasmCtx, request)
	}
	v1Ctx := contextutil.WithMetadataContext(ctx)
	_, v1Err := wh.deleteScheduleWorkflow(v1Ctx, request)

	// At least one side actually deleted -> success.
	if (chasmEnabled && chasmErr == nil) || v1Err == nil {
		// CHASM owns the schedule unless it returned a routable error
		// (NotFound/sentinel/closed), in which case V1 is the owner.
		winnerCtx := v1Ctx
		if chasmEnabled && (chasmErr == nil || !isSchedulerErrorLegacyRoutable(chasmErr)) {
			winnerCtx = chasmCtx
		}
		for k, v := range contextutil.ContextMetadataGetAll(winnerCtx) {
			contextutil.ContextMetadataSet(ctx, k, v)
		}
		return &workflowservice.DeleteScheduleResponse{}, nil
	}

	// Neither side deleted. Surface a real (non-routable) CHASM failure first;
	// otherwise return the V1 error (which is either a real failure or the
	// canonical NotFound when neither stack had the schedule).
	if chasmEnabled && chasmErr != nil && !isSchedulerErrorLegacyRoutable(chasmErr) {
		return nil, chasmErr
	}
	return nil, v1Err
}

func (wh *WorkflowHandler) deleteScheduleCHASM(ctx context.Context, request *workflowservice.DeleteScheduleRequest) (*workflowservice.DeleteScheduleResponse, error) {
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	res, err := wh.schedulerClient.DeleteSchedule(ctx, &schedulerpb.DeleteScheduleRequest{
		FrontendRequest: request,
		NamespaceId:     namespaceID.String(),
	})
	if err != nil {
		return nil, err
	}

	return res.GetFrontendResponse(), err
}

func (wh *WorkflowHandler) deleteScheduleWorkflow(ctx context.Context, request *workflowservice.DeleteScheduleRequest) (*workflowservice.DeleteScheduleResponse, error) {
	workflowID := scheduler.WorkflowIDPrefix + request.ScheduleId

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.TerminateWorkflowExecution(ctx, &historyservice.TerminateWorkflowExecutionRequest{
		NamespaceId: namespaceID.String(),
		TerminateRequest: &workflowservice.TerminateWorkflowExecutionRequest{
			Namespace:         request.Namespace,
			WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: workflowID},
			Reason:            "terminated by DeleteSchedule",
			Identity:          request.Identity,
		},
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.DeleteScheduleResponse{}, nil
}

// List all schedules in a namespace.
func (wh *WorkflowHandler) ListSchedules(
	ctx context.Context,
	request *workflowservice.ListSchedulesRequest,
) (_ *workflowservice.ListSchedulesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	maxPageSize := int32(wh.config.VisibilityMaxPageSize(request.GetNamespace()))
	if request.GetMaximumPageSize() <= 0 || request.GetMaximumPageSize() > maxPageSize {
		request.MaximumPageSize = maxPageSize
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
		return nil, errListNotAllowed
	}

	chasmEnabled := wh.chasmSchedulerEnabled(ctx, namespaceName.String())
	query, err := wh.prepareSchedulerQuery(chasmEnabled, request.Query, namespaceName)
	if err != nil {
		return nil, err
	}

	if chasmEnabled {
		// CHASM ListSchedules will include schedules created in the V1/workflow stack.
		return wh.listSchedulesChasm(ctx, request, namespaceName, namespaceID, query)
	}
	return wh.listSchedulesWorkflow(ctx, request, namespaceName, namespaceID, query)
}

// prepareSchedulerQuery validates a scheduler RPC's query argument, and wraps it
// in the appropriate base query.
func (wh *WorkflowHandler) prepareSchedulerQuery(
	chasmEnabled bool,
	query string,
	namespaceName namespace.Name,
) (string, error) {
	// Use different base queries based on code path:
	// - CHASM path uses TemporalSystemExecutionStatus (translated via archetype ID)
	// - V1 path uses ExecutionStatus directly (no archetype ID available)
	baseQuery := scheduler.VisibilityListQueryV1
	if chasmEnabled {
		baseQuery = scheduler.VisibilityListQueryChasm
	}

	result := baseQuery
	if strings.TrimSpace(query) != "" {
		saNameType, err := wh.saProvider.GetSearchAttributes(wh.visibilityMgr.GetIndexName(), false)
		if err != nil {
			return "", serviceerror.NewUnavailablef(errUnableToGetSearchAttributesMessage, err)
		}

		if err := scheduler.ValidateVisibilityQuery(
			namespaceName,
			saNameType,
			wh.saMapperProvider,
			wh.config.VisibilityEnableUnifiedQueryConverter,
			query,
		); err != nil {
			return "", err
		}

		saMapper, err := wh.saMapperProvider.GetMapper(namespaceName)
		if err != nil {
			return "", serviceerror.NewUnavailablef(errUnableToGetSearchAttributesMessage, err)
		}
		query, err = scheduler.RewriteScheduleIDQuery(query, chasmEnabled, saMapper, saNameType, namespaceName)
		if err != nil {
			return "", err
		}

		result = fmt.Sprintf("%s AND (%s)", baseQuery, query)
	}

	return result, nil
}

func (wh *WorkflowHandler) listSchedulesChasm(
	ctx context.Context,
	request *workflowservice.ListSchedulesRequest,
	namespaceName namespace.Name,
	namespaceID namespace.ID,
	query string,
) (_ *workflowservice.ListSchedulesResponse, retError error) {
	resp, err := chasm.ListExecutions[*chasmscheduler.Scheduler, *schedulepb.ScheduleListInfo](ctx, &chasm.ListExecutionsRequest{
		NamespaceName: namespaceName.String(),
		PageSize:      int(request.GetMaximumPageSize()),
		NextPageToken: request.NextPageToken,
		Query:         query,
	})
	if err != nil {
		return nil, err
	}

	schedules := make([]*schedulepb.ScheduleListEntry, len(resp.Executions))
	for i, ex := range resp.Executions {
		// Schedules returned may have either a Memo (V1) or a ChasmMemo (V2) containing
		// the ScheduleListInfo. Both must be handled, as migration means a mix of
		// versions can be returned.
		listInfo := ex.ChasmMemo // V2
		customMemo := ex.Memo
		if listInfo.GetSpec() == nil {
			listInfo = wh.decodeScheduleListInfo(customMemo) // V1
			wh.cleanScheduleMemo(customMemo)
		} else {
			scheduler.CleanSpec(listInfo.Spec) // done as part of decodeScheduleListInfo for V1
		}

		workflowID := ex.BusinessID
		scheduleID := strings.TrimPrefix(workflowID, scheduler.WorkflowIDPrefix) // needed for V1 schedules, not CHASM

		schedules[i] = &schedulepb.ScheduleListEntry{
			ScheduleId: scheduleID,
			Memo:       customMemo,
			// cleanScheduleSearchAttributes is only needed for V1 schedules
			SearchAttributes: wh.cleanScheduleSearchAttributes(&commonpb.SearchAttributes{
				IndexedFields: ex.CustomSearchAttributes,
			}),
			Info: listInfo,
		}
	}

	return &workflowservice.ListSchedulesResponse{
		Schedules:     schedules,
		NextPageToken: resp.NextPageToken,
	}, nil
}

func (wh *WorkflowHandler) listSchedulesWorkflow(
	ctx context.Context,
	request *workflowservice.ListSchedulesRequest,
	namespaceName namespace.Name,
	namespaceID namespace.ID,
	query string,
) (_ *workflowservice.ListSchedulesResponse, retError error) {
	persistenceResp, err := wh.visibilityMgr.ListWorkflowExecutions(
		ctx,
		&manager.ListWorkflowExecutionsRequestV2{
			NamespaceID:   namespaceID,
			Namespace:     namespaceName,
			PageSize:      int(request.GetMaximumPageSize()),
			NextPageToken: request.NextPageToken,
			Query:         query,
		},
	)
	if err != nil {
		return nil, err
	}

	schedules := make([]*schedulepb.ScheduleListEntry, len(persistenceResp.Executions))
	for i, ex := range persistenceResp.Executions {
		memo := ex.GetMemo()
		info := wh.decodeScheduleListInfo(memo)
		memo = wh.cleanScheduleMemo(memo)
		workflowID := ex.GetExecution().GetWorkflowId()
		scheduleID := strings.TrimPrefix(workflowID, scheduler.WorkflowIDPrefix)
		schedules[i] = &schedulepb.ScheduleListEntry{
			ScheduleId:       scheduleID,
			Memo:             memo,
			SearchAttributes: wh.cleanScheduleSearchAttributes(ex.GetSearchAttributes()),
			Info:             info,
		}
	}

	return &workflowservice.ListSchedulesResponse{
		Schedules:     schedules,
		NextPageToken: persistenceResp.NextPageToken,
	}, nil
}

func (wh *WorkflowHandler) CountSchedules(
	ctx context.Context,
	request *workflowservice.CountSchedulesRequest,
) (_ *workflowservice.CountSchedulesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableSchedules(request.Namespace) {
		return nil, errSchedulesNotAllowed
	}

	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if wh.config.DisableListVisibilityByFilter(namespaceName.String()) {
		return nil, errListNotAllowed
	}

	chasmEnabled := wh.chasmSchedulerEnabled(ctx, namespaceName.String())
	query, err := wh.prepareSchedulerQuery(chasmEnabled, request.Query, namespaceName)
	if err != nil {
		return nil, err
	}

	// Route to CHASM or V1 based on config (same pattern as ListSchedules)
	if chasmEnabled {
		return wh.countSchedulesChasm(ctx, namespaceID, namespaceName, query)
	}
	return wh.countSchedulesWorkflow(ctx, namespaceID, namespaceName, query)
}

// countSchedulesChasm counts schedules using CHASM APIs
func (wh *WorkflowHandler) countSchedulesChasm(
	ctx context.Context,
	namespaceID namespace.ID,
	namespaceName namespace.Name,
	query string,
) (*workflowservice.CountSchedulesResponse, error) {
	resp, err := chasm.CountExecutions[*chasmscheduler.Scheduler](ctx, &chasm.CountExecutionsRequest{
		NamespaceName: namespaceName.String(),
		Query:         query,
	})
	if err != nil {
		return nil, err
	}

	groups := make([]*workflowservice.CountSchedulesResponse_AggregationGroup, 0, len(resp.Groups))
	for _, g := range resp.Groups {
		groups = append(groups, &workflowservice.CountSchedulesResponse_AggregationGroup{
			GroupValues: g.Values,
			Count:       g.Count,
		})
	}

	return &workflowservice.CountSchedulesResponse{
		Count:  resp.Count,
		Groups: groups,
	}, nil
}

// countSchedulesWorkflow counts schedules using direct visibility query (V1 path)
func (wh *WorkflowHandler) countSchedulesWorkflow(
	ctx context.Context,
	namespaceID namespace.ID,
	namespaceName namespace.Name,
	query string,
) (*workflowservice.CountSchedulesResponse, error) {
	persistenceResp, err := wh.visibilityMgr.CountWorkflowExecutions(ctx, &manager.CountWorkflowExecutionsRequest{
		NamespaceID: namespaceID,
		Namespace:   namespaceName,
		Query:       query,
	})
	if err != nil {
		return nil, err
	}

	groups := make([]*workflowservice.CountSchedulesResponse_AggregationGroup, 0, len(persistenceResp.Groups))
	for _, g := range persistenceResp.Groups {
		groups = append(groups, &workflowservice.CountSchedulesResponse_AggregationGroup{
			GroupValues: g.GroupValues,
			Count:       g.Count,
		})
	}

	return &workflowservice.CountSchedulesResponse{
		Count:  persistenceResp.Count,
		Groups: groups,
	}, nil
}

func (wh *WorkflowHandler) UpdateWorkflowExecution(
	ctx context.Context,
	request *workflowservice.UpdateWorkflowExecutionRequest,
) (_ *workflowservice.UpdateWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if err := wh.prepareUpdateWorkflowRequest(request); err != nil {
		return nil, err
	}

	nsID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	metricsHandler := wh.metricsScope(ctx).WithTags(metrics.HeaderCallsiteTag("UpdateWorkflowExecution"))
	metrics.HeaderSize.With(metricsHandler).Record(int64(request.GetRequest().GetInput().GetHeader().Size()))
	metrics.EventBlobSize.With(wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String()))).Record(int64(request.GetRequest().GetInput().GetArgs().Size()), metrics.OperationTag("UpdateWorkflowExecution"))

	switch request.WaitPolicy.LifecycleStage { // nolint:exhaustive
	case enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED:
		metrics.WorkflowExecutionUpdateWaitStageAccepted.With(wh.metricsScope(ctx)).Record(1)
	case enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED:
		metrics.WorkflowExecutionUpdateWaitStageCompleted.With(wh.metricsScope(ctx)).Record(1)
	}

	histResp, err := wh.historyClient.UpdateWorkflowExecution(ctx, &historyservice.UpdateWorkflowExecutionRequest{
		NamespaceId: nsID.String(),
		Request:     request,
	})
	if err != nil {
		return nil, err
	}
	return histResp.GetResponse(), nil
}

func (wh *WorkflowHandler) prepareUpdateWorkflowRequest(
	request *workflowservice.UpdateWorkflowExecutionRequest,
) error {
	if request == nil {
		return errRequestNotSet
	}

	if err := validateExecution(request.GetWorkflowExecution()); err != nil {
		return err
	}

	if request.GetRequest().GetMeta() == nil {
		return errUpdateMetaNotSet
	}

	if len(request.GetRequest().GetMeta().GetUpdateId()) > wh.config.MaxIDLengthLimit() {
		return errUpdateIDTooLong
	}

	if request.GetRequest().GetMeta().GetUpdateId() == "" {
		request.GetRequest().GetMeta().UpdateId = uuid.NewString()
	}

	if request.GetRequest().GetInput() == nil {
		return errUpdateInputNotSet
	}

	if request.GetRequest().GetInput().GetName() == "" {
		return errUpdateNameNotSet
	}

	if request.GetWaitPolicy() == nil {
		request.WaitPolicy = &updatepb.WaitPolicy{}
	}

	if !wh.config.EnableUpdateWorkflowExecution(request.Namespace) {
		return errUpdateWorkflowExecutionAPINotAllowed
	}

	enums.SetDefaultUpdateWorkflowExecutionLifecycleStage(&request.GetWaitPolicy().LifecycleStage)

	if request.WaitPolicy.LifecycleStage == enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ADMITTED {
		return errUpdateWorkflowExecutionAsyncAdmittedNotAllowed
	}

	if request.WaitPolicy.LifecycleStage == enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED &&
		!wh.config.EnableUpdateWorkflowExecutionAsyncAccepted(request.Namespace) {
		return errUpdateWorkflowExecutionAsyncAcceptedNotAllowed
	}

	return nil
}

func (wh *WorkflowHandler) PollWorkflowExecutionUpdate(
	ctx context.Context,
	request *workflowservice.PollWorkflowExecutionUpdateRequest,
) (_ *workflowservice.PollWorkflowExecutionUpdateResponse, retError error) {
	if request == nil {
		return nil, errRequestNotSet
	}

	if request.GetUpdateRef() == nil {
		return nil, errUpdateRefNotSet
	}

	if request.GetWaitPolicy() == nil {
		request.WaitPolicy = &updatepb.WaitPolicy{}
	}

	nsID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	if !wh.config.EnableUpdateWorkflowExecution(request.Namespace) {
		return nil, errUpdateWorkflowExecutionAPINotAllowed
	}

	ctx, cancel := context.WithTimeout(ctx, frontend.DefaultLongPollTimeout)
	defer cancel()

	histResp, err := wh.historyClient.PollWorkflowExecutionUpdate(
		ctx,
		&historyservice.PollWorkflowExecutionUpdateRequest{
			NamespaceId: nsID.String(),
			Request:     request,
		},
	)
	if err != nil {
		return nil, err
	}
	return histResp.GetResponse(), nil
}

func (wh *WorkflowHandler) UpdateWorkerBuildIdCompatibility(ctx context.Context, request *workflowservice.UpdateWorkerBuildIdCompatibilityRequest) (_ *workflowservice.UpdateWorkerBuildIdCompatibilityResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableWorkerVersioningData(request.Namespace) {
		return nil, errWorkerVersioningV1_0NotAllowed
	}

	if err := wh.validateBuildIdCompatibilityUpdate(request); err != nil {
		return nil, err
	}

	taskQueue := &taskqueuepb.TaskQueue{Name: request.GetTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
	if err := tqid.NormalizeAndValidate(taskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	matchingResponse, err := wh.matchingClient.UpdateWorkerBuildIdCompatibility(ctx, &matchingservice.UpdateWorkerBuildIdCompatibilityRequest{
		NamespaceId: namespaceID.String(),
		TaskQueue:   request.GetTaskQueue(),
		Operation: &matchingservice.UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest_{
			ApplyPublicRequest: &matchingservice.UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest{
				Request: request,
			},
		},
	})

	if matchingResponse == nil {
		return nil, err
	}

	return &workflowservice.UpdateWorkerBuildIdCompatibilityResponse{}, err
}

func (wh *WorkflowHandler) GetWorkerBuildIdCompatibility(ctx context.Context, request *workflowservice.GetWorkerBuildIdCompatibilityRequest) (_ *workflowservice.GetWorkerBuildIdCompatibilityResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableWorkerVersioningData(request.Namespace) {
		return nil, errWorkerVersioningV1_0NotAllowed
	}

	taskQueue := &taskqueuepb.TaskQueue{Name: request.GetTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
	if err := tqid.NormalizeAndValidate(taskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	matchingResponse, err := wh.matchingClient.GetWorkerBuildIdCompatibility(ctx, &matchingservice.GetWorkerBuildIdCompatibilityRequest{
		NamespaceId: namespaceID.String(),
		Request:     request,
	})

	if matchingResponse == nil {
		return nil, err
	}

	return matchingResponse.Response, err
}

func (wh *WorkflowHandler) UpdateWorkerVersioningRules(ctx context.Context, request *workflowservice.UpdateWorkerVersioningRulesRequest) (_ *workflowservice.UpdateWorkerVersioningRulesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableWorkerVersioningRules(request.Namespace) {
		return nil, errWorkerVersioningV2_0NotAllowed
	}

	if err := wh.validateVersionRuleBuildId(request); err != nil {
		return nil, err
	}

	taskQueue := &taskqueuepb.TaskQueue{Name: request.GetTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
	if err := tqid.NormalizeAndValidate(taskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	matchingResponse, err := wh.matchingClient.UpdateWorkerVersioningRules(ctx, &matchingservice.UpdateWorkerVersioningRulesRequest{
		NamespaceId: namespaceID.String(),
		TaskQueue:   request.GetTaskQueue(),
		Command: &matchingservice.UpdateWorkerVersioningRulesRequest_Request{
			Request: request,
		},
	})

	if matchingResponse == nil {
		return nil, err
	}

	return matchingResponse.Response, err
}

func (wh *WorkflowHandler) GetWorkerVersioningRules(ctx context.Context, request *workflowservice.GetWorkerVersioningRulesRequest) (_ *workflowservice.GetWorkerVersioningRulesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableWorkerVersioningRules(request.Namespace) {
		return nil, errWorkerVersioningV2_0NotAllowed
	}

	taskQueue := &taskqueuepb.TaskQueue{Name: request.GetTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
	if err := tqid.NormalizeAndValidate(taskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	matchingResponse, err := wh.matchingClient.GetWorkerVersioningRules(ctx, &matchingservice.GetWorkerVersioningRulesRequest{
		NamespaceId: namespaceID.String(),
		TaskQueue:   request.GetTaskQueue(),
		Command: &matchingservice.GetWorkerVersioningRulesRequest_Request{
			Request: &workflowservice.GetWorkerVersioningRulesRequest{
				Namespace: request.GetNamespace(),
				TaskQueue: request.GetTaskQueue(),
			},
		},
	})

	if matchingResponse == nil {
		return nil, err
	}

	return matchingResponse.Response, err
}

func (wh *WorkflowHandler) GetWorkerTaskReachability(ctx context.Context, request *workflowservice.GetWorkerTaskReachabilityRequest) (_ *workflowservice.GetWorkerTaskReachabilityResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableWorkerVersioningData(request.Namespace) {
		return nil, errWorkerVersioningV2_0NotAllowed
	}

	if len(request.GetBuildIds()) == 0 {
		return nil, serviceerror.NewInvalidArgument("Must query at least one build ID (or empty string for unversioned worker)")
	}
	if len(request.GetBuildIds()) > wh.config.ReachabilityQueryBuildIdLimit() {
		return nil, serviceerror.NewInvalidArgumentf("Too many build ids queried at once, limit: %d", wh.config.ReachabilityQueryBuildIdLimit())
	}
	gotUnversionedRequest := false
	for _, buildId := range request.GetBuildIds() {
		if buildId == "" {
			gotUnversionedRequest = true
		}
		if len(buildId) > wh.config.WorkerBuildIdSizeLimit() {
			return nil, errBuildIdTooLong
		}
	}
	if gotUnversionedRequest && len(request.GetTaskQueues()) == 0 {
		return nil, serviceerror.NewInvalidArgument("Cannot get reachability of an unversioned worker without specifying at least one task queue (empty build ID is interpereted as unversioned)")
	}

	for _, taskQueue := range request.GetTaskQueues() {
		taskQueue := &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
		if err := tqid.NormalizeAndValidate(taskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
			return nil, err
		}
	}

	ns, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	response, err := wh.getWorkerTaskReachabilityValidated(ctx, ns, request)
	if err != nil {
		var invalidArgument *serviceerror.InvalidArgument
		if errors.As(err, &invalidArgument) {
			return nil, err
		}
		// Intentionally treat all errors as internal errors
		wh.logger.Error("Failed getting worker task reachability", tag.Error(err))
		return nil, serviceerror.NewInternal("Internal error")
	}
	return response, nil
}

func (wh *WorkflowHandler) StartBatchOperation(
	ctx context.Context,
	request *workflowservice.StartBatchOperationRequest,
) (_ *workflowservice.StartBatchOperationResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if err := wh.versionChecker.ClientSupported(ctx); err != nil {
		return nil, err
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.EnableBatcher(request.Namespace) {
		return nil, errBatchAPINotAllowed
	}

	if err := batcher.ValidateBatchOperation(request); err != nil {
		return nil, err
	}

	// Validate visibility query syntax before starting the batch workflow.
	// Malformed queries (e.g. "()") would otherwise cause the batch activity
	// to retry indefinitely since the error is not marked non-retryable.
	if q := request.GetVisibilityQuery(); len(q) > 0 {
		if _, err := sqlparser.Parse("select * from dummy where " + q); err != nil {
			return nil, serviceerror.NewInvalidArgumentf("invalid visibility query: %v", err)
		}
	}

	// Validate concurrent batch operation
	maxConcurrentBatchOperation := wh.config.MaxConcurrentBatchOperation(request.GetNamespace())
	countResp, err := wh.CountWorkflowExecutions(ctx, &workflowservice.CountWorkflowExecutionsRequest{
		Namespace: request.GetNamespace(),
		Query:     batcher.OpenBatchOperationQuery,
	})
	if err != nil {
		return nil, err
	}

	openBatchOperationCount := int(countResp.GetCount())
	if openBatchOperationCount >= maxConcurrentBatchOperation {
		return nil, &serviceerror.ResourceExhausted{
			Cause:   enumspb.RESOURCE_EXHAUSTED_CAUSE_CONCURRENT_LIMIT,
			Scope:   enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE,
			Message: "Max concurrent batch operations is reached",
		}
	}

	visibilityQuery := request.GetVisibilityQuery()

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	input := &batchspb.BatchOperationInput{
		Request:     request,
		NamespaceId: namespaceID.String(),
	}

	var identity string
	switch op := request.Operation.(type) {
	case *workflowservice.StartBatchOperationRequest_TerminationOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_TERMINATE
		identity = op.TerminationOperation.GetIdentity()
	case *workflowservice.StartBatchOperationRequest_SignalOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_SIGNAL
		identity = op.SignalOperation.GetIdentity()
	case *workflowservice.StartBatchOperationRequest_CancellationOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_CANCEL
		identity = op.CancellationOperation.GetIdentity()
	case *workflowservice.StartBatchOperationRequest_DeletionOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_DELETE
		identity = op.DeletionOperation.GetIdentity()
	case *workflowservice.StartBatchOperationRequest_ResetOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_RESET
		identity = op.ResetOperation.GetIdentity()
		for _, postOp := range op.ResetOperation.GetPostResetOperations() {
			if updateOpts := postOp.GetUpdateWorkflowOptions(); updateOpts != nil {
				if err := wh.validateTimeSkippingConfig(
					updateOpts.GetWorkflowExecutionOptions().GetTimeSkippingConfig(),
					namespace.Name(request.GetNamespace()),
				); err != nil {
					return nil, err
				}
			}
		}
	case *workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UPDATE_EXECUTION_OPTIONS
		identity = op.UpdateWorkflowOptionsOperation.GetIdentity()
		if err := wh.validateTimeSkippingConfig(
			op.UpdateWorkflowOptionsOperation.GetWorkflowExecutionOptions().GetTimeSkippingConfig(),
			namespace.Name(request.GetNamespace()),
		); err != nil {
			return nil, err
		}
	case *workflowservice.StartBatchOperationRequest_UnpauseActivitiesOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UNPAUSE_ACTIVITY
		identity = op.UnpauseActivitiesOperation.GetIdentity()

		switch a := op.UnpauseActivitiesOperation.GetActivity().(type) {
		case *batchpb.BatchOperationUnpauseActivities_Type:
			searchValue := fmt.Sprintf("property:activityType=%s", a.Type)
			escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
			input.Request.VisibilityQuery = fmt.Sprintf("%s = %s", sadefs.TemporalPauseInfo, escapedSearchValue)
		case *batchpb.BatchOperationUnpauseActivities_MatchAll:
			wildCardUnpause := fmt.Sprintf("%s STARTS_WITH 'property:activityType='", sadefs.TemporalPauseInfo)
			input.Request.VisibilityQuery = fmt.Sprintf("(%s) AND (%s)", visibilityQuery, wildCardUnpause)
		}
	case *workflowservice.StartBatchOperationRequest_ResetActivitiesOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_RESET_ACTIVITY
		identity = op.ResetActivitiesOperation.GetIdentity()
	case *workflowservice.StartBatchOperationRequest_UpdateActivityOptionsOperation:
		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UPDATE_ACTIVITY_OPTIONS
		identity = op.UpdateActivityOptionsOperation.GetIdentity()
	default:
		return nil, serviceerror.NewInvalidArgumentf("The operation type %T is not supported", op)
	}

	inputPayload, err := payloads.Encode(input)
	if err != nil {
		return nil, err
	}

	memo := &commonpb.Memo{
		Fields: map[string]*commonpb.Payload{
			batcher.BatchOperationTypeMemo: payload.EncodeString(snakeCaseBatchType(input.BatchType)),
			batcher.BatchReasonMemo:        payload.EncodeString(request.GetReason()),
		},
	}

	// Add pre-define search attributes
	var searchAttributes *commonpb.SearchAttributes
	searchattribute.AddSearchAttribute(&searchAttributes, sadefs.BatcherUser, payload.EncodeString(identity))
	searchattribute.AddSearchAttribute(&searchAttributes, sadefs.TemporalNamespaceDivision, payload.EncodeString(batcher.NamespaceDivision))

	startReq := &workflowservice.StartWorkflowExecutionRequest{
		Namespace:                request.Namespace,
		WorkflowId:               request.GetJobId(),
		WorkflowType:             &commonpb.WorkflowType{Name: batcher.BatchWFTypeProtobufName},
		TaskQueue:                &taskqueuepb.TaskQueue{Name: primitives.PerNSWorkerTaskQueue},
		Input:                    inputPayload,
		Identity:                 identity,
		RequestId:                uuid.NewString(),
		WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL,
		WorkflowIdReusePolicy:    enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
		Memo:                     memo,
		SearchAttributes:         searchAttributes,
		Priority:                 &commonpb.Priority{}, // ie default priority
	}

	_, err = wh.historyClient.StartWorkflowExecution(
		ctx,
		common.CreateHistoryStartWorkflowRequest(
			namespaceID.String(),
			startReq,
			nil,
			nil,
			time.Now().UTC(),
		),
	)
	if err != nil {
		return nil, err
	}
	return &workflowservice.StartBatchOperationResponse{}, nil
}

func snakeCaseBatchType(batchType enumspb.BatchOperationType) string {
	switch batchType {
	case enumspb.BATCH_OPERATION_TYPE_TERMINATE, enumspb.BATCH_OPERATION_TYPE_CANCEL, enumspb.BATCH_OPERATION_TYPE_SIGNAL, enumspb.BATCH_OPERATION_TYPE_DELETE, enumspb.BATCH_OPERATION_TYPE_RESET:
		return strings.ToLower(batchType.String())
	case enumspb.BATCH_OPERATION_TYPE_UPDATE_EXECUTION_OPTIONS:
		return "update_options"
	case enumspb.BATCH_OPERATION_TYPE_UNPAUSE_ACTIVITY:
		return "unpause_activities"
	case enumspb.BATCH_OPERATION_TYPE_UPDATE_ACTIVITY_OPTIONS:
		return "update_activity_options"
	case enumspb.BATCH_OPERATION_TYPE_RESET_ACTIVITY:
		return "reset_activities"
	default:
		return ""
	}
}

func (wh *WorkflowHandler) StopBatchOperation(
	ctx context.Context,
	request *workflowservice.StopBatchOperationRequest,
) (_ *workflowservice.StopBatchOperationResponse, retError error) {

	defer log.CapturePanic(wh.logger, &retError)

	if err := wh.versionChecker.ClientSupported(ctx); err != nil {
		return nil, err
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.GetJobId()) == 0 {
		return nil, errBatchJobIDNotSet
	}
	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}
	if len(request.Reason) == 0 {
		return nil, errReasonNotSet
	}

	if !wh.config.EnableBatcher(request.Namespace) {
		return nil, errBatchAPINotAllowed
	}

	terminateReq := &workflowservice.TerminateWorkflowExecutionRequest{
		Namespace: request.GetNamespace(),
		WorkflowExecution: &commonpb.WorkflowExecution{
			WorkflowId: request.GetJobId(),
		},
		Reason:   request.GetReason(),
		Identity: request.GetIdentity(),
	}
	_, err := wh.TerminateWorkflowExecution(ctx, terminateReq)
	if err != nil {
		return nil, err
	}
	return &workflowservice.StopBatchOperationResponse{}, nil
}

func (wh *WorkflowHandler) DescribeBatchOperation(
	ctx context.Context,
	request *workflowservice.DescribeBatchOperationRequest,
) (_ *workflowservice.DescribeBatchOperationResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if err := wh.versionChecker.ClientSupported(ctx); err != nil {
		return nil, err
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.GetJobId()) == 0 {
		return nil, errBatchJobIDNotSet
	}
	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableBatcher(request.Namespace) {
		return nil, errBatchAPINotAllowed
	}

	execution := &commonpb.WorkflowExecution{
		WorkflowId: request.GetJobId(),
		RunId:      "",
	}
	resp, err := wh.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
		Namespace: request.GetNamespace(),
		Execution: execution,
	})
	if err != nil {
		return nil, err
	}

	executionInfo := resp.GetWorkflowExecutionInfo()
	operationState := getBatchOperationState(executionInfo.GetStatus())
	memo := executionInfo.GetMemo().GetFields()
	typePayload := memo[batcher.BatchOperationTypeMemo]
	operationReason := memo[batcher.BatchReasonMemo]
	var reason string
	err = payload.Decode(operationReason, &reason)
	if err != nil {
		return nil, err
	}
	var identity string
	encodedBatcherIdentity := executionInfo.GetSearchAttributes().GetIndexedFields()[sadefs.BatcherUser]
	err = payload.Decode(encodedBatcherIdentity, &identity)
	if err != nil {
		return nil, err
	}
	var operationTypeString string
	err = payload.Decode(typePayload, &operationTypeString)
	if err != nil {
		return nil, err
	}
	var operationType enumspb.BatchOperationType
	switch operationTypeString {
	case batcher.BatchTypeCancel:
		operationType = enumspb.BATCH_OPERATION_TYPE_CANCEL
	case batcher.BatchTypeSignal:
		operationType = enumspb.BATCH_OPERATION_TYPE_SIGNAL
	case batcher.BatchTypeTerminate:
		operationType = enumspb.BATCH_OPERATION_TYPE_TERMINATE
	case batcher.BatchTypeDelete:
		operationType = enumspb.BATCH_OPERATION_TYPE_DELETE
	case batcher.BatchTypeReset:
		operationType = enumspb.BATCH_OPERATION_TYPE_RESET
	case batcher.BatchTypeUpdateOptions:
		operationType = enumspb.BATCH_OPERATION_TYPE_UPDATE_EXECUTION_OPTIONS
	case batcher.BatchTypeUpdateActivitiesOptions:
		operationType = enumspb.BATCH_OPERATION_TYPE_UPDATE_ACTIVITY_OPTIONS
	case batcher.BatchTypeResetActivities:
		operationType = enumspb.BATCH_OPERATION_TYPE_RESET_ACTIVITY
	case batcher.BatchTypeUnpauseActivities:
		operationType = enumspb.BATCH_OPERATION_TYPE_UNPAUSE_ACTIVITY
	default:
		operationType = enumspb.BATCH_OPERATION_TYPE_UNSPECIFIED
		wh.throttledLogger.Warn("Unknown batch operation type", tag.String("batch-operation-type", operationTypeString))
	}

	batchOperationResp := &workflowservice.DescribeBatchOperationResponse{
		OperationType: operationType,
		JobId:         executionInfo.Execution.GetWorkflowId(),
		State:         operationState,
		StartTime:     executionInfo.StartTime,
		CloseTime:     executionInfo.CloseTime,
		Identity:      identity,
		Reason:        reason,
	}
	if executionInfo.GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
		stats, err := wh.getCompletedBatchOperationStats(memo)
		if err != nil {
			return nil, err
		}
		batchOperationResp.TotalOperationCount = int64(stats.NumSuccess + stats.NumFailure)
		batchOperationResp.FailureOperationCount = int64(stats.NumFailure)
		batchOperationResp.CompleteOperationCount = int64(stats.NumSuccess)
	} else {
		if len(resp.GetPendingActivities()) > 0 {
			hbdPayload := resp.GetPendingActivities()[0].HeartbeatDetails
			var hbd batcher.HeartBeatDetails
			err = payloads.Decode(hbdPayload, &hbd)
			if err != nil {
				return nil, err
			}
			batchOperationResp.TotalOperationCount = hbd.TotalEstimate
			batchOperationResp.CompleteOperationCount = int64(hbd.SuccessCount)
			batchOperationResp.FailureOperationCount = int64(hbd.ErrorCount)
		}
	}
	return batchOperationResp, nil
}

func (wh *WorkflowHandler) getCompletedBatchOperationStats(memo map[string]*commonpb.Payload) (stats batcher.BatchOperationStats, err error) {
	statsPayload, ok := memo[batcher.BatchOperationStatsMemo]
	if !ok {
		return stats, errors.New("batch operation stats are not present in the memo")
	}
	err = payload.Decode(statsPayload, &stats)
	return stats, err
}

func (wh *WorkflowHandler) ListBatchOperations(
	ctx context.Context,
	request *workflowservice.ListBatchOperationsRequest,
) (_ *workflowservice.ListBatchOperationsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if err := wh.versionChecker.ClientSupported(ctx); err != nil {
		return nil, err
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	if len(request.Namespace) == 0 {
		return nil, errNamespaceNotSet
	}

	if !wh.config.EnableBatcher(request.Namespace) {
		return nil, errBatchAPINotAllowed
	}

	maxPageSize := int32(wh.config.VisibilityMaxPageSize(request.GetNamespace()))
	if request.GetPageSize() <= 0 || request.GetPageSize() > maxPageSize {
		request.PageSize = maxPageSize
	}

	resp, err := wh.ListWorkflowExecutions(ctx, &workflowservice.ListWorkflowExecutionsRequest{
		Namespace:     request.GetNamespace(),
		PageSize:      request.PageSize,
		NextPageToken: request.GetNextPageToken(),
		Query: fmt.Sprintf("%s = '%s'",
			sadefs.TemporalNamespaceDivision,
			batcher.NamespaceDivision,
		),
	})
	if err != nil {
		return nil, err
	}

	var operations []*batchpb.BatchOperationInfo
	for _, execution := range resp.GetExecutions() {
		operations = append(operations, &batchpb.BatchOperationInfo{
			JobId:     execution.GetExecution().GetWorkflowId(),
			State:     getBatchOperationState(execution.GetStatus()),
			StartTime: execution.GetStartTime(),
			CloseTime: execution.GetCloseTime(),
		})
	}
	return &workflowservice.ListBatchOperationsResponse{
		OperationInfo: operations,
		NextPageToken: resp.NextPageToken,
	}, nil
}

func (wh *WorkflowHandler) PollNexusTaskQueue(ctx context.Context, request *workflowservice.PollNexusTaskQueueRequest) (_ *workflowservice.PollNexusTaskQueueResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	callTime := time.Now().UTC()

	if request == nil {
		return nil, errRequestNotSet
	}

	wh.logger.Debug("Received PollNexusTaskQueue")
	if err := common.ValidateLongPollContextTimeout(ctx, "PollNexusTaskQueue", wh.throttledLogger); err != nil {
		return nil, err
	}

	namespaceName := namespace.Name(request.GetNamespace())

	if err := tqid.NormalizeAndValidate(request.TaskQueue, "", wh.config.MaxIDLengthLimit()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, errIdentityTooLong
	}

	// route heartbeat to the matching service
	if len(request.WorkerHeartbeat) > 0 && wh.config.WorkerHeartbeatsEnabled(request.GetNamespace()) {
		workerHeartbeat := request.WorkerHeartbeat
		request.WorkerHeartbeat = nil // Clear the field to avoid sending it to matching service.

		go func() {
			_, err := wh.matchingClient.RecordWorkerHeartbeat(ctx, &matchingservice.RecordWorkerHeartbeatRequest{
				NamespaceId: namespaceID.String(),
				HeartbeartRequest: &workflowservice.RecordWorkerHeartbeatRequest{
					Namespace:       request.Namespace,
					Identity:        request.Identity,
					WorkerHeartbeat: workerHeartbeat,
				},
			})
			if err != nil {
				wh.logger.Error("Failed to record worker heartbeat from nexus poll request.",
					tag.NexusTaskQueueName(request.GetTaskQueue().GetName()),
					tag.Error(err))
			}
		}()
	}

	//nolint:staticcheck // SA1019: worker versioning v0.31
	if err := wh.validateVersioningInfo(request.Namespace, request.WorkerVersionCapabilities, request.DeploymentOptions, request.TaskQueue); err != nil {
		return nil, err
	}

	if contextNearDeadline(ctx, longPollTailRoom) {
		return &workflowservice.PollNexusTaskQueueResponse{}, nil
	}

	pollerID := uuid.NewString()
	childCtx := wh.registerOutstandingPollContext(ctx, pollerID, namespaceID.String())
	defer wh.unregisterOutstandingPollContext(pollerID, namespaceID.String())
	matchingResponse, err := wh.matchingClient.PollNexusTaskQueue(childCtx, &matchingservice.PollNexusTaskQueueRequest{
		NamespaceId: namespaceID.String(),
		PollerId:    pollerID,
		Request:     request,
	})
	if err != nil {
		contextWasCanceled := wh.cancelOutstandingPoll(childCtx, namespaceID, enumspb.TASK_QUEUE_TYPE_NEXUS, request.TaskQueue, pollerID)
		if contextWasCanceled {
			// Clear error as we don't want to report context cancellation error to count against our SLA.
			return &workflowservice.PollNexusTaskQueueResponse{}, nil
		}

		// These errors are expected from some versioning situations. We should not log them, it'd be too noisy.
		var newerBuild *serviceerror.NewerBuildExists      // expected when versioned poller is superceded
		var failedPrecond *serviceerror.FailedPrecondition // expected when user data is disabled
		if errors.As(err, &newerBuild) || errors.As(err, &failedPrecond) {
			return nil, err
		}

		// For all other errors log an error and return it back to client.
		ctxTimeout := "not-set"
		ctxDeadline, ok := childCtx.Deadline()
		if ok {
			ctxTimeout = ctxDeadline.Sub(callTime).String()
		}
		wh.logger.Error("Unable to call matching.PollNexusTaskQueue.",
			tag.WorkflowTaskQueueName(request.GetTaskQueue().GetName()),
			tag.Timeout(ctxTimeout),
			tag.Error(err))

		return nil, err
	}

	return matchingResponse.GetResponse(), nil
}

func (wh *WorkflowHandler) RespondNexusTaskCompleted(ctx context.Context, request *workflowservice.RespondNexusTaskCompletedRequest) (_ *workflowservice.RespondNexusTaskCompletedResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if r := request.GetResponse().GetStartOperation().GetAsyncSuccess(); r != nil {
		operationToken := r.OperationToken
		if operationToken == "" && r.OperationId != "" { //nolint:staticcheck // SA1019 this field might be by old clients.
			operationToken = r.OperationId //nolint:staticcheck // SA1019 this field might be set by old clients.
		}
		if operationToken == "" {
			return nil, serviceerror.NewInvalidArgument("missing opration token in response")
		}

		tokenLimit := wh.config.MaxNexusOperationTokenLength(request.Namespace)
		if len(operationToken) > tokenLimit {
			return nil, serviceerror.NewInvalidArgumentf("operation token length exceeds allowed limit (%d/%d)", len(operationToken), tokenLimit)
		}
	}

	// Both the task token and the request have a reference to a namespace. We prefer using the namespace ID from
	// the token as it is a more stable identifier.
	// There's no need to validate that the namespace in the token and the request match,
	// NamespaceValidatorInterceptor does this for us.
	tt, err := wh.tokenSerializer.DeserializeNexusTaskToken(request.GetTaskToken())
	if err != nil {
		return nil, errDeserializingToken
	}
	if tt.GetTaskQueue() == "" || tt.GetTaskId() == "" {
		return nil, errInvalidTaskToken
	}
	namespaceId := namespace.ID(tt.GetNamespaceId())

	// NOTE: Not checking blob size limit here as we already enforce the 4 MB gRPC request limit and since this
	// doesn't go into workflow history, and the Nexus request caller is unknown, there doesn't seem like there's a
	// good reason to fail at this point.

	// nolint:staticcheck // checking deprecated field for backwards compatibility
	if opErr := request.GetResponse().GetStartOperation().GetOperationError(); opErr != nil {
		if details := opErr.GetFailure().GetDetails(); details != nil && !json.Valid(details) {
			return nil, serviceerror.NewInvalidArgument("failure details must be JSON serializable")
		}
	}

	matchingRequest := &matchingservice.RespondNexusTaskCompletedRequest{
		NamespaceId: namespaceId.String(),
		TaskQueue: &taskqueuepb.TaskQueue{
			Name: tt.GetTaskQueue(),
			Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
		},
		TaskId:  tt.GetTaskId(),
		Request: request,
	}

	_, err = wh.matchingClient.RespondNexusTaskCompleted(ctx, matchingRequest)
	if err != nil {
		return nil, err
	}
	return &workflowservice.RespondNexusTaskCompletedResponse{}, nil
}

func (wh *WorkflowHandler) RespondNexusTaskFailed(ctx context.Context, request *workflowservice.RespondNexusTaskFailedRequest) (_ *workflowservice.RespondNexusTaskFailedResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	// Both the task token and the request have a reference to a namespace. We prefer using the namespace ID from
	// the token as it is a more stable identifier.
	// There's no need to validate that the namespace in the token and the request match,
	// NamespaceValidatorInterceptor does this for us.
	tt, err := wh.tokenSerializer.DeserializeNexusTaskToken(request.GetTaskToken())
	if err != nil {
		return nil, errDeserializingToken
	}
	if tt.GetTaskQueue() == "" || tt.GetTaskId() == "" {
		return nil, errInvalidTaskToken
	}
	namespaceId := namespace.ID(tt.GetNamespaceId())

	if request.Error == nil && request.Failure == nil { // nolint:staticcheck // checking deprecated field for backwards compatibility
		return nil, serviceerror.NewInvalidArgument("request must contain error or failure")
	}
	if request.GetError() != nil { // nolint:staticcheck // checking deprecated field for backwards compatibility
		if details := request.GetError().GetFailure().GetDetails(); details != nil && !json.Valid(details) { // nolint:staticcheck // checking deprecated field for backwards compatibility
			return nil, serviceerror.NewInvalidArgument("failure details must be JSON serializable")
		}
	}
	if request.GetFailure() != nil && request.GetFailure().GetNexusHandlerFailureInfo() == nil {
		return nil, serviceerror.NewInvalidArgument("request Failure must contain error or failure with NexusHandlerFailureInfo")
	}

	// NOTE: Not checking blob size limit here as we already enforce the 4 MB gRPC request limit and since this
	// doesn't go into workflow history, and the Nexus request caller is unknown, there doesn't seem like there's a
	// good reason to fail at this point.

	matchingRequest := &matchingservice.RespondNexusTaskFailedRequest{
		NamespaceId: namespaceId.String(),
		TaskQueue: &taskqueuepb.TaskQueue{
			Name: tt.GetTaskQueue(),
			Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
		},
		TaskId:  tt.GetTaskId(),
		Request: request,
	}

	_, err = wh.matchingClient.RespondNexusTaskFailed(ctx, matchingRequest)
	if err != nil {
		return nil, err
	}
	return &workflowservice.RespondNexusTaskFailedResponse{}, nil
}

func (wh *WorkflowHandler) validateVersionRuleBuildId(request *workflowservice.UpdateWorkerVersioningRulesRequest) error {
	validateBuildId := func(bid string) error {
		if len(bid) > 255 {
			return serviceerror.NewInvalidArgumentf("BuildId must be <= 255 characters, was %d", len(bid))
		}
		return nil
	}
	switch request.GetOperation().(type) {
	case *workflowservice.UpdateWorkerVersioningRulesRequest_InsertAssignmentRule:
		return validateBuildId(request.GetInsertAssignmentRule().GetRule().GetTargetBuildId())
	case *workflowservice.UpdateWorkerVersioningRulesRequest_ReplaceAssignmentRule:
		return validateBuildId(request.GetReplaceAssignmentRule().GetRule().GetTargetBuildId())
	case *workflowservice.UpdateWorkerVersioningRulesRequest_DeleteAssignmentRule:
		return nil
	case *workflowservice.UpdateWorkerVersioningRulesRequest_AddCompatibleRedirectRule:
		return cmp.Or(
			validateBuildId(request.GetAddCompatibleRedirectRule().GetRule().GetTargetBuildId()),
			validateBuildId(request.GetAddCompatibleRedirectRule().GetRule().GetSourceBuildId()),
		)
	case *workflowservice.UpdateWorkerVersioningRulesRequest_ReplaceCompatibleRedirectRule:
		return validateBuildId(request.GetReplaceCompatibleRedirectRule().GetRule().GetTargetBuildId())
	case *workflowservice.UpdateWorkerVersioningRulesRequest_DeleteCompatibleRedirectRule:
		return nil
	case *workflowservice.UpdateWorkerVersioningRulesRequest_CommitBuildId_:
		return validateBuildId(request.GetCommitBuildId().GetTargetBuildId())
	}
	return nil
}

func (wh *WorkflowHandler) validateOnConflictOptions(opts *workflowpb.OnConflictOptions) error {
	if opts == nil {
		return nil
	}
	if opts.AttachCompletionCallbacks && !opts.AttachRequestId {
		return serviceerror.NewInvalidArgument("attaching request ID is required for attaching completion callbacks")
	}
	return nil
}

func (wh *WorkflowHandler) checkWorkerDeploymentReadRateLimit(ctx context.Context, namespaceName, apiName string) error {
	callerInfo := headers.GetCallerInfo(ctx)
	req := quotas.NewRequest(
		apiName,
		1,
		namespaceName,
		callerInfo.CallerType,
		-1,
		callerInfo.CallOrigin,
	)

	if !wh.workerDeploymentReadRateLimiter.Allow(time.Now().UTC(), req) {
		return serviceerror.NewResourceExhausted(
			enumspb.RESOURCE_EXHAUSTED_CAUSE_RPS_LIMIT,
			fmt.Sprintf("Worker Deployment Read API rate limit exceeded for namespace %q", namespaceName),
		)
	}
	return nil
}

func dedupLinksFromCallbacks(
	links []*commonpb.Link,
	callbacks []*commonpb.Callback,
) []*commonpb.Link {
	if len(links) == 0 {
		return nil
	}
	var res []*commonpb.Link
	callbacksLinks := make([]*commonpb.Link, 0, len(callbacks))
	for _, cb := range callbacks {
		if cb.GetNexus() != nil {
			// Only dedup links from Nexus callbacks.
			callbacksLinks = append(callbacksLinks, cb.GetLinks()...)
		}
	}
	for _, link := range links {
		isDup := false
		for _, cbLink := range callbacksLinks {
			if proto.Equal(link, cbLink) {
				isDup = true
				break
			}
		}
		if !isDup {
			res = append(res, link)
		}
	}
	return res
}

func (wh *WorkflowHandler) validateLinks(
	ns namespace.Name,
	links []*commonpb.Link,
) error {
	maxAllowedLinks := wh.config.MaxLinksPerRequest(ns.String())
	if len(links) > maxAllowedLinks {
		return serviceerror.NewInvalidArgumentf("cannot attach more than %d links per request, got %d", maxAllowedLinks, len(links))
	}

	maxSize := wh.config.LinkMaxSize(ns.String())
	for _, l := range links {
		if l.Size() > maxSize {
			return serviceerror.NewInvalidArgumentf("link exceeds allowed size of %d, got %d", maxSize, l.Size())
		}
		switch t := l.Variant.(type) {
		case *commonpb.Link_WorkflowEvent_:
			if t.WorkflowEvent.GetNamespace() == "" {
				return serviceerror.NewInvalidArgument("workflow event link must not have an empty namespace field")
			}
			if t.WorkflowEvent.GetWorkflowId() == "" {
				return serviceerror.NewInvalidArgument("workflow event link must not have an empty workflow ID field")
			}
			if t.WorkflowEvent.GetRunId() == "" {
				return serviceerror.NewInvalidArgument("workflow event link must not have an empty run ID field")
			}
			if t.WorkflowEvent.GetEventRef().GetEventType() == enumspb.EVENT_TYPE_UNSPECIFIED && t.WorkflowEvent.GetEventRef().GetEventId() != 0 {
				return serviceerror.NewInvalidArgument("workflow event link ref cannot have an unspecified event type and a non-zero event ID")
			}
		case *commonpb.Link_BatchJob_:
			if t.BatchJob.GetJobId() == "" {
				return serviceerror.NewInvalidArgument("batch job link must not have an empty job ID")
			}
		case *commonpb.Link_NexusOperation_:
			if t.NexusOperation.GetNamespace() == "" {
				return serviceerror.NewInvalidArgument("nexus operation link must not have an empty namespace field")
			}
			if t.NexusOperation.GetOperationId() == "" {
				return serviceerror.NewInvalidArgument("nexus operation link must not have an empty operation ID field")
			}
			if t.NexusOperation.GetRunId() == "" {
				return serviceerror.NewInvalidArgument("nexus operation link must not have an empty run ID field")
			}
		default:
			return serviceerror.NewInvalidArgument("unsupported link variant")
		}
	}
	return nil
}

type buildIdAndFlag interface {
	GetBuildId() string
	GetUseVersioning() bool
}

func (wh *WorkflowHandler) validateVersioningInfo(nsName string, id buildIdAndFlag, deploymentOptions *deploymentpb.WorkerDeploymentOptions, tq *taskqueuepb.TaskQueue) error {
	// TODO: Deprecate old versioning checks
	if id.GetUseVersioning() && !wh.config.EnableWorkerVersioningWorkflow(nsName) {
		return errWorkerVersioningWorkflowAPIsNotAllowed
	}
	if tq.GetKind() == enumspb.TASK_QUEUE_KIND_STICKY && len(tq.GetNormalName()) == 0 {
		if id.GetUseVersioning() || deploymentOptions != nil {
			// Versioned pollers require a normal name to be set when polling on a sticky queue
			return errUseVersioningWithoutNormalName
		}
	}
	if id.GetUseVersioning() && len(id.GetBuildId()) == 0 {
		return errUseVersioningWithoutBuildId
	}
	if len(id.GetBuildId()) > wh.config.WorkerBuildIdSizeLimit() {
		return errBuildIdTooLong
	}

	return wh.validateDeploymentOptions(deploymentOptions)
}

func (wh *WorkflowHandler) validateDeploymentOptions(deploymentOptions *deploymentpb.WorkerDeploymentOptions) error {
	if deploymentOptions == nil {
		return nil
	}
	if deploymentOptions.GetWorkerVersioningMode() != enumspb.WORKER_VERSIONING_MODE_VERSIONED {
		return nil // both deployment name and build ID fields are optional for unversioned workers
	}
	if deploymentOptions.GetDeploymentName() == "" || deploymentOptions.GetBuildId() == "" {
		return errDeploymentOptionsNotSet
	}
	deploymentVersion := &deploymentspb.WorkerDeploymentVersion{
		DeploymentName: deploymentOptions.GetDeploymentName(),
		BuildId:        deploymentOptions.GetBuildId(),
	}
	return worker_versioning.ValidateDeploymentVersion(deploymentVersion, wh.config.MaxIDLengthLimit())
}

//nolint:revive // cyclomatic complexity
func (wh *WorkflowHandler) validateBuildIdCompatibilityUpdate(
	req *workflowservice.UpdateWorkerBuildIdCompatibilityRequest,
) error {
	errDeets := []string{"request to update worker build ID compatability requires: "}

	checkIdLen := func(id string) {
		if len(id) > wh.config.WorkerBuildIdSizeLimit() {
			errDeets = append(errDeets, fmt.Sprintf(" Worker build IDs to be no larger than %v characters",
				wh.config.WorkerBuildIdSizeLimit()))
		}
	}

	if req.GetNamespace() == "" {
		errDeets = append(errDeets, "`namespace` to be set")
	}
	if req.GetTaskQueue() == "" {
		errDeets = append(errDeets, "`task_queue` to be set")
	}
	if req.GetOperation() == nil {
		errDeets = append(errDeets, "an operation to be specified")
	}
	if op, ok := req.GetOperation().(*workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleBuildId); ok {
		if op.AddNewCompatibleBuildId.GetNewBuildId() == "" {
			errDeets = append(errDeets, "`add_new_compatible_version` to be set")
		} else {
			checkIdLen(op.AddNewCompatibleBuildId.GetNewBuildId())
		}
		if op.AddNewCompatibleBuildId.GetExistingCompatibleBuildId() == "" {
			errDeets = append(errDeets, "`existing_compatible_version` to be set")
		}
	} else if op, ok := req.GetOperation().(*workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewBuildIdInNewDefaultSet); ok {
		if op.AddNewBuildIdInNewDefaultSet == "" {
			errDeets = append(errDeets, "`add_new_version_id_in_new_default_set` to be set")
		} else {
			checkIdLen(op.AddNewBuildIdInNewDefaultSet)
		}
	} else if op, ok := req.GetOperation().(*workflowservice.UpdateWorkerBuildIdCompatibilityRequest_PromoteSetByBuildId); ok {
		if op.PromoteSetByBuildId == "" {
			errDeets = append(errDeets, "`promote_set_by_version_id` to be set")
		} else {
			checkIdLen(op.PromoteSetByBuildId)
		}
	} else if op, ok := req.GetOperation().(*workflowservice.UpdateWorkerBuildIdCompatibilityRequest_PromoteBuildIdWithinSet); ok {
		if op.PromoteBuildIdWithinSet == "" {
			errDeets = append(errDeets, "`promote_version_id_within_set` to be set")
		} else {
			checkIdLen(op.PromoteBuildIdWithinSet)
		}
	}
	if len(errDeets) > 1 {
		return serviceerror.NewInvalidArgument(strings.Join(errDeets, ", "))
	}
	return nil
}

func (wh *WorkflowHandler) historyArchived(ctx context.Context, request *workflowservice.GetWorkflowExecutionHistoryRequest, namespaceID namespace.ID) bool {
	if request.GetExecution() == nil || request.GetExecution().GetRunId() == "" {
		return false
	}
	getMutableStateRequest := &historyservice.GetMutableStateRequest{
		NamespaceId: namespaceID.String(),
		Execution:   request.Execution,
	}
	_, err := wh.historyClient.GetMutableState(ctx, getMutableStateRequest)
	if err == nil {
		return false
	}
	switch err.(type) {
	case *serviceerror.NotFound:
		// the only case in which history is assumed to be archived is if getting mutable state returns entity not found error
		return true
	}

	return false
}

func (wh *WorkflowHandler) getArchivedHistory(
	ctx context.Context,
	request *workflowservice.GetWorkflowExecutionHistoryRequest,
	namespaceID namespace.ID,
) (*workflowservice.GetWorkflowExecutionHistoryResponse, error) {
	entry, err := wh.namespaceRegistry.GetNamespaceByID(namespaceID)
	if err != nil {
		return nil, err
	}

	URIString := entry.HistoryArchivalState().URI
	if URIString == "" {
		// if URI is empty, it means the namespace has never enabled for archival.
		// the error is not "workflow has passed retention period", because
		// we have no way to tell if the requested workflow exists or not.
		return nil, errHistoryNotFound
	}

	URI, err := archiver.NewURI(URIString)
	if err != nil {
		return nil, err
	}

	historyArchiver, err := wh.archiverProvider.GetHistoryArchiver(URI.Scheme())
	if err != nil {
		return nil, err
	}

	resp, err := historyArchiver.Get(ctx, URI, &archiver.GetHistoryRequest{
		NamespaceID:   namespaceID.String(),
		WorkflowID:    request.GetExecution().GetWorkflowId(),
		RunID:         request.GetExecution().GetRunId(),
		NextPageToken: request.GetNextPageToken(),
		PageSize:      int(request.GetMaximumPageSize()),
	})
	if err != nil {
		return nil, err
	}

	history := &historypb.History{}
	for _, batch := range resp.HistoryBatches {
		history.Events = append(history.Events, batch.Events...)
	}
	return &workflowservice.GetWorkflowExecutionHistoryResponse{
		History:       history,
		NextPageToken: resp.NextPageToken,
		Archived:      true,
	}, nil
}

// cancelOutstandingPoll cancel outstanding poll if context was canceled and returns true. Otherwise returns false.
func (wh *WorkflowHandler) cancelOutstandingPoll(
	ctx context.Context,
	namespaceID namespace.ID,
	taskQueueType enumspb.TaskQueueType,
	taskQueue *taskqueuepb.TaskQueue,
	pollerID string,
) bool {
	// First check if this err is due to context cancellation.  This means client connection to frontend is closed.
	if !errors.Is(ctx.Err(), context.Canceled) {
		return false
	}
	// Our rpc stack does not propagates context cancellation to the other service.  Lets make an explicit
	// call to matching to notify this poller is gone to prevent any tasks being dispatched to zombie pollers.
	// TODO: specify a reasonable timeout for CancelOutstandingPoll.
	_, err := wh.matchingClient.CancelOutstandingPoll(
		rpc.CopyContextValues(context.TODO(), ctx),
		&matchingservice.CancelOutstandingPollRequest{
			NamespaceId:   namespaceID.String(),
			TaskQueueType: taskQueueType,
			TaskQueue:     taskQueue,
			PollerId:      pollerID,
		},
	)
	// We can not do much if this call fails.  Just log the error and move on.
	if err != nil {
		wh.logger.Warn("Failed to cancel outstanding poller.",
			tag.WorkflowTaskQueueName(taskQueue.GetName()), tag.Error(err))
	}

	return true
}

func (wh *WorkflowHandler) registerOutstandingPollContext(
	ctx context.Context,
	pollerID string,
	namespaceID string,
) context.Context {

	if pollerID != "" {
		nsPollers, ok := wh.outstandingPollers.Get(namespaceID)
		if !ok {
			nsPollers, _ = wh.outstandingPollers.GetOrSet(namespaceID, collection.NewSyncMap[string, context.CancelFunc]())
		}
		childCtx, cancel := context.WithCancel(ctx)
		nsPollers.Set(pollerID, cancel)
		return childCtx
	}
	return ctx
}

func (wh *WorkflowHandler) unregisterOutstandingPollContext(
	pollerID string,
	namespaceID string,
) {
	nsPollers, ok := wh.outstandingPollers.Get(namespaceID)
	if ok {
		if cancel, exist := nsPollers.Pop(pollerID); exist {
			cancel()
		}
	}
}

func (wh *WorkflowHandler) checkBadBinary(namespaceEntry *namespace.Namespace, binaryChecksum string) error {
	if err := namespaceEntry.VerifyBinaryChecksum(binaryChecksum); err != nil {
		return serviceerror.NewInvalidArgumentf("Binary %v already marked as bad deployment.", binaryChecksum)
	}
	return nil
}

func validateRequestId(requestID *string, lenLimit int) error {
	if requestID == nil {
		// should never happen, but just in case.
		return serviceerror.NewInvalidArgument("RequestId is nil")
	}
	if *requestID == "" {
		// For easy direct API use, we default the request ID here but expect all
		// SDKs and other auto-retrying clients to set it
		*requestID = uuid.NewString()
	}

	if len(*requestID) > lenLimit {
		return errRequestIDTooLong
	}

	return nil
}

func (wh *WorkflowHandler) metricsScope(ctx context.Context) metrics.Handler {
	return interceptor.GetMetricsHandlerFromContext(ctx, wh.logger)
}

func (wh *WorkflowHandler) validateNamespace(
	namespace string,
) error {
	if len(namespace) > wh.config.MaxIDLengthLimit() {
		return errNamespaceTooLong
	}
	return nil
}

func (wh *WorkflowHandler) validateWorkflowID(
	workflowID string,
) error {
	if workflowID == "" {
		return errWorkflowIDNotSet
	}
	if len(workflowID) > wh.config.MaxIDLengthLimit() {
		return errWorkflowIDTooLong
	}
	return nil
}

// validateSchedulePayloadSize validates the blob size of the schedule's memo
// and action input. This runs for both CHASM and V1 paths.
func (wh *WorkflowHandler) validateSchedulePayloadSize(
	ctx context.Context,
	namespaceName string,
	scheduleID string,
	schedule *schedulepb.Schedule,
	memo *commonpb.Memo,
	operation string,
) error {
	action := schedule.GetAction().GetStartWorkflow()
	if action == nil {
		return serviceerror.NewInvalidArgument("Only StartWorkflow action is supported for schedules")
	}
	payloadSize := memo.Size() + action.GetInput().Size()
	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)
	return common.CheckEventBlobSizeLimit(
		payloadSize,
		sizeLimitWarn,
		sizeLimitError,
		namespaceName,
		scheduleID,
		"",
		wh.metricsScope(ctx).WithTags(metrics.CommandTypeTag(enumspb.COMMAND_TYPE_UNSPECIFIED.String())),
		wh.throttledLogger,
		operation,
	)
}

func (wh *WorkflowHandler) canonicalizeScheduleSpec(schedule *schedulepb.Schedule) error {
	if schedule.Spec == nil {
		schedule.Spec = &schedulepb.ScheduleSpec{}
	}
	compiledSpec, err := wh.scheduleSpecBuilder.NewCompiledSpec(schedule.Spec)
	if err != nil {
		return serviceerror.NewInvalidArgumentf("Invalid schedule spec: %v", err)
	}
	// This mutates a part of the request message, but it's safe even in the presence of
	// retries (reusing the same message) because canonicalization is idempotent.
	schedule.Spec = compiledSpec.CanonicalForm()
	return nil
}

func (wh *WorkflowHandler) decodeScheduleListInfo(memo *commonpb.Memo) *schedulepb.ScheduleListInfo {
	var listInfo schedulepb.ScheduleListInfo
	var listInfoBytes []byte
	if p := memo.GetFields()[scheduler.MemoFieldInfo]; p == nil {
		return nil
	} else if err := payload.Decode(p, &listInfoBytes); err != nil {
		wh.logger.Error("decoding schedule list info from payload", tag.Error(err))
		return nil
	} else if err := listInfo.Unmarshal(listInfoBytes); err != nil {
		wh.logger.Error("decoding schedule list info from payload", tag.Error(err))
		return nil
	}
	scheduler.CleanSpec(listInfo.Spec)
	return &listInfo
}

// This mutates searchAttributes
func (wh *WorkflowHandler) cleanScheduleSearchAttributes(searchAttributes *commonpb.SearchAttributes) *commonpb.SearchAttributes {
	fields := searchAttributes.GetIndexedFields()
	if len(fields) == 0 {
		return nil
	}

	delete(fields, sadefs.TemporalSchedulePaused)
	delete(fields, "TemporalScheduleInfoJSON") // used by older version, clean this up if present
	// these aren't schedule-related but they aren't relevant to the user for
	// scheduler workflows since it's the server worker
	delete(fields, sadefs.BinaryChecksums)
	delete(fields, sadefs.BuildIds)
	// all schedule workflows should be in this namespace division so there's no need to include it
	delete(fields, sadefs.TemporalNamespaceDivision)

	if len(fields) == 0 {
		return nil
	}
	return searchAttributes
}

// This mutates memo
func (wh *WorkflowHandler) cleanScheduleMemo(memo *commonpb.Memo) *commonpb.Memo {
	fields := memo.GetFields()
	if len(fields) == 0 {
		return nil
	}
	delete(fields, scheduler.MemoFieldInfo)
	if len(fields) == 0 {
		return nil
	}
	return memo
}

// This mutates request (but idempotent so safe for retries)
func (wh *WorkflowHandler) addInitialScheduleMemo(request *workflowservice.CreateScheduleRequest, args *schedulespb.StartScheduleArgs) {
	info := scheduler.GetListInfoFromStartArgs(args, time.Now().UTC(), wh.scheduleSpecBuilder)
	infoBytes, err := info.Marshal()
	if err != nil {
		wh.logger.Error("encoding initial schedule memo failed", tag.Error(err))
		return
	}
	p, err := sdk.PreferProtoDataConverter.ToPayload(infoBytes)
	if err != nil {
		wh.logger.Error("encoding initial schedule memo failed", tag.Error(err))
		return
	}
	if request.Memo == nil {
		request.Memo = &commonpb.Memo{}
	}
	if request.Memo.Fields == nil {
		request.Memo.Fields = make(map[string]*commonpb.Payload)
	}
	request.Memo.Fields[scheduler.MemoFieldInfo] = p
}

func getBatchOperationState(workflowState enumspb.WorkflowExecutionStatus) enumspb.BatchOperationState {
	var operationState enumspb.BatchOperationState
	switch workflowState {
	case enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING:
		operationState = enumspb.BATCH_OPERATION_STATE_RUNNING
	case enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED:
		operationState = enumspb.BATCH_OPERATION_STATE_COMPLETED
	default:
		operationState = enumspb.BATCH_OPERATION_STATE_FAILED
	}
	return operationState
}

func (wh *WorkflowHandler) UpdateWorkflowExecutionOptions(
	ctx context.Context,
	request *workflowservice.UpdateWorkflowExecutionOptionsRequest,
) (_ *workflowservice.UpdateWorkflowExecutionOptionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)
	wh.logger.Debug("Received UpdateWorkflowExecutionOptions for ", tag.WorkflowID(request.GetWorkflowExecution().GetWorkflowId()))
	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetWorkflowExecution().GetWorkflowId() == "" {
		return nil, errWorkflowIDNotSet
	}
	if request.GetUpdateMask() == nil {
		return nil, serviceerror.NewInvalidArgument("UpdateMask is required")
	}
	opts := request.GetWorkflowExecutionOptions()
	if opts == nil {
		return nil, serviceerror.NewInvalidArgument("WorkflowExecutionOptions is required")
	}
	_, err := fieldmaskpb.New(opts, request.GetUpdateMask().GetPaths()...) // errors if paths are not valid for WorkflowExecutionOptions
	if err != nil {
		return nil, serviceerror.NewInvalidArgumentf("error parsing UpdateMask: %s", err.Error())
	}
	if err := priorities.Validate(opts.GetPriority()); err != nil {
		return nil, err
	}
	if err := wh.validateTimeSkippingConfig(opts.GetTimeSkippingConfig(), namespace.Name(request.GetNamespace())); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	response, err := wh.historyClient.UpdateWorkflowExecutionOptions(ctx, &historyservice.UpdateWorkflowExecutionOptionsRequest{
		NamespaceId:   namespaceID.String(),
		UpdateRequest: request,
	})
	if err != nil {
		return nil, err
	}

	return &workflowservice.UpdateWorkflowExecutionOptionsResponse{
		WorkflowExecutionOptions: response.WorkflowExecutionOptions,
	}, nil
}

func (wh *WorkflowHandler) UpdateActivityOptions(
	ctx context.Context,
	request *workflowservice.UpdateActivityOptionsRequest,
) (_ *workflowservice.UpdateActivityOptionsResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetExecution().GetWorkflowId() == "" {
		return nil, errWorkflowIDNotSet
	}
	if request.GetActivity() == nil {
		return nil, errActivityIDOrTypeNotSet
	}
	if err := priorities.Validate(request.GetActivityOptions().GetPriority()); err != nil {
		return nil, err
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	response, err := wh.historyClient.UpdateActivityOptions(ctx, &historyservice.UpdateActivityOptionsRequest{
		NamespaceId:   namespaceID.String(),
		UpdateRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.UpdateActivityOptionsResponse{
		ActivityOptions: response.ActivityOptions,
	}, nil
}

func (wh *WorkflowHandler) PauseActivity(
	ctx context.Context,
	request *workflowservice.PauseActivityRequest,
) (_ *workflowservice.PauseActivityResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetExecution().GetWorkflowId() == "" {
		return nil, errWorkflowIDNotSet
	}
	if request.GetActivity() == nil {
		return nil, errActivityIDOrTypeNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.PauseActivity(ctx, &historyservice.PauseActivityRequest{
		NamespaceId:     namespaceID.String(),
		FrontendRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.PauseActivityResponse{}, nil
}

func (wh *WorkflowHandler) UnpauseActivity(
	ctx context.Context, request *workflowservice.UnpauseActivityRequest,
) (_ *workflowservice.UnpauseActivityResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetExecution().GetWorkflowId() == "" {
		return nil, errWorkflowIDNotSet
	}
	if request.GetActivity() == nil {
		return nil, errActivityIDOrTypeNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.UnpauseActivity(ctx, &historyservice.UnpauseActivityRequest{
		NamespaceId:     namespaceID.String(),
		FrontendRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.UnpauseActivityResponse{}, nil
}

func (wh *WorkflowHandler) ResetActivity(
	ctx context.Context, request *workflowservice.ResetActivityRequest,
) (_ *workflowservice.ResetActivityResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetExecution().GetWorkflowId() == "" {
		return nil, errWorkflowIDNotSet
	}
	if request.GetActivity() == nil {
		return nil, errActivityIDOrTypeNotSet
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, err = wh.historyClient.ResetActivity(ctx, &historyservice.ResetActivityRequest{
		NamespaceId:     namespaceID.String(),
		FrontendRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.ResetActivityResponse{}, nil
}

func (wh *WorkflowHandler) CreateWorkflowRule(
	ctx context.Context,
	request *workflowservice.CreateWorkflowRuleRequest,
) (_ *workflowservice.CreateWorkflowRuleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if !wh.config.WorkflowRulesAPIsEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplemented("method CreateWorkflowRule not supported")
	}

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetSpec() == nil {
		return nil, serviceerror.NewInvalidArgument("Rule Specification is not set.")
	}

	if request.GetSpec().GetId() == "" {
		return nil, serviceerror.NewInvalidArgument("Workflow Rule ID is not set.")
	}

	if len(request.GetSpec().GetId()) > wh.config.MaxIDLengthLimit() {
		return nil, errWorkflowRuleIDTooLong
	}

	rule, err := wh.namespaceHandler.CreateWorkflowRule(ctx,
		request.GetSpec(),
		request.GetIdentity(),
		request.GetDescription(),
		request.GetNamespace())
	if err != nil {
		return nil, err
	}

	response := &workflowservice.CreateWorkflowRuleResponse{
		Rule: rule,
	}

	return response, nil
}

func (wh *WorkflowHandler) DescribeWorkflowRule(
	ctx context.Context,
	request *workflowservice.DescribeWorkflowRuleRequest,
) (_ *workflowservice.DescribeWorkflowRuleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if !wh.config.WorkflowRulesAPIsEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplemented("method DescribeWorkflowRule not supported")
	}

	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetRuleId() == "" {
		return nil, serviceerror.NewInvalidArgument("Workflow Rule ID is not set.")
	}

	rule, err := wh.namespaceHandler.DescribeWorkflowRule(ctx, request.GetRuleId(), request.GetNamespace())
	if err != nil {
		return nil, err
	}

	return &workflowservice.DescribeWorkflowRuleResponse{Rule: rule}, nil
}

func (wh *WorkflowHandler) DeleteWorkflowRule(
	ctx context.Context,
	request *workflowservice.DeleteWorkflowRuleRequest,
) (_ *workflowservice.DeleteWorkflowRuleResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if !wh.config.WorkflowRulesAPIsEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplemented("method DeleteWorkflowRule not supported")
	}
	if request == nil {
		return nil, errRequestNotSet
	}
	if request.GetRuleId() == "" {
		return nil, serviceerror.NewInvalidArgument("Workflow Rule ID is not set.")
	}

	err := wh.namespaceHandler.DeleteWorkflowRule(ctx, request.GetRuleId(), request.GetNamespace())
	if err != nil {
		return nil, err
	}

	return &workflowservice.DeleteWorkflowRuleResponse{}, nil
}

func (wh *WorkflowHandler) ListWorkflowRules(
	ctx context.Context,
	request *workflowservice.ListWorkflowRulesRequest,
) (_ *workflowservice.ListWorkflowRulesResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if !wh.config.WorkflowRulesAPIsEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplemented("method ListWorkflowRules not supported")
	}

	if request == nil {
		return nil, errRequestNotSet
	}

	workflowRules, err := wh.namespaceHandler.ListWorkflowRules(ctx, request.GetNamespace())
	if err != nil {
		return nil, err
	}
	return &workflowservice.ListWorkflowRulesResponse{Rules: workflowRules}, nil
}

// RecordWorkerHeartbeat receive heartbeat request from the worker
// and forwards it to the corresponding matching service.
func (wh *WorkflowHandler) RecordWorkerHeartbeat(
	ctx context.Context, request *workflowservice.RecordWorkerHeartbeatRequest,
) (*workflowservice.RecordWorkerHeartbeatResponse, error) {
	if !wh.config.WorkerHeartbeatsEnabled(request.GetNamespace()) {
		return &workflowservice.RecordWorkerHeartbeatResponse{}, nil
	}
	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	_, err = wh.matchingClient.RecordWorkerHeartbeat(ctx, &matchingservice.RecordWorkerHeartbeatRequest{
		NamespaceId:       namespaceID.String(),
		HeartbeartRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.RecordWorkerHeartbeatResponse{}, nil
}

// ListWorkers retrieves a list of workers in the specified namespace that match the provided filters.
func (wh *WorkflowHandler) ListWorkers(
	ctx context.Context, request *workflowservice.ListWorkersRequest,
) (*workflowservice.ListWorkersResponse, error) {
	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	resp, err := wh.matchingClient.ListWorkers(ctx, &matchingservice.ListWorkersRequest{
		NamespaceId: namespaceID.String(),
		ListRequest: request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.ListWorkersResponse{
		WorkersInfo:   resp.GetWorkersInfo(),
		Workers:       resp.GetWorkers(),
		NextPageToken: resp.GetNextPageToken(),
	}, nil
}

func (wh *WorkflowHandler) UpdateTaskQueueConfig(
	ctx context.Context, request *workflowservice.UpdateTaskQueueConfigRequest,
) (*workflowservice.UpdateTaskQueueConfigResponse, error) {
	if request == nil {
		return nil, errRequestNotSet
	}
	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	// Validate rate limits
	queueRateLimit := request.GetUpdateQueueRateLimit()
	if queueRateLimit.GetRateLimit() != nil && request.TaskQueueType == enumspb.TASK_QUEUE_TYPE_WORKFLOW {
		return nil, serviceerror.NewInvalidArgument("Setting rate limit on workflow task queues is not allowed.")
	}
	if err := validateRateLimit(queueRateLimit, "UpdateQueueRateLimit"); err != nil {
		return nil, err
	}
	fairnessKeyRateLimitDefault := request.GetUpdateFairnessKeyRateLimitDefault()
	if fairnessKeyRateLimitDefault.GetRateLimit() != nil && request.TaskQueueType == enumspb.TASK_QUEUE_TYPE_WORKFLOW {
		return nil, serviceerror.NewInvalidArgument("Setting fairness key rate limit on workflow task queues is not allowed.")
	}
	if err := validateRateLimit(fairnessKeyRateLimitDefault, "UpdateFairnessKeyRateLimitDefault"); err != nil {
		return nil, err
	}

	// Validate identity field
	if err := validateStringField("Identity", request.GetIdentity(), wh.config.MaxIDLengthLimit(), false); err != nil {
		return nil, err
	}

	// Validate Fairness Weight Updates
	setFairnessWeightOverrides := request.GetSetFairnessWeightOverrides()
	unsetFairnessWeightOverrides := request.GetUnsetFairnessWeightOverrides()
	limit := wh.config.MaxFairnessWeightOverrideConfigLimit(request.GetNamespace(), request.TaskQueue, request.TaskQueueType)
	if err := validateFairnessWeightUpdate(setFairnessWeightOverrides, unsetFairnessWeightOverrides, limit); err != nil {
		return nil, err
	}

	resp, err := wh.matchingClient.UpdateTaskQueueConfig(ctx, &matchingservice.UpdateTaskQueueConfigRequest{
		NamespaceId:           namespaceID.String(),
		UpdateTaskqueueConfig: request,
	})
	if err != nil {
		return nil, err
	}
	return &workflowservice.UpdateTaskQueueConfigResponse{
		Config: resp.UpdatedTaskqueueConfig,
	}, nil
}

func (wh *WorkflowHandler) FetchWorkerConfig(_ context.Context, request *workflowservice.FetchWorkerConfigRequest,
) (*workflowservice.FetchWorkerConfigResponse, error) {
	if !wh.config.WorkerCommandsEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplemented("FetchWorkerConfig command is not enabled.")
	}
	return nil, serviceerror.NewUnimplemented("FetchWorkerConfig command is not enabled.")
}

func (wh *WorkflowHandler) UpdateWorkerConfig(_ context.Context, request *workflowservice.UpdateWorkerConfigRequest,
) (*workflowservice.UpdateWorkerConfigResponse, error) {
	if !wh.config.WorkerCommandsEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplemented("UpdateWorkerConfig command is not enabled.")
	}
	if request == nil {
		return nil, errRequestNotSet
	}

	if request.GetWorkerConfig() == nil {
		return nil, serviceerror.NewInvalidArgument("WorkerConfig is not set")
	}

	_, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	return nil, serviceerror.NewUnimplemented("UpdateWorkerConfig command is not enabled.")
}

func (wh *WorkflowHandler) DescribeWorker(ctx context.Context, request *workflowservice.DescribeWorkerRequest,
) (*workflowservice.DescribeWorkerResponse, error) {
	namespaceName := namespace.Name(request.GetNamespace())
	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
	if err != nil {
		return nil, err
	}

	resp, err := wh.matchingClient.DescribeWorker(ctx, &matchingservice.DescribeWorkerRequest{
		NamespaceId: namespaceID.String(),
		Request:     request,
	})

	if err != nil {
		return nil, err
	}

	return &workflowservice.DescribeWorkerResponse{
		WorkerInfo: resp.GetWorkerInfo(),
	}, nil
}

func (wh *WorkflowHandler) TriggerWorkflowRule(context.Context, *workflowservice.TriggerWorkflowRuleRequest) (*workflowservice.TriggerWorkflowRuleResponse, error) {
	return nil, serviceerror.NewUnimplemented("method TriggerWorkflowRule not supported")
}

// PauseWorkflowExecution pauses a workflow execution.
func (wh *WorkflowHandler) PauseWorkflowExecution(ctx context.Context, request *workflowservice.PauseWorkflowExecutionRequest) (_ *workflowservice.PauseWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	if !wh.config.WorkflowPauseEnabled(request.GetNamespace()) {
		return nil, serviceerror.NewUnimplementedf("workflow pause is not enabled for namespace: %s", request.GetNamespace())
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, historyErr := wh.historyClient.PauseWorkflowExecution(ctx, &historyservice.PauseWorkflowExecutionRequest{
		NamespaceId:  namespaceID.String(),
		PauseRequest: request,
	})
	if historyErr != nil {
		return nil, historyErr
	}

	return &workflowservice.PauseWorkflowExecutionResponse{}, nil
}

func (wh *WorkflowHandler) UnpauseWorkflowExecution(ctx context.Context, request *workflowservice.UnpauseWorkflowExecutionRequest) (_ *workflowservice.UnpauseWorkflowExecutionResponse, retError error) {
	defer log.CapturePanic(wh.logger, &retError)

	if request == nil {
		return nil, errRequestNotSet
	}

	// verify size limits of reason, request id and identity.
	if len(request.GetReason()) > wh.config.MaxIDLengthLimit() {
		return nil, serviceerror.NewInvalidArgument("reason is too long.")
	}
	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
		return nil, serviceerror.NewInvalidArgument("request id is too long.")
	}
	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
		return nil, serviceerror.NewInvalidArgument("identity is too long.")
	}

	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
	if err != nil {
		return nil, err
	}

	_, historyErr := wh.historyClient.UnpauseWorkflowExecution(ctx, &historyservice.UnpauseWorkflowExecutionRequest{
		NamespaceId:    namespaceID.String(),
		UnpauseRequest: request,
	})
	if historyErr != nil {
		return nil, historyErr
	}

	return &workflowservice.UnpauseWorkflowExecutionResponse{}, nil
}

func (wh *WorkflowHandler) PauseActivityExecution(context.Context, *workflowservice.PauseActivityExecutionRequest) (*workflowservice.PauseActivityExecutionResponse, error) {
	return nil, serviceerror.NewUnimplemented("PauseActivityExecution not implemented")
}

func (wh *WorkflowHandler) ResetActivityExecution(context.Context, *workflowservice.ResetActivityExecutionRequest) (*workflowservice.ResetActivityExecutionResponse, error) {
	return nil, serviceerror.NewUnimplemented("ResetActivityExecution not implemented")
}

func (wh *WorkflowHandler) UnpauseActivityExecution(context.Context, *workflowservice.UnpauseActivityExecutionRequest) (*workflowservice.UnpauseActivityExecutionResponse, error) {
	return nil, serviceerror.NewUnimplemented("UnpauseActivityExecution not implemented")
}

func (wh *WorkflowHandler) UpdateActivityExecutionOptions(context.Context, *workflowservice.UpdateActivityExecutionOptionsRequest) (*workflowservice.UpdateActivityExecutionOptionsResponse, error) {
	return nil, serviceerror.NewUnimplemented("UpdateActivityExecutionOptions not implemented")
}