improvement: bypass middle-man-struct for batch operations (#8081)

## What changed?
Remove `BatchParams` to safely serialize proto -> json -> proto and add
a `BatchOperation` proto definition

## Why?
Remove custom serialization of proto for json safety.

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

## Potential risks
Could break batching backwards compatibility but tests should catch this

api/batch/v1/request_response.go-helpers.pb.go

@@ -0,0 +1,43 @@
+// Code generated by protoc-gen-go-helpers. DO NOT EDIT.
+package batch
+
+import (
+	"google.golang.org/protobuf/proto"
+)
+
+// Marshal an object of type BatchOperationInput to the protobuf v3 wire format
+func (val *BatchOperationInput) Marshal() ([]byte, error) {
+	return proto.Marshal(val)
+}
+
+// Unmarshal an object of type BatchOperationInput from the protobuf v3 wire format
+func (val *BatchOperationInput) Unmarshal(buf []byte) error {
+	return proto.Unmarshal(buf, val)
+}
+
+// Size returns the size of the object, in bytes, once serialized
+func (val *BatchOperationInput) Size() int {
+	return proto.Size(val)
+}
+
+// Equal returns whether two BatchOperationInput values are equivalent by recursively
+// comparing the message's fields.
+// For more information see the documentation for
+// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
+func (this *BatchOperationInput) Equal(that interface{}) bool {
+	if that == nil {
+		return this == nil
+	}
+
+	var that1 *BatchOperationInput
+	switch t := that.(type) {
+	case *BatchOperationInput:
+		that1 = t
+	case BatchOperationInput:
+		that1 = &t
+	default:
+		return false
+	}
+
+	return proto.Equal(this, that1)
+}

api/batch/v1/request_response.pb.go

@@ -0,0 +1,188 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// plugins:
+// 	protoc-gen-go
+// 	protoc
+// source: temporal/server/api/batch/v1/request_response.proto
+
+package batch
+
+import (
+	reflect "reflect"
+	sync "sync"
+	unsafe "unsafe"
+
+	v1 "go.temporal.io/api/enums/v1"
+	v11 "go.temporal.io/api/workflowservice/v1"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	durationpb "google.golang.org/protobuf/types/known/durationpb"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type BatchOperationInput struct {
+	state                    protoimpl.MessageState `protogen:"open.v1"`
+	NamespaceId              string                 `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
+	Concurrency              int64                  `protobuf:"varint,2,opt,name=concurrency,proto3" json:"concurrency,omitempty"`
+	AttemptsOnRetryableError int64                  `protobuf:"varint,3,opt,name=attempts_on_retryable_error,json=attemptsOnRetryableError,proto3" json:"attempts_on_retryable_error,omitempty"`
+	ActivityHeartbeatTimeout *durationpb.Duration   `protobuf:"bytes,4,opt,name=activity_heartbeat_timeout,json=activityHeartbeatTimeout,proto3" json:"activity_heartbeat_timeout,omitempty"`
+	NonRetryableErrors       []string               `protobuf:"bytes,5,rep,name=non_retryable_errors,json=nonRetryableErrors,proto3" json:"non_retryable_errors,omitempty"`
+	BatchType                v1.BatchOperationType  `protobuf:"varint,6,opt,name=batch_type,json=batchType,proto3,enum=temporal.api.enums.v1.BatchOperationType" json:"batch_type,omitempty"`
+	// The request to start the batch operation.
+	Request       *v11.StartBatchOperationRequest `protobuf:"bytes,7,opt,name=request,proto3" json:"request,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *BatchOperationInput) Reset() {
+	*x = BatchOperationInput{}
+	mi := &file_temporal_server_api_batch_v1_request_response_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *BatchOperationInput) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*BatchOperationInput) ProtoMessage() {}
+
+func (x *BatchOperationInput) ProtoReflect() protoreflect.Message {
+	mi := &file_temporal_server_api_batch_v1_request_response_proto_msgTypes[0]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use BatchOperationInput.ProtoReflect.Descriptor instead.
+func (*BatchOperationInput) Descriptor() ([]byte, []int) {
+	return file_temporal_server_api_batch_v1_request_response_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *BatchOperationInput) GetNamespaceId() string {
+	if x != nil {
+		return x.NamespaceId
+	}
+	return ""
+}
+
+func (x *BatchOperationInput) GetConcurrency() int64 {
+	if x != nil {
+		return x.Concurrency
+	}
+	return 0
+}
+
+func (x *BatchOperationInput) GetAttemptsOnRetryableError() int64 {
+	if x != nil {
+		return x.AttemptsOnRetryableError
+	}
+	return 0
+}
+
+func (x *BatchOperationInput) GetActivityHeartbeatTimeout() *durationpb.Duration {
+	if x != nil {
+		return x.ActivityHeartbeatTimeout
+	}
+	return nil
+}
+
+func (x *BatchOperationInput) GetNonRetryableErrors() []string {
+	if x != nil {
+		return x.NonRetryableErrors
+	}
+	return nil
+}
+
+func (x *BatchOperationInput) GetBatchType() v1.BatchOperationType {
+	if x != nil {
+		return x.BatchType
+	}
+	return v1.BatchOperationType(0)
+}
+
+func (x *BatchOperationInput) GetRequest() *v11.StartBatchOperationRequest {
+	if x != nil {
+		return x.Request
+	}
+	return nil
+}
+
+var File_temporal_server_api_batch_v1_request_response_proto protoreflect.FileDescriptor
+
+const file_temporal_server_api_batch_v1_request_response_proto_rawDesc = "" +
+	"\n" +
+	"3temporal/server/api/batch/v1/request_response.proto\x12\x1ctemporal.server.api.batch.v1\x1a6temporal/api/workflowservice/v1/request_response.proto\x1a+temporal/api/enums/v1/batch_operation.proto\x1a\x1egoogle/protobuf/duration.proto\"\xc5\x03\n" +
+	"\x13BatchOperationInput\x12!\n" +
+	"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12 \n" +
+	"\vconcurrency\x18\x02 \x01(\x03R\vconcurrency\x12=\n" +
+	"\x1battempts_on_retryable_error\x18\x03 \x01(\x03R\x18attemptsOnRetryableError\x12W\n" +
+	"\x1aactivity_heartbeat_timeout\x18\x04 \x01(\v2\x19.google.protobuf.DurationR\x18activityHeartbeatTimeout\x120\n" +
+	"\x14non_retryable_errors\x18\x05 \x03(\tR\x12nonRetryableErrors\x12H\n" +
+	"\n" +
+	"batch_type\x18\x06 \x01(\x0e2).temporal.api.enums.v1.BatchOperationTypeR\tbatchType\x12U\n" +
+	"\arequest\x18\a \x01(\v2;.temporal.api.workflowservice.v1.StartBatchOperationRequestR\arequestB*Z(go.temporal.io/server/api/batch/v1;batchb\x06proto3"
+
+var (
+	file_temporal_server_api_batch_v1_request_response_proto_rawDescOnce sync.Once
+	file_temporal_server_api_batch_v1_request_response_proto_rawDescData []byte
+)
+
+func file_temporal_server_api_batch_v1_request_response_proto_rawDescGZIP() []byte {
+	file_temporal_server_api_batch_v1_request_response_proto_rawDescOnce.Do(func() {
+		file_temporal_server_api_batch_v1_request_response_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_temporal_server_api_batch_v1_request_response_proto_rawDesc), len(file_temporal_server_api_batch_v1_request_response_proto_rawDesc)))
+	})
+	return file_temporal_server_api_batch_v1_request_response_proto_rawDescData
+}
+
+var file_temporal_server_api_batch_v1_request_response_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_temporal_server_api_batch_v1_request_response_proto_goTypes = []any{
+	(*BatchOperationInput)(nil),            // 0: temporal.server.api.batch.v1.BatchOperationInput
+	(*durationpb.Duration)(nil),            // 1: google.protobuf.Duration
+	(v1.BatchOperationType)(0),             // 2: temporal.api.enums.v1.BatchOperationType
+	(*v11.StartBatchOperationRequest)(nil), // 3: temporal.api.workflowservice.v1.StartBatchOperationRequest
+}
+var file_temporal_server_api_batch_v1_request_response_proto_depIdxs = []int32{
+	1, // 0: temporal.server.api.batch.v1.BatchOperationInput.activity_heartbeat_timeout:type_name -> google.protobuf.Duration
+	2, // 1: temporal.server.api.batch.v1.BatchOperationInput.batch_type:type_name -> temporal.api.enums.v1.BatchOperationType
+	3, // 2: temporal.server.api.batch.v1.BatchOperationInput.request:type_name -> temporal.api.workflowservice.v1.StartBatchOperationRequest
+	3, // [3:3] is the sub-list for method output_type
+	3, // [3:3] is the sub-list for method input_type
+	3, // [3:3] is the sub-list for extension type_name
+	3, // [3:3] is the sub-list for extension extendee
+	0, // [0:3] is the sub-list for field type_name
+}
+
+func init() { file_temporal_server_api_batch_v1_request_response_proto_init() }
+func file_temporal_server_api_batch_v1_request_response_proto_init() {
+	if File_temporal_server_api_batch_v1_request_response_proto != nil {
+		return
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_batch_v1_request_response_proto_rawDesc), len(file_temporal_server_api_batch_v1_request_response_proto_rawDesc)),
+			NumEnums:      0,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_temporal_server_api_batch_v1_request_response_proto_goTypes,
+		DependencyIndexes: file_temporal_server_api_batch_v1_request_response_proto_depIdxs,
+		MessageInfos:      file_temporal_server_api_batch_v1_request_response_proto_msgTypes,
+	}.Build()
+	File_temporal_server_api_batch_v1_request_response_proto = out.File
+	file_temporal_server_api_batch_v1_request_response_proto_goTypes = nil
+	file_temporal_server_api_batch_v1_request_response_proto_depIdxs = nil
+}

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

@@ -0,0 +1,25 @@
+syntax = "proto3";
+
+package temporal.server.api.batch.v1;
+option go_package = "go.temporal.io/server/api/batch/v1;batch";
+
+import "temporal/api/workflowservice/v1/request_response.proto";
+import "temporal/api/enums/v1/batch_operation.proto";
+import "google/protobuf/duration.proto";
+
+message BatchOperationInput { 
+  string namespace_id = 1; 
+  
+  int64 concurrency = 2;
+
+  int64 attempts_on_retryable_error = 3;
+
+  google.protobuf.Duration activity_heartbeat_timeout = 4;
+
+  repeated string non_retryable_errors = 5;
+
+  temporal.api.enums.v1.BatchOperationType batch_type = 6;
+
+  // The request to start the batch operation.
+  temporal.api.workflowservice.v1.StartBatchOperationRequest request = 7;
+}

service/frontend/workflow_handler.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/pborman/uuid"
+	"github.com/temporalio/sqlparser"
 	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
@@ -26,6 +27,7 @@ import (
 	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"
@@ -4462,26 +4464,8 @@ func (wh *WorkflowHandler) StartBatchOperation(
 		return nil, errRequestNotSet
 	}
 
-	if len(request.GetJobId()) == 0 {
-		return nil, errBatchJobIDNotSet
-	}
-	if len(request.Namespace) == 0 {
-		return nil, errNamespaceNotSet
-	}
-	if len(request.VisibilityQuery) == 0 && len(request.Executions) == 0 {
-		return nil, errBatchOpsWorkflowFilterNotSet
-	}
-	if len(request.VisibilityQuery) != 0 && len(request.Executions) != 0 {
-		return nil, errBatchOpsWorkflowFiltersNotAllowed
-	}
-	if len(request.Executions) > wh.config.MaxExecutionCountBatchOperation(request.Namespace) {
-		return nil, errBatchOpsMaxWorkflowExecutionCount
-	}
-	if len(request.Reason) == 0 {
-		return nil, errReasonNotSet
-	}
-	if request.Operation == nil {
-		return nil, errBatchOperationNotSet
+	if err := batcher.ValidateBatchOperation(request); err != nil {
+		return nil, err
 	}
 
 	if !wh.config.EnableBatcher(request.Namespace) {
@@ -4513,216 +4497,67 @@ func (wh *WorkflowHandler) StartBatchOperation(
 	if err != nil {
 		return nil, err
 	}
+
+	input := &batchspb.BatchOperationInput{
+		Request:     request,
+		NamespaceId: namespaceID.String(),
+	}
+
 	var identity string
-	var operationType string
-	var signalParams batcher.SignalParams
-	var resetParams batcher.ResetParams
-	var updateOptionsParams batcher.UpdateOptionsParams
-	var unpauseActivitiesParams batcher.UnpauseActivitiesParams
-	var resetActivitiesParams batcher.ResetActivitiesParams
-	var updateActivitiesOptionsParams batcher.UpdateActivitiesOptionsParams
 	switch op := request.Operation.(type) {
 	case *workflowservice.StartBatchOperationRequest_TerminationOperation:
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_TERMINATE
 		identity = op.TerminationOperation.GetIdentity()
-		operationType = batcher.BatchTypeTerminate
 	case *workflowservice.StartBatchOperationRequest_SignalOperation:
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_SIGNAL
 		identity = op.SignalOperation.GetIdentity()
-		operationType = batcher.BatchTypeSignal
-		signalParams.SignalName = op.SignalOperation.GetSignal()
-		signalParams.Input = op.SignalOperation.GetInput()
 	case *workflowservice.StartBatchOperationRequest_CancellationOperation:
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_CANCEL
 		identity = op.CancellationOperation.GetIdentity()
-		operationType = batcher.BatchTypeCancel
 	case *workflowservice.StartBatchOperationRequest_DeletionOperation:
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_DELETE
 		identity = op.DeletionOperation.GetIdentity()
-		operationType = batcher.BatchTypeDelete
 	case *workflowservice.StartBatchOperationRequest_ResetOperation:
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_RESET
 		identity = op.ResetOperation.GetIdentity()
-		operationType = batcher.BatchTypeReset
-		if op.ResetOperation.Options != nil {
-			if op.ResetOperation.Options.Target == nil {
-				return nil, serviceerror.NewInvalidArgument("batch reset missing target")
-			}
-			encodedResetOptions, err := op.ResetOperation.Options.Marshal()
-			if err != nil {
-				return nil, err
-			}
-			resetParams.ResetOptions = encodedResetOptions
-			resetParams.PostResetOperations = make([][]byte, len(op.ResetOperation.PostResetOperations))
-			for i, postResetOperation := range op.ResetOperation.PostResetOperations {
-				encodedPostResetOperations, err := postResetOperation.Marshal()
-				if err != nil {
-					return nil, err
-				}
-				resetParams.PostResetOperations[i] = encodedPostResetOperations
-			}
-		} else {
-			// TODO: remove support for old fields later
-			resetType := op.ResetOperation.GetResetType()
-			if _, ok := enumspb.ResetType_name[int32(resetType)]; !ok || resetType == enumspb.RESET_TYPE_UNSPECIFIED {
-				return nil, serviceerror.NewInvalidArgumentf("unknown batch reset type %v", resetType)
-			}
-			resetParams.ResetType = resetType
-			resetParams.ResetReapplyType = op.ResetOperation.GetResetReapplyType()
-		}
 	case *workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation:
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UPDATE_EXECUTION_OPTIONS
 		identity = op.UpdateWorkflowOptionsOperation.GetIdentity()
-		operationType = batcher.BatchTypeUpdateOptions
-		updateOptionsParams.WorkflowExecutionOptions = op.UpdateWorkflowOptionsOperation.GetWorkflowExecutionOptions()
-		if updateMask := op.UpdateWorkflowOptionsOperation.GetUpdateMask(); updateMask != nil {
-			updateOptionsParams.UpdateMask = &batcher.FieldMask{Paths: updateMask.Paths}
-		}
-		// TODO(carlydf): remove hacky usage of deprecated fields later, after adding support for oneof in BatchParams encoder
-		if o := updateOptionsParams.WorkflowExecutionOptions.VersioningOverride; o.GetOverride() != nil {
-			deprecatedOverride := &workflowpb.VersioningOverride{}
-			if o.GetAutoUpgrade() {
-				deprecatedOverride.Behavior = enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE //nolint:staticcheck // SA1019: worker versioning v0.31
-			} else if o.GetPinned().GetBehavior() == workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED {
-				deprecatedOverride.Behavior = enumspb.VERSIONING_BEHAVIOR_PINNED                                                            //nolint:staticcheck // SA1019: worker versioning v0.31
-				deprecatedOverride.PinnedVersion = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(o.GetPinned().GetVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31
-			}
-			updateOptionsParams.WorkflowExecutionOptions.VersioningOverride = deprecatedOverride
-		}
 	case *workflowservice.StartBatchOperationRequest_UnpauseActivitiesOperation:
-		operationType = batcher.BatchTypeUnpauseActivities
-		if op.UnpauseActivitiesOperation == nil {
-			return nil, serviceerror.NewInvalidArgument("unpause activities operation is not set")
-		}
-		if op.UnpauseActivitiesOperation.GetActivity() == nil {
-			return nil, serviceerror.NewInvalidArgument("activity filter must be set")
-		}
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UNPAUSE_ACTIVITY
+		identity = op.UnpauseActivitiesOperation.GetIdentity()
 
 		switch a := op.UnpauseActivitiesOperation.GetActivity().(type) {
 		case *batchpb.BatchOperationUnpauseActivities_Type:
-			if len(a.Type) == 0 {
-				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
-			}
-			unpauseCause := fmt.Sprintf("%s = 'property:activityType=%s'", searchattribute.TemporalPauseInfo, a.Type)
-			visibilityQuery = fmt.Sprintf("(%s) AND (%s)", visibilityQuery, unpauseCause)
-			unpauseActivitiesParams.ActivityType = a.Type
+			searchValue := fmt.Sprintf("property:activityType=%s", a.Type)
+			escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
+			input.Request.VisibilityQuery = fmt.Sprintf("%s = %s", searchattribute.TemporalPauseInfo, escapedSearchValue)
 		case *batchpb.BatchOperationUnpauseActivities_MatchAll:
-			if !a.MatchAll {
-				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
-			}
 			wildCardUnpause := fmt.Sprintf("%s STARTS_WITH 'property:activityType='", searchattribute.TemporalPauseInfo)
-			visibilityQuery = fmt.Sprintf("(%s) AND (%s)", visibilityQuery, wildCardUnpause)
-			unpauseActivitiesParams.MatchAll = true
+			input.Request.VisibilityQuery = fmt.Sprintf("(%s) AND (%s)", visibilityQuery, wildCardUnpause)
 		}
-
-		unpauseActivitiesParams.ResetAttempts = op.UnpauseActivitiesOperation.ResetAttempts
-		unpauseActivitiesParams.ResetHeartbeat = op.UnpauseActivitiesOperation.ResetHeartbeat
-		unpauseActivitiesParams.Jitter = op.UnpauseActivitiesOperation.Jitter.AsDuration()
-		unpauseActivitiesParams.Identity = op.UnpauseActivitiesOperation.GetIdentity()
 	case *workflowservice.StartBatchOperationRequest_ResetActivitiesOperation:
-		operationType = batcher.BatchTypeResetActivities
-		if op.ResetActivitiesOperation == nil {
-			return nil, serviceerror.NewInvalidArgument("reset activities operation is not set")
-		}
-		if op.ResetActivitiesOperation.GetActivity() == nil {
-			return nil, serviceerror.NewInvalidArgument("activity filter must be set")
-		}
-
-		switch a := op.ResetActivitiesOperation.GetActivity().(type) {
-		case *batchpb.BatchOperationResetActivities_Type:
-			if len(a.Type) == 0 {
-				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
-			}
-			resetActivitiesParams.ActivityType = a.Type
-		case *batchpb.BatchOperationResetActivities_MatchAll:
-			if !a.MatchAll {
-				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
-			}
-			resetActivitiesParams.MatchAll = true
-		}
-
-		resetActivitiesParams.ResetAttempts = op.ResetActivitiesOperation.ResetAttempts
-		resetActivitiesParams.ResetHeartbeat = op.ResetActivitiesOperation.ResetHeartbeat
-		resetActivitiesParams.Jitter = op.ResetActivitiesOperation.Jitter.AsDuration()
-		resetActivitiesParams.KeepPaused = op.ResetActivitiesOperation.KeepPaused
-		resetActivitiesParams.RestoreOriginalOptions = op.ResetActivitiesOperation.RestoreOriginalOptions
-		resetActivitiesParams.Identity = op.ResetActivitiesOperation.GetIdentity()
+		input.BatchType = enumspb.BATCH_OPERATION_TYPE_RESET_ACTIVITY
+		identity = op.ResetActivitiesOperation.GetIdentity()
 	case *workflowservice.StartBatchOperationRequest_UpdateActivityOptionsOperation:
-		operationType = batcher.BatchTypeUpdateActivitiesOptions
-		if op.UpdateActivityOptionsOperation == nil {
-			return nil, serviceerror.NewInvalidArgument("update activity options operation is not set")
-		}
-		if op.UpdateActivityOptionsOperation.GetActivityOptions() != nil && op.UpdateActivityOptionsOperation.GetRestoreOriginal() {
-			return nil, serviceerror.NewInvalidArgument("cannot set both activity options and restore original")
-		}
-		if op.UpdateActivityOptionsOperation.GetActivityOptions() == nil && !op.UpdateActivityOptionsOperation.GetRestoreOriginal() {
-			return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or restore original should be set to true")
-		}
-
-		switch a := op.UpdateActivityOptionsOperation.GetActivity().(type) {
-		case *batchpb.BatchOperationUpdateActivityOptions_Type:
-			if len(a.Type) == 0 {
-				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
-			}
-			updateActivitiesOptionsParams.ActivityType = a.Type
-		case *batchpb.BatchOperationUpdateActivityOptions_MatchAll:
-			if !a.MatchAll {
-				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
-			}
-			updateActivitiesOptionsParams.MatchAll = true
-		}
-
-		updateActivitiesOptionsParams.RestoreOriginal = op.UpdateActivityOptionsOperation.GetRestoreOriginal()
-		updateActivitiesOptionsParams.Identity = op.UpdateActivityOptionsOperation.GetIdentity()
-		if updateMask := op.UpdateActivityOptionsOperation.GetUpdateMask(); updateMask != nil {
-			updateActivitiesOptionsParams.UpdateMask = &batcher.FieldMask{Paths: updateMask.Paths}
-		}
-		if ao := op.UpdateActivityOptionsOperation.GetActivityOptions(); ao != nil {
-			updateActivitiesOptionsParams.ActivityOptions = &batcher.ActivityOptions{
-				ScheduleToStartTimeout: ao.ScheduleToStartTimeout.AsDuration(),
-				ScheduleToCloseTime:    ao.ScheduleToCloseTimeout.AsDuration(),
-				StartToCloseTimeout:    ao.StartToCloseTimeout.AsDuration(),
-				HeartbeatTimeout:       ao.HeartbeatTimeout.AsDuration(),
-			}
-			if rp := ao.RetryPolicy; rp != nil {
-				updateActivitiesOptionsParams.ActivityOptions.RetryPolicy = &batcher.RetryPolicy{
-					InitialInterval:        rp.InitialInterval.AsDuration(),
-					MaximumInterval:        rp.MaximumInterval.AsDuration(),
-					BackoffCoefficient:     rp.BackoffCoefficient,
-					NonRetryableErrorTypes: rp.NonRetryableErrorTypes,
-					MaximumAttempts:        rp.MaximumAttempts,
-				}
-			}
-			if tq := ao.TaskQueue; tq != nil {
-				updateActivitiesOptionsParams.ActivityOptions.TaskQueue = &batcher.TaskQueue{
-					Name: tq.Name,
-					Kind: int32(tq.Kind),
-				}
-			}
-		}
+		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)
 	}
 
-	input := &batcher.BatchParams{
-		Namespace:                     request.GetNamespace(),
-		Query:                         visibilityQuery,
-		Executions:                    request.GetExecutions(),
-		Reason:                        request.GetReason(),
-		BatchType:                     operationType,
-		RPS:                           float64(request.GetMaxOperationsPerSecond()),
-		TerminateParams:               batcher.TerminateParams{},
-		CancelParams:                  batcher.CancelParams{},
-		SignalParams:                  signalParams,
-		DeleteParams:                  batcher.DeleteParams{},
-		ResetParams:                   resetParams,
-		UpdateOptionsParams:           updateOptionsParams,
-		UnpauseActivitiesParams:       unpauseActivitiesParams,
-		ResetActivitiesParams:         resetActivitiesParams,
-		UpdateActivitiesOptionsParams: updateActivitiesOptionsParams,
+	if err := batcher.ValidateBatchOperation(request); err != nil {
+		return nil, err
 	}
-	inputPayload, err := sdk.PreferProtoDataConverter.ToPayloads(input)
+
+	inputPayload, err := payloads.Encode(input)
 	if err != nil {
 		return nil, err
 	}
 
 	memo := &commonpb.Memo{
 		Fields: map[string]*commonpb.Payload{
-			batcher.BatchOperationTypeMemo: payload.EncodeString(operationType),
+			batcher.BatchOperationTypeMemo: payload.EncodeString(snakeCaseBatchType(input.BatchType)),
 			batcher.BatchReasonMemo:        payload.EncodeString(request.GetReason()),
 		},
 	}
@@ -4735,7 +4570,7 @@ func (wh *WorkflowHandler) StartBatchOperation(
 	startReq := &workflowservice.StartWorkflowExecutionRequest{
 		Namespace:                request.Namespace,
 		WorkflowId:               request.GetJobId(),
-		WorkflowType:             &commonpb.WorkflowType{Name: batcher.BatchWFTypeName},
+		WorkflowType:             &commonpb.WorkflowType{Name: batcher.BatchWFTypeProtobufName},
 		TaskQueue:                &taskqueuepb.TaskQueue{Name: primitives.PerNSWorkerTaskQueue},
 		Input:                    inputPayload,
 		Identity:                 identity,
@@ -4763,6 +4598,23 @@ func (wh *WorkflowHandler) StartBatchOperation(
 	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,
@@ -4965,7 +4817,7 @@ func (wh *WorkflowHandler) ListBatchOperations(
 		NextPageToken: request.GetNextPageToken(),
 		Query: fmt.Sprintf("%s = '%s' and %s = '%s'",
 			searchattribute.WorkflowType,
-			batcher.BatchWFTypeName,
+			batcher.BatchWFTypeProtobufName,
 			searchattribute.TemporalNamespaceDivision,
 			batcher.NamespaceDivision,
 		),

service/frontend/workflow_handler_test.go

@@ -28,6 +28,7 @@ import (
 	updatepb "go.temporal.io/api/update/v1"
 	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/api/workflowservice/v1"
+	batchspb "go.temporal.io/server/api/batch/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/historyservicemock/v1"
@@ -125,6 +126,7 @@ func (s *WorkflowHandlerSuite) TearDownSuite() {
 
 func (s *WorkflowHandlerSuite) SetupTest() {
 	s.Assertions = require.New(s.T())
+	s.ProtoAssertions = protorequire.New(s.T())
 
 	s.testNamespace = "test-namespace"
 	s.testNamespaceID = "e4f90ec0-1313-45be-9877-8aa41f72a45a"
@@ -2270,14 +2272,24 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_Terminate() {
 	testNamespace := namespace.Name("test-namespace")
 	namespaceID := namespace.ID(uuid.NewString())
 	inputString := "unit test"
+	jobId := uuid.NewString()
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)
 
-	params := &batcher.BatchParams{
-		Namespace: testNamespace.String(),
-		Reason:    inputString,
-		BatchType: batcher.BatchTypeTerminate,
-		Query:     inputString,
+	params := &batchspb.BatchOperationInput{
+		NamespaceId: namespaceID.String(),
+		BatchType:   enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		Request: &workflowservice.StartBatchOperationRequest{
+			Namespace:       testNamespace.String(),
+			VisibilityQuery: inputString,
+			JobId:           jobId,
+			Reason:          inputString,
+			Operation: &workflowservice.StartBatchOperationRequest_TerminationOperation{
+				TerminationOperation: &batchpb.BatchOperationTermination{
+					Identity: inputString,
+				},
+			},
+		},
 	}
 	inputPayload, err := payloads.Encode(params)
 	s.NoError(err)
@@ -2289,21 +2301,21 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_Terminate() {
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			s.Equal(namespaceID.String(), request.NamespaceId)
-			s.Equal(batcher.BatchWFTypeName, request.StartRequest.WorkflowType.Name)
+			s.Equal(batcher.BatchWFTypeProtobufName, request.StartRequest.WorkflowType.Name)
 			s.Equal(primitives.PerNSWorkerTaskQueue, request.StartRequest.TaskQueue.Name)
 			s.Equal(enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, request.StartRequest.WorkflowIdReusePolicy)
 			s.Equal(inputString, request.StartRequest.Identity)
-			s.Equal(payload.EncodeString(batcher.BatchTypeTerminate), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
-			s.Equal(payload.EncodeString(inputString), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
-			s.Equal(payload.EncodeString(inputString), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
-			s.Equal(inputPayload, request.StartRequest.Input)
+			s.ProtoEqual(payload.EncodeString(batcher.BatchTypeTerminate), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
+			s.ProtoEqual(payload.EncodeString(inputString), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
+			s.ProtoEqual(payload.EncodeString(inputString), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
+			s.ProtoEqual(inputPayload, request.StartRequest.Input)
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
 		},
 	)
 	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
 	request := &workflowservice.StartBatchOperationRequest{
 		Namespace: testNamespace.String(),
-		JobId:     uuid.NewString(),
+		JobId:     jobId,
 		Reason:    inputString,
 		Operation: &workflowservice.StartBatchOperationRequest_TerminationOperation{
 			TerminationOperation: &batchpb.BatchOperationTermination{
@@ -2321,14 +2333,24 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_Cancellation() {
 	testNamespace := namespace.Name("test-namespace")
 	namespaceID := namespace.ID(uuid.NewString())
 	inputString := "unit test"
+	jobId := uuid.NewString()
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)
 
-	params := &batcher.BatchParams{
-		Namespace: testNamespace.String(),
-		Reason:    inputString,
-		BatchType: batcher.BatchTypeCancel,
-		Query:     inputString,
+	params := &batchspb.BatchOperationInput{
+		NamespaceId: namespaceID.String(),
+		BatchType:   enumspb.BATCH_OPERATION_TYPE_CANCEL,
+		Request: &workflowservice.StartBatchOperationRequest{
+			Namespace:       testNamespace.String(),
+			VisibilityQuery: inputString,
+			JobId:           jobId,
+			Reason:          inputString,
+			Operation: &workflowservice.StartBatchOperationRequest_CancellationOperation{
+				CancellationOperation: &batchpb.BatchOperationCancellation{
+					Identity: inputString,
+				},
+			},
+		},
 	}
 	inputPayload, err := payloads.Encode(params)
 	s.NoError(err)
@@ -2340,21 +2362,21 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_Cancellation() {
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			s.Equal(namespaceID.String(), request.NamespaceId)
-			s.Equal(batcher.BatchWFTypeName, request.StartRequest.WorkflowType.Name)
+			s.Equal(batcher.BatchWFTypeProtobufName, request.StartRequest.WorkflowType.Name)
 			s.Equal(primitives.PerNSWorkerTaskQueue, request.StartRequest.TaskQueue.Name)
 			s.Equal(enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, request.StartRequest.WorkflowIdReusePolicy)
 			s.Equal(inputString, request.StartRequest.Identity)
-			s.Equal(payload.EncodeString(batcher.BatchTypeCancel), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
-			s.Equal(payload.EncodeString(inputString), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
-			s.Equal(payload.EncodeString(inputString), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
-			s.Equal(inputPayload, request.StartRequest.Input)
+			s.ProtoEqual(payload.EncodeString(batcher.BatchTypeCancel), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
+			s.ProtoEqual(payload.EncodeString(inputString), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
+			s.ProtoEqual(payload.EncodeString(inputString), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
+			s.ProtoEqual(inputPayload, request.StartRequest.Input)
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
 		},
 	)
 	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
 	request := &workflowservice.StartBatchOperationRequest{
 		Namespace: testNamespace.String(),
-		JobId:     uuid.NewString(),
+		JobId:     jobId,
 		Reason:    inputString,
 		Operation: &workflowservice.StartBatchOperationRequest_CancellationOperation{
 			CancellationOperation: &batchpb.BatchOperationCancellation{
@@ -2373,17 +2395,25 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_Signal() {
 	namespaceID := namespace.ID(uuid.NewString())
 	inputString := "unit test"
 	signalName := "signal name"
+	jobId := uuid.NewString()
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)
 	signalPayloads := payloads.EncodeString(signalName)
-	params := &batcher.BatchParams{
-		Namespace: testNamespace.String(),
-		Query:     inputString,
-		Reason:    inputString,
-		BatchType: batcher.BatchTypeSignal,
-		SignalParams: batcher.SignalParams{
-			SignalName: signalName,
-			Input:      signalPayloads,
+	params := &batchspb.BatchOperationInput{
+		NamespaceId: namespaceID.String(),
+		BatchType:   enumspb.BATCH_OPERATION_TYPE_SIGNAL,
+		Request: &workflowservice.StartBatchOperationRequest{
+			Namespace:       testNamespace.String(),
+			VisibilityQuery: inputString,
+			JobId:           jobId,
+			Reason:          inputString,
+			Operation: &workflowservice.StartBatchOperationRequest_SignalOperation{
+				SignalOperation: &batchpb.BatchOperationSignal{
+					Signal:   signalName,
+					Input:    signalPayloads,
+					Identity: inputString,
+				},
+			},
 		},
 	}
 	inputPayload, err := payloads.Encode(params)
@@ -2396,21 +2426,21 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_Signal() {
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			s.Equal(namespaceID.String(), request.NamespaceId)
-			s.Equal(batcher.BatchWFTypeName, request.StartRequest.WorkflowType.Name)
+			s.Equal(batcher.BatchWFTypeProtobufName, request.StartRequest.WorkflowType.Name)
 			s.Equal(primitives.PerNSWorkerTaskQueue, request.StartRequest.TaskQueue.Name)
 			s.Equal(enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, request.StartRequest.WorkflowIdReusePolicy)
 			s.Equal(inputString, request.StartRequest.Identity)
-			s.Equal(payload.EncodeString(batcher.BatchTypeSignal), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
-			s.Equal(payload.EncodeString(inputString), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
-			s.Equal(payload.EncodeString(inputString), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
-			s.Equal(inputPayload, request.StartRequest.Input)
+			s.ProtoEqual(payload.EncodeString(batcher.BatchTypeSignal), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
+			s.ProtoEqual(payload.EncodeString(inputString), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
+			s.ProtoEqual(payload.EncodeString(inputString), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
+			s.ProtoEqual(inputPayload, request.StartRequest.Input)
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
 		},
 	)
 	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
 	request := &workflowservice.StartBatchOperationRequest{
 		Namespace: testNamespace.String(),
-		JobId:     uuid.NewString(),
+		JobId:     jobId,
 		Operation: &workflowservice.StartBatchOperationRequest_SignalOperation{
 			SignalOperation: &batchpb.BatchOperationSignal{
 				Signal:   signalName,
@@ -2438,19 +2468,28 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_WorkflowExecutions_Signal
 	reason := "reason"
 	identity := "identity"
 	signalName := "signal name"
+	jobId := uuid.NewString()
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)
 	signalPayloads := payloads.EncodeString(signalName)
-	params := &batcher.BatchParams{
+	request := &workflowservice.StartBatchOperationRequest{
 		Namespace:  testNamespace.String(),
-		Executions: executions,
+		JobId:      jobId,
 		Reason:     reason,
-		BatchType:  batcher.BatchTypeSignal,
-		SignalParams: batcher.SignalParams{
-			SignalName: signalName,
-			Input:      signalPayloads,
+		Executions: executions,
+		Operation: &workflowservice.StartBatchOperationRequest_SignalOperation{
+			SignalOperation: &batchpb.BatchOperationSignal{
+				Signal:   signalName,
+				Input:    signalPayloads,
+				Identity: identity,
+			},
 		},
 	}
+	params := &batchspb.BatchOperationInput{
+		NamespaceId: namespaceID.String(),
+		BatchType:   enumspb.BATCH_OPERATION_TYPE_SIGNAL,
+		Request:     request,
+	}
 	inputPayload, err := payloads.Encode(params)
 	s.NoError(err)
 	s.mockNamespaceCache.EXPECT().GetNamespaceID(gomock.Any()).Return(namespaceID, nil).AnyTimes()
@@ -2461,31 +2500,18 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_WorkflowExecutions_Signal
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			s.Equal(namespaceID.String(), request.NamespaceId)
-			s.Equal(batcher.BatchWFTypeName, request.StartRequest.WorkflowType.Name)
+			s.Equal(batcher.BatchWFTypeProtobufName, request.StartRequest.WorkflowType.Name)
 			s.Equal(primitives.PerNSWorkerTaskQueue, request.StartRequest.TaskQueue.Name)
 			s.Equal(enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, request.StartRequest.WorkflowIdReusePolicy)
 			s.Equal(identity, request.StartRequest.Identity)
-			s.Equal(payload.EncodeString(batcher.BatchTypeSignal), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
-			s.Equal(payload.EncodeString(reason), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
-			s.Equal(payload.EncodeString(identity), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
-			s.Equal(inputPayload, request.StartRequest.Input)
+			s.ProtoEqual(payload.EncodeString(batcher.BatchTypeSignal), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
+			s.ProtoEqual(payload.EncodeString(reason), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
+			s.ProtoEqual(payload.EncodeString(identity), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
+			s.ProtoEqual(inputPayload, request.StartRequest.Input)
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
 		},
 	)
 	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
-	request := &workflowservice.StartBatchOperationRequest{
-		Namespace: testNamespace.String(),
-		JobId:     uuid.NewString(),
-		Operation: &workflowservice.StartBatchOperationRequest_SignalOperation{
-			SignalOperation: &batchpb.BatchOperationSignal{
-				Signal:   signalName,
-				Input:    signalPayloads,
-				Identity: identity,
-			},
-		},
-		Reason:     reason,
-		Executions: executions,
-	}
 
 	_, err = wh.StartBatchOperation(context.Background(), request)
 	s.NoError(err)
@@ -2502,16 +2528,24 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_WorkflowExecutions_Reset(
 	}
 	reason := "reason"
 	identity := "identity"
+	jobId := uuid.NewString()
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)
-	params := &batcher.BatchParams{
-		Namespace:  testNamespace.String(),
-		Executions: executions,
-		Reason:     reason,
-		BatchType:  batcher.BatchTypeReset,
-		ResetParams: batcher.ResetParams{
-			ResetType:        enumspb.RESET_TYPE_LAST_WORKFLOW_TASK,
-			ResetReapplyType: enumspb.RESET_REAPPLY_TYPE_NONE,
+	params := &batchspb.BatchOperationInput{
+		NamespaceId: namespaceID.String(),
+		BatchType:   enumspb.BATCH_OPERATION_TYPE_RESET,
+		Request: &workflowservice.StartBatchOperationRequest{
+			Namespace:  testNamespace.String(),
+			JobId:      jobId,
+			Reason:     reason,
+			Executions: executions,
+			Operation: &workflowservice.StartBatchOperationRequest_ResetOperation{
+				ResetOperation: &batchpb.BatchOperationReset{
+					Identity:         identity,
+					ResetType:        enumspb.RESET_TYPE_LAST_WORKFLOW_TASK,
+					ResetReapplyType: enumspb.RESET_REAPPLY_TYPE_NONE,
+				},
+			},
 		},
 	}
 	inputPayload, err := payloads.Encode(params)
@@ -2524,21 +2558,21 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_WorkflowExecutions_Reset(
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			s.Equal(namespaceID.String(), request.NamespaceId)
-			s.Equal(batcher.BatchWFTypeName, request.StartRequest.WorkflowType.Name)
+			s.Equal(batcher.BatchWFTypeProtobufName, request.StartRequest.WorkflowType.Name)
 			s.Equal(primitives.PerNSWorkerTaskQueue, request.StartRequest.TaskQueue.Name)
 			s.Equal(enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, request.StartRequest.WorkflowIdReusePolicy)
 			s.Equal(identity, request.StartRequest.Identity)
-			s.Equal(payload.EncodeString(batcher.BatchTypeReset), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
-			s.Equal(payload.EncodeString(reason), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
-			s.Equal(payload.EncodeString(identity), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
-			s.Equal(inputPayload, request.StartRequest.Input)
+			s.ProtoEqual(payload.EncodeString(batcher.BatchTypeReset), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
+			s.ProtoEqual(payload.EncodeString(reason), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
+			s.ProtoEqual(payload.EncodeString(identity), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
+			s.ProtoEqual(inputPayload, request.StartRequest.Input)
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
 		},
 	)
 	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
 	request := &workflowservice.StartBatchOperationRequest{
 		Namespace: testNamespace.String(),
-		JobId:     uuid.NewString(),
+		JobId:     jobId,
 		Operation: &workflowservice.StartBatchOperationRequest_ResetOperation{
 			ResetOperation: &batchpb.BatchOperationReset{
 				ResetType:        enumspb.RESET_TYPE_LAST_WORKFLOW_TASK,
@@ -2588,34 +2622,24 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_WorkflowExecutions_Reset_
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			s.Equal(namespaceID.String(), request.NamespaceId)
-			s.Equal(batcher.BatchWFTypeName, request.StartRequest.WorkflowType.Name)
+			s.Equal(batcher.BatchWFTypeProtobufName, request.StartRequest.WorkflowType.Name)
 			s.Equal(primitives.PerNSWorkerTaskQueue, request.StartRequest.TaskQueue.Name)
 			s.Equal(enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE, request.StartRequest.WorkflowIdReusePolicy)
 			s.Equal(identity, request.StartRequest.Identity)
-			s.Equal(payload.EncodeString(batcher.BatchTypeReset), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
-			s.Equal(payload.EncodeString(reason), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
-			s.Equal(payload.EncodeString(identity), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
+			s.ProtoEqual(payload.EncodeString(batcher.BatchTypeReset), request.StartRequest.Memo.Fields[batcher.BatchOperationTypeMemo])
+			s.ProtoEqual(payload.EncodeString(reason), request.StartRequest.Memo.Fields[batcher.BatchReasonMemo])
+			s.ProtoEqual(payload.EncodeString(identity), request.StartRequest.SearchAttributes.IndexedFields[searchattribute.BatcherUser])
 
 			// Decode the input and verify PostResetOperations are correctly set
-			var batchParams batcher.BatchParams
+			var batchParams batchspb.BatchOperationInput
 			err := payloads.Decode(request.StartRequest.Input, &batchParams)
 			s.NoError(err)
 
 			// Verify that PostResetOperations slice has the correct length and no nil values
-			s.Len(batchParams.ResetParams.PostResetOperations, len(postResetOps))
+			s.Len(batchParams.Request.Operation.(*workflowservice.StartBatchOperationRequest_ResetOperation).ResetOperation.PostResetOperations, len(postResetOps))
 
-			// Ensure no nil values exist in the slice
-			for i, encoded := range batchParams.ResetParams.PostResetOperations {
-				s.NotNil(encoded, "PostResetOperations[%d] should not be nil", i)
-				s.Greater(len(encoded), 0, "PostResetOperations[%d] should not be empty", i)
-
-				// Verify we can unmarshal back to the original operation
-				var decodedOp workflowpb.PostResetOperation
-				err := decodedOp.Unmarshal(encoded)
-				s.NoError(err, "Should be able to unmarshal PostResetOperations[%d]", i)
-
-				// Verify the content matches the original
-				s.ProtoEqual(postResetOps[i], &decodedOp)
+			for i, encoded := range batchParams.Request.Operation.(*workflowservice.StartBatchOperationRequest_ResetOperation).ResetOperation.PostResetOperations {
+				s.ProtoEqual(postResetOps[i], encoded)
 			}
 
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
@@ -2667,13 +2691,10 @@ func (s *WorkflowHandlerSuite) TestStartBatchOperation_WorkflowExecutions_Reset_
 			_ ...grpc.CallOption,
 		) (*historyservice.StartWorkflowExecutionResponse, error) {
 			// Decode the input and verify PostResetOperations slice is properly initialized
-			var batchParams batcher.BatchParams
+			var batchParams batchspb.BatchOperationInput
 			err := payloads.Decode(request.StartRequest.Input, &batchParams)
 			s.NoError(err)
-
-			// When PostResetOperations is empty, the slice should be empty, not nil
-			s.NotNil(batchParams.ResetParams.PostResetOperations)
-			s.Len(batchParams.ResetParams.PostResetOperations, 0)
+			s.Len(batchParams.Request.Operation.(*workflowservice.StartBatchOperationRequest_ResetOperation).ResetOperation.PostResetOperations, 0)
 
 			return &historyservice.StartWorkflowExecutionResponse{}, nil
 		},

service/worker/batcher/activities.go

@@ -5,10 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"slices"
 	"time"
 
 	"github.com/pborman/uuid"
 	activitypb "go.temporal.io/api/activity/v1"
+	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
@@ -16,6 +18,7 @@ import (
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/activity"
 	sdkclient "go.temporal.io/sdk/client"
+	batchspb "go.temporal.io/server/api/batch/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
@@ -55,6 +58,13 @@ func (a *activities) checkNamespace(namespace string) error {
 	return nil
 }
 
+func (a *activities) checkNamespaceID(namespaceID string) error {
+	if namespaceID != a.namespaceID.String() {
+		return errNamespaceMismatch
+	}
+	return nil
+}
+
 // BatchActivity is an activity for processing batch operation.
 func (a *activities) BatchActivity(ctx context.Context, batchParams BatchParams) (HeartBeatDetails, error) {
 	logger := a.getActivityLogger(ctx)
@@ -102,7 +112,7 @@ func (a *activities) BatchActivity(ctx context.Context, batchParams BatchParams)
 		}
 	}
 
-	adjustedQuery := a.adjustQuery(batchParams)
+	adjustedQuery := a.adjustQuery(batchParams.Query, batchParams.BatchType)
 
 	if startOver {
 		estimateCount := int64(len(batchParams.Executions))
@@ -198,6 +208,128 @@ func (a *activities) BatchActivity(ctx context.Context, batchParams BatchParams)
 	return hbd, nil
 }
 
+// BatchActivityWithProtobuf is an activity for processing batch operations using protobuf as the input type.
+// nolint:revive,cognitive-complexity
+func (a *activities) BatchActivityWithProtobuf(ctx context.Context, batchParams *batchspb.BatchOperationInput) (HeartBeatDetails, error) {
+	logger := a.getActivityLogger(ctx)
+	hbd := HeartBeatDetails{}
+	metricsHandler := a.MetricsHandler.WithTags(metrics.OperationTag(metrics.BatcherScope), metrics.NamespaceIDTag(batchParams.NamespaceId))
+
+	if err := a.checkNamespaceID(batchParams.NamespaceId); err != nil {
+		metrics.BatcherOperationFailures.With(metricsHandler).Record(1)
+		logger.Error("Failed to run batch operation due to namespace mismatch", tag.Error(err))
+		return hbd, err
+	}
+
+	sdkClient := a.ClientFactory.NewClient(sdkclient.Options{
+		Namespace:     a.namespace.String(),
+		DataConverter: sdk.PreferProtoDataConverter,
+	})
+	startOver := true
+	if activity.HasHeartbeatDetails(ctx) {
+		if err := activity.GetHeartbeatDetails(ctx, &hbd); err == nil {
+			startOver = false
+		} else {
+			logger.Error("Failed to recover from last heartbeat, start over from beginning", tag.Error(err))
+		}
+	}
+
+	adjustedQuery := a.adjustQueryBatchTypeEnum(batchParams.Request.VisibilityQuery, batchParams.BatchType)
+
+	if startOver {
+		estimateCount := int64(len(batchParams.Request.Executions))
+		if len(adjustedQuery) > 0 {
+			resp, err := sdkClient.CountWorkflow(ctx, &workflowservice.CountWorkflowExecutionsRequest{
+				Query: adjustedQuery,
+			})
+			if err != nil {
+				metrics.BatcherOperationFailures.With(metricsHandler).Record(1)
+				logger.Error("Failed to get estimate workflow count", tag.Error(err))
+				return HeartBeatDetails{}, err
+			}
+			estimateCount = resp.GetCount()
+		}
+		hbd.TotalEstimate = estimateCount
+	}
+	rps := a.rps(a.namespace.String())
+	rateLimit := rate.Limit(rps)
+	burstLimit := rps // should never be zero because everything would be rejected
+	rateLimiter := rate.NewLimiter(rateLimit, burstLimit)
+	taskCh := make(chan taskDetail, pageSize)
+	respCh := make(chan error, pageSize)
+	for i := 0; i < a.getOperationConcurrency(int(batchParams.Concurrency)); i++ {
+		go startTaskProcessorProtobuf(ctx, batchParams, a.namespace.String(), taskCh, respCh, rateLimiter, sdkClient, a.FrontendClient, metricsHandler, logger)
+	}
+
+	for {
+		executions := batchParams.Request.Executions
+		pageToken := hbd.PageToken
+		if len(adjustedQuery) > 0 {
+			resp, err := sdkClient.ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
+				PageSize:      int32(pageSize),
+				NextPageToken: pageToken,
+				Query:         adjustedQuery,
+			})
+			if err != nil {
+				metrics.BatcherOperationFailures.With(metricsHandler).Record(1)
+				logger.Error("Failed to list workflow executions", tag.Error(err))
+				return HeartBeatDetails{}, err
+			}
+			pageToken = resp.NextPageToken
+			for _, wf := range resp.Executions {
+				executions = append(executions, wf.Execution)
+			}
+		}
+
+		batchCount := len(executions)
+		if batchCount <= 0 {
+			break
+		}
+		// send all tasks
+		for _, wf := range executions {
+			taskCh <- taskDetail{
+				execution: wf,
+				attempts:  1,
+				hbd:       hbd,
+			}
+		}
+
+		succCount := 0
+		errCount := 0
+		// wait for counters indicate this batch is done
+	Loop:
+		for {
+			select {
+			case err := <-respCh:
+				if err == nil {
+					succCount++
+				} else {
+					errCount++
+				}
+				if succCount+errCount == batchCount {
+					break Loop
+				}
+			case <-ctx.Done():
+				metrics.BatcherOperationFailures.With(metricsHandler).Record(1)
+				logger.Error("Failed to complete batch operation", tag.Error(ctx.Err()))
+				return HeartBeatDetails{}, ctx.Err()
+			}
+		}
+
+		hbd.CurrentPage++
+		hbd.PageToken = pageToken
+		hbd.SuccessCount += succCount
+		hbd.ErrorCount += errCount
+		activity.RecordHeartbeat(ctx, hbd)
+
+		if len(hbd.PageToken) == 0 {
+			break
+		}
+	}
+
+	return hbd, nil
+}
+
 func (a *activities) getActivityLogger(ctx context.Context) log.Logger {
 	wfInfo := activity.GetInfo(ctx)
 	return log.With(
@@ -208,17 +340,31 @@ func (a *activities) getActivityLogger(ctx context.Context) log.Logger {
 	)
 }
 
-func (a *activities) adjustQuery(batchParams BatchParams) string {
-	if len(batchParams.Query) == 0 {
+func (a *activities) adjustQuery(query, batchType string) string {
+	if len(query) == 0 {
 		// don't add anything if query is empty
-		return batchParams.Query
+		return query
 	}
 
-	switch batchParams.BatchType {
+	switch batchType {
 	case BatchTypeTerminate, BatchTypeSignal, BatchTypeCancel, BatchTypeUpdateOptions, BatchTypeUnpauseActivities:
-		return fmt.Sprintf("(%s) AND (%s)", batchParams.Query, statusRunningQueryFilter)
+		return fmt.Sprintf("(%s) AND (%s)", query, statusRunningQueryFilter)
 	default:
-		return batchParams.Query
+		return query
+	}
+}
+
+func (a *activities) adjustQueryBatchTypeEnum(query string, batchType enumspb.BatchOperationType) string {
+	if len(query) == 0 {
+		// don't add anything if query is empty
+		return query
+	}
+
+	switch batchType {
+	case enumspb.BATCH_OPERATION_TYPE_TERMINATE, enumspb.BATCH_OPERATION_TYPE_SIGNAL, enumspb.BATCH_OPERATION_TYPE_CANCEL, enumspb.BATCH_OPERATION_TYPE_UPDATE_EXECUTION_OPTIONS, enumspb.BATCH_OPERATION_TYPE_UNPAUSE_ACTIVITY:
+		return fmt.Sprintf("(%s) AND (%s)", query, statusRunningQueryFilter)
+	default:
+		return query
 	}
 }
 
@@ -261,22 +407,22 @@ func startTaskProcessor(
 			switch batchParams.BatchType {
 			case BatchTypeTerminate:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
-						return sdkClient.TerminateWorkflow(ctx, workflowID, runID, batchParams.Reason)
+					func(execution *commonpb.WorkflowExecution) error {
+						return sdkClient.TerminateWorkflow(ctx, execution.WorkflowId, execution.RunId, batchParams.Reason)
 					})
 			case BatchTypeCancel:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
-						return sdkClient.CancelWorkflow(ctx, workflowID, runID)
+					func(execution *commonpb.WorkflowExecution) error {
+						return sdkClient.CancelWorkflow(ctx, execution.WorkflowId, execution.RunId)
 					})
 			case BatchTypeSignal:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						_, err := frontendClient.SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
 							Namespace: batchParams.Namespace,
 							WorkflowExecution: &commonpb.WorkflowExecution{
-								WorkflowId: workflowID,
-								RunId:      runID,
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
 							},
 							SignalName: batchParams.SignalParams.SignalName,
 							Input:      batchParams.SignalParams.Input,
@@ -285,22 +431,22 @@ func startTaskProcessor(
 					})
 			case BatchTypeDelete:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						_, err := frontendClient.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
 							Namespace: batchParams.Namespace,
 							WorkflowExecution: &commonpb.WorkflowExecution{
-								WorkflowId: workflowID,
-								RunId:      runID,
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
 							},
 						})
 						return err
 					})
 			case BatchTypeReset:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						workflowExecution := &commonpb.WorkflowExecution{
-							WorkflowId: workflowID,
-							RunId:      runID,
+							WorkflowId: execution.WorkflowId,
+							RunId:      execution.RunId,
 						}
 						var eventId int64
 						var err error
@@ -334,12 +480,12 @@ func startTaskProcessor(
 					})
 			case BatchTypeUnpauseActivities:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						unpauseRequest := &workflowservice.UnpauseActivityRequest{
 							Namespace: batchParams.Namespace,
 							Execution: &commonpb.WorkflowExecution{
-								WorkflowId: workflowID,
-								RunId:      runID,
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
 							},
 							Identity:       batchParams.UnpauseActivitiesParams.Identity,
 							Activity:       &workflowservice.UnpauseActivityRequest_Type{Type: batchParams.UnpauseActivitiesParams.ActivityType},
@@ -359,13 +505,13 @@ func startTaskProcessor(
 
 			case BatchTypeUpdateOptions:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						var err error
 						_, err = frontendClient.UpdateWorkflowExecutionOptions(ctx, &workflowservice.UpdateWorkflowExecutionOptionsRequest{
 							Namespace: batchParams.Namespace,
 							WorkflowExecution: &commonpb.WorkflowExecution{
-								WorkflowId: workflowID,
-								RunId:      runID,
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
 							},
 							WorkflowExecutionOptions: batchParams.UpdateOptionsParams.WorkflowExecutionOptions,
 							UpdateMask:               &fieldmaskpb.FieldMask{Paths: batchParams.UpdateOptionsParams.UpdateMask.Paths},
@@ -375,12 +521,12 @@ func startTaskProcessor(
 
 			case BatchTypeResetActivities:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						resetRequest := &workflowservice.ResetActivityRequest{
 							Namespace: batchParams.Namespace,
 							Execution: &commonpb.WorkflowExecution{
-								WorkflowId: workflowID,
-								RunId:      runID,
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
 							},
 							Identity:               batchParams.ResetActivitiesParams.Identity,
 							Activity:               &workflowservice.ResetActivityRequest_Type{Type: batchParams.ResetActivitiesParams.ActivityType},
@@ -401,12 +547,12 @@ func startTaskProcessor(
 					})
 			case BatchTypeUpdateActivitiesOptions:
 				err = processTask(ctx, limiter, task,
-					func(workflowID, runID string) error {
+					func(execution *commonpb.WorkflowExecution) error {
 						updateRequest := &workflowservice.UpdateActivityOptionsRequest{
 							Namespace: batchParams.Namespace,
 							Execution: &commonpb.WorkflowExecution{
-								WorkflowId: workflowID,
-								RunId:      runID,
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
 							},
 							Activity:        &workflowservice.UpdateActivityOptionsRequest_Type{Type: batchParams.UpdateActivitiesOptionsParams.ActivityType},
 							UpdateMask:      &fieldmaskpb.FieldMask{Paths: batchParams.UpdateActivitiesOptionsParams.UpdateMask.Paths},
@@ -442,7 +588,8 @@ func startTaskProcessor(
 						_, err = frontendClient.UpdateActivityOptions(ctx, updateRequest)
 						return err
 					})
-				// QUESTION seankane (2025-07-18): why do we not have a default case and return an error? @yuri/@chetan
+			default:
+				err = errors.New("unknown batch type: " + batchParams.BatchType)
 			}
 			if err != nil {
 				metrics.BatcherProcessorFailures.With(metricsHandler).Record(1)
@@ -464,11 +611,238 @@ func startTaskProcessor(
 	}
 }
 
+// nolint:revive,cognitive-complexity
+func startTaskProcessorProtobuf(
+	ctx context.Context,
+	batchOperation *batchspb.BatchOperationInput,
+	namespace string,
+	taskCh chan taskDetail,
+	respCh chan error,
+	limiter *rate.Limiter,
+	sdkClient sdkclient.Client,
+	frontendClient workflowservice.WorkflowServiceClient,
+	metricsHandler metrics.Handler,
+	logger log.Logger,
+) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case task := <-taskCh:
+			if isDone(ctx) {
+				return
+			}
+			var err error
+
+			switch operation := batchOperation.Request.Operation.(type) {
+			case *workflowservice.StartBatchOperationRequest_TerminationOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						return sdkClient.TerminateWorkflow(ctx, execution.WorkflowId, execution.RunId, batchOperation.Request.Reason)
+					})
+			case *workflowservice.StartBatchOperationRequest_CancellationOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						return sdkClient.CancelWorkflow(ctx, execution.WorkflowId, execution.RunId)
+					})
+			case *workflowservice.StartBatchOperationRequest_SignalOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						_, err := frontendClient.SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
+							Namespace: namespace,
+							WorkflowExecution: &commonpb.WorkflowExecution{
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
+							},
+							SignalName: operation.SignalOperation.GetSignal(),
+							Input:      operation.SignalOperation.GetInput(),
+							Identity:   operation.SignalOperation.GetIdentity(),
+						})
+						return err
+					})
+			case *workflowservice.StartBatchOperationRequest_DeletionOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						_, err := frontendClient.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
+							Namespace: namespace,
+							WorkflowExecution: &commonpb.WorkflowExecution{
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
+							},
+						})
+						return err
+					})
+			case *workflowservice.StartBatchOperationRequest_ResetOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						workflowExecution := &commonpb.WorkflowExecution{
+							WorkflowId: execution.WorkflowId,
+							RunId:      execution.RunId,
+						}
+						var eventId int64
+						var err error
+						//nolint:staticcheck // SA1019: worker versioning v0.31
+						var resetReapplyType enumspb.ResetReapplyType
+						var resetReapplyExcludeTypes []enumspb.ResetReapplyExcludeType
+						if operation.ResetOperation.Options != nil {
+							// Using ResetOptions
+							// Note: getResetEventIDByOptions may modify workflowExecution.RunId, if reset should be to a prior run
+							//nolint:staticcheck // SA1019: worker versioning v0.31
+							eventId, err = getResetEventIDByOptions(ctx, operation.ResetOperation.Options, namespace, workflowExecution, frontendClient, logger)
+							//nolint:staticcheck // SA1019: worker versioning v0.31
+							resetReapplyType = operation.ResetOperation.Options.ResetReapplyType
+							//nolint:staticcheck // SA1019: worker versioning v0.31
+							resetReapplyExcludeTypes = operation.ResetOperation.Options.ResetReapplyExcludeTypes
+						} else {
+							// Old fields
+							//nolint:staticcheck // SA1019: worker versioning v0.31
+							eventId, err = getResetEventIDByType(ctx, operation.ResetOperation.ResetType, batchOperation.Request.Namespace, workflowExecution, frontendClient, logger)
+							//nolint:staticcheck // SA1019: worker versioning v0.31
+							resetReapplyType = operation.ResetOperation.ResetReapplyType
+						}
+						if err != nil {
+							return err
+						}
+						_, err = frontendClient.ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
+							Namespace:                 namespace,
+							WorkflowExecution:         workflowExecution,
+							Reason:                    batchOperation.Request.Reason,
+							RequestId:                 uuid.New(),
+							WorkflowTaskFinishEventId: eventId,
+							ResetReapplyType:          resetReapplyType,
+							ResetReapplyExcludeTypes:  resetReapplyExcludeTypes,
+							PostResetOperations:       operation.ResetOperation.PostResetOperations,
+							Identity:                  operation.ResetOperation.Identity,
+						})
+						return err
+					})
+			case *workflowservice.StartBatchOperationRequest_UnpauseActivitiesOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						unpauseRequest := &workflowservice.UnpauseActivityRequest{
+							Namespace: namespace,
+							Execution: &commonpb.WorkflowExecution{
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
+							},
+							Identity:       operation.UnpauseActivitiesOperation.Identity,
+							ResetAttempts:  operation.UnpauseActivitiesOperation.ResetAttempts,
+							ResetHeartbeat: operation.UnpauseActivitiesOperation.ResetHeartbeat,
+							Jitter:         operation.UnpauseActivitiesOperation.Jitter,
+						}
+
+						switch ao := operation.UnpauseActivitiesOperation.GetActivity().(type) {
+						case *batchpb.BatchOperationUnpauseActivities_Type:
+							unpauseRequest.Activity = &workflowservice.UnpauseActivityRequest_Type{
+								Type: ao.Type,
+							}
+						case *batchpb.BatchOperationUnpauseActivities_MatchAll:
+							unpauseRequest.Activity = &workflowservice.UnpauseActivityRequest_UnpauseAll{UnpauseAll: true}
+						default:
+							return errors.New(fmt.Sprintf("unknown activity type: %v", operation.UnpauseActivitiesOperation.GetActivity()))
+						}
+
+						_, err = frontendClient.UnpauseActivity(ctx, unpauseRequest)
+						return err
+					})
+
+			case *workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						var err error
+						_, err = frontendClient.UpdateWorkflowExecutionOptions(ctx, &workflowservice.UpdateWorkflowExecutionOptionsRequest{
+							Namespace: namespace,
+							WorkflowExecution: &commonpb.WorkflowExecution{
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
+							},
+							WorkflowExecutionOptions: operation.UpdateWorkflowOptionsOperation.WorkflowExecutionOptions,
+							UpdateMask:               &fieldmaskpb.FieldMask{Paths: operation.UpdateWorkflowOptionsOperation.UpdateMask.Paths},
+						})
+						return err
+					})
+			case *workflowservice.StartBatchOperationRequest_ResetActivitiesOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						resetRequest := &workflowservice.ResetActivityRequest{
+							Namespace: namespace,
+							Execution: &commonpb.WorkflowExecution{
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
+							},
+							Identity:               operation.ResetActivitiesOperation.Identity,
+							ResetHeartbeat:         operation.ResetActivitiesOperation.ResetHeartbeat,
+							Jitter:                 operation.ResetActivitiesOperation.Jitter,
+							KeepPaused:             operation.ResetActivitiesOperation.KeepPaused,
+							RestoreOriginalOptions: operation.ResetActivitiesOperation.RestoreOriginalOptions,
+						}
+
+						switch ao := operation.ResetActivitiesOperation.GetActivity().(type) {
+						case *batchpb.BatchOperationResetActivities_Type:
+							resetRequest.Activity = &workflowservice.ResetActivityRequest_Type{Type: ao.Type}
+						case *batchpb.BatchOperationResetActivities_MatchAll:
+							resetRequest.Activity = &workflowservice.ResetActivityRequest_MatchAll{MatchAll: true}
+						default:
+							return errors.New(fmt.Sprintf("unknown activity type: %v", operation.ResetActivitiesOperation.GetActivity()))
+						}
+
+						_, err = frontendClient.ResetActivity(ctx, resetRequest)
+						return err
+					})
+			case *workflowservice.StartBatchOperationRequest_UpdateActivityOptionsOperation:
+				err = processTask(ctx, limiter, task,
+					func(execution *commonpb.WorkflowExecution) error {
+						updateRequest := &workflowservice.UpdateActivityOptionsRequest{
+							Namespace: namespace,
+							Execution: &commonpb.WorkflowExecution{
+								WorkflowId: execution.WorkflowId,
+								RunId:      execution.RunId,
+							},
+							UpdateMask:      &fieldmaskpb.FieldMask{Paths: operation.UpdateActivityOptionsOperation.UpdateMask.Paths},
+							RestoreOriginal: operation.UpdateActivityOptionsOperation.RestoreOriginal,
+							Identity:        operation.UpdateActivityOptionsOperation.Identity,
+						}
+
+						switch ao := operation.UpdateActivityOptionsOperation.GetActivity().(type) {
+						case *batchpb.BatchOperationUpdateActivityOptions_Type:
+							updateRequest.Activity = &workflowservice.UpdateActivityOptionsRequest_Type{Type: ao.Type}
+						case *batchpb.BatchOperationUpdateActivityOptions_MatchAll:
+							updateRequest.Activity = &workflowservice.UpdateActivityOptionsRequest_MatchAll{MatchAll: true}
+						default:
+							return errors.New(fmt.Sprintf("unknown activity type: %v", operation.UpdateActivityOptionsOperation.GetActivity()))
+						}
+
+						updateRequest.ActivityOptions = operation.UpdateActivityOptionsOperation.GetActivityOptions()
+						_, err = frontendClient.UpdateActivityOptions(ctx, updateRequest)
+						return err
+					})
+			default:
+				err = errors.New(fmt.Sprintf("unknown batch type: %v", batchOperation.BatchType))
+			}
+			if err != nil {
+				metrics.BatcherProcessorFailures.With(metricsHandler).Record(1)
+				logger.Error("Failed to process batch operation task", tag.Error(err))
+				nonRetryable := slices.Contains(batchOperation.NonRetryableErrors, err.Error())
+				if nonRetryable || task.attempts > int(batchOperation.AttemptsOnRetryableError) {
+					respCh <- err
+				} else {
+					// put back to the channel if less than attemptsOnError
+					task.attempts++
+					taskCh <- task
+				}
+			} else {
+				metrics.BatcherProcessorSuccess.With(metricsHandler).Record(1)
+				respCh <- nil
+			}
+		}
+	}
+}
+
 func processTask(
 	ctx context.Context,
 	limiter *rate.Limiter,
 	task taskDetail,
-	procFn func(string, string) error,
+	procFn func(*commonpb.WorkflowExecution) error,
 ) error {
 
 	err := limiter.Wait(ctx)
@@ -477,7 +851,7 @@ func processTask(
 	}
 	activity.RecordHeartbeat(ctx, task.hbd)
 
-	err = procFn(task.execution.GetWorkflowId(), task.execution.GetRunId())
+	err = procFn(task.execution)
 	if err != nil {
 		// NotFound means wf is not running or deleted
 		if !common.IsNotFoundError(err) {

service/worker/batcher/activities_test.go

@@ -343,7 +343,66 @@ func (s *activitiesSuite) TestAdjustQuery() {
 		s.Run(testRun.name, func() {
 			a := activities{}
 			batchParams := BatchParams{Query: testRun.query, BatchType: testRun.batchType}
-			adjustedQuery := a.adjustQuery(batchParams)
+			adjustedQuery := a.adjustQuery(batchParams.Query, batchParams.BatchType)
+			s.Equal(testRun.expectedResult, adjustedQuery)
+		})
+	}
+}
+
+func (s *activitiesSuite) TestAdjustQueryBatchTypeEnum() {
+	tests := []struct {
+		name           string
+		query          string
+		expectedResult string
+		batchType      enumspb.BatchOperationType
+	}{
+		{
+			name:           "Empty query",
+			query:          "",
+			expectedResult: "",
+			batchType:      enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		},
+		{
+			name:           "Acceptance",
+			query:          "A=B",
+			expectedResult: fmt.Sprintf("(A=B) AND (%s)", statusRunningQueryFilter),
+			batchType:      enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		},
+		{
+			name:           "Acceptance with parenthesis",
+			query:          "(A=B)",
+			expectedResult: fmt.Sprintf("((A=B)) AND (%s)", statusRunningQueryFilter),
+			batchType:      enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		},
+		{
+			name:           "Acceptance with multiple conditions",
+			query:          "(A=B) OR C=D",
+			expectedResult: fmt.Sprintf("((A=B) OR C=D) AND (%s)", statusRunningQueryFilter),
+			batchType:      enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		},
+		{
+			name:           "Contains status - 1",
+			query:          "ExecutionStatus=Completed",
+			expectedResult: fmt.Sprintf("(ExecutionStatus=Completed) AND (%s)", statusRunningQueryFilter),
+			batchType:      enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		},
+		{
+			name:           "Contains status - 2",
+			query:          "A=B OR ExecutionStatus='Completed'",
+			expectedResult: fmt.Sprintf("(A=B OR ExecutionStatus='Completed') AND (%s)", statusRunningQueryFilter),
+			batchType:      enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+		},
+		{
+			name:           "Not supported batch type",
+			query:          "A=B",
+			expectedResult: "A=B",
+			batchType:      enumspb.BATCH_OPERATION_TYPE_UNSPECIFIED,
+		},
+	}
+	for _, testRun := range tests {
+		s.Run(testRun.name, func() {
+			a := activities{}
+			adjustedQuery := a.adjustQueryBatchTypeEnum(testRun.query, testRun.batchType)
 			s.Equal(testRun.expectedResult, adjustedQuery)
 		})
 	}

service/worker/batcher/fx.go

@@ -15,8 +15,9 @@ import (
 
 const (
 	// BatchWFTypeName is the workflow type
-	BatchWFTypeName   = "temporal-sys-batch-workflow"
-	NamespaceDivision = "TemporalBatcher"
+	BatchWFTypeName         = "temporal-sys-batch-workflow"
+	BatchWFTypeProtobufName = "temporal-sys-batch-workflow-protobuf"
+	NamespaceDivision       = "TemporalBatcher"
 )
 
 type (
@@ -67,6 +68,8 @@ func (s *workerComponent) DedicatedWorkerOptions(ns *namespace.Namespace) *worke
 
 func (s *workerComponent) Register(registry sdkworker.Registry, ns *namespace.Namespace, _ workercommon.RegistrationDetails) func() {
 	registry.RegisterWorkflowWithOptions(BatchWorkflow, workflow.RegisterOptions{Name: BatchWFTypeName})
+	// Newer version of the batch workflow which was rewritten to accept a proto struct as input.
+	registry.RegisterWorkflowWithOptions(BatchWorkflowProtobuf, workflow.RegisterOptions{Name: BatchWFTypeProtobufName})
 	registry.RegisterActivity(s.activities(ns.Name(), ns.ID()))
 	return nil
 }

service/worker/batcher/workflow.go

@@ -5,13 +5,18 @@ import (
 	"fmt"
 	"time"
 
+	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
 	workflowpb "go.temporal.io/api/workflow/v1"
+	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
+	batchspb "go.temporal.io/server/api/batch/v1"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/worker_versioning"
+	"google.golang.org/protobuf/types/known/durationpb"
 )
 
 const (
@@ -40,7 +45,7 @@ const (
 	BatchTypeReset = "reset"
 	// BatchTypeUpdateOptions is batch type for updating the options of workflow executions
 	BatchTypeUpdateOptions = "update_options"
-	// BatchTypePauseActivities is batch type for unpausing activities
+	// BatchTypeUnpauseActivities is batch type for unpausing activities
 	BatchTypeUnpauseActivities = "unpause_activities"
 	// BatchTypeUpdateActivitiesOptions is batch type for updating the options of activities
 	BatchTypeUpdateActivitiesOptions = "update_activity_options"
@@ -262,6 +267,34 @@ func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDeta
 	return result, err
 }
 
+// BatchWorkflowProtobuf is the workflow that runs a batch job of resetting workflows.
+func BatchWorkflowProtobuf(ctx workflow.Context, batchParams *batchspb.BatchOperationInput) (HeartBeatDetails, error) {
+	if batchParams == nil {
+		return HeartBeatDetails{}, errors.New("batchParams is nil")
+	}
+
+	batchParams = setDefaultParamsProtobuf(batchParams)
+	err := ValidateBatchOperation(batchParams.Request)
+	if err != nil {
+		return HeartBeatDetails{}, err
+	}
+
+	batchActivityOptions.HeartbeatTimeout = batchParams.ActivityHeartbeatTimeout.AsDuration()
+	opt := workflow.WithActivityOptions(ctx, batchActivityOptions)
+	var result HeartBeatDetails
+	var ac *activities
+	err = workflow.ExecuteActivity(opt, ac.BatchActivityWithProtobuf, batchParams).Get(ctx, &result)
+	if err != nil {
+		return HeartBeatDetails{}, err
+	}
+
+	err = attachBatchOperationStats(ctx, result)
+	if err != nil {
+		return HeartBeatDetails{}, err
+	}
+	return result, err
+}
+
 type BatchOperationStats struct {
 	NumSuccess int
 	NumFailure int
@@ -284,25 +317,25 @@ func validateParams(params BatchParams) error {
 		params.Reason == "" ||
 		params.Namespace == "" ||
 		(params.Query == "" && len(params.Executions) == 0) {
-		return fmt.Errorf("must provide required parameters: BatchType/Reason/Namespace/Query/Executions")
+		return errors.New("must provide required parameters: BatchType/Reason/Namespace/Query/Executions")
 	}
 
 	if len(params.Query) > 0 && len(params.Executions) > 0 {
-		return fmt.Errorf("batch query and executions are mutually exclusive")
+		return errors.New("batch query and executions are mutually exclusive")
 	}
 
 	switch params.BatchType {
 	case BatchTypeSignal:
 		if params.SignalParams.SignalName == "" {
-			return fmt.Errorf("must provide signal name")
+			return errors.New("must provide signal name")
 		}
 		return nil
 	case BatchTypeUpdateOptions:
 		if params.UpdateOptionsParams.WorkflowExecutionOptions == nil {
-			return fmt.Errorf("must provide UpdateOptions")
+			return errors.New("must provide UpdateOptions")
 		}
 		if params.UpdateOptionsParams.UpdateMask == nil {
-			return fmt.Errorf("must provide UpdateMask")
+			return errors.New("must provide UpdateMask")
 		}
 		return worker_versioning.ValidateVersioningOverride(params.UpdateOptionsParams.WorkflowExecutionOptions.VersioningOverride)
 	case BatchTypeCancel, BatchTypeTerminate, BatchTypeDelete, BatchTypeReset:
@@ -327,6 +360,132 @@ func validateParams(params BatchParams) error {
 	}
 }
 
+// nolint:revive,cognitive-complexity
+func ValidateBatchOperation(params *workflowservice.StartBatchOperationRequest) error {
+	if params.GetOperation() == nil ||
+		params.GetReason() == "" ||
+		params.GetNamespace() == "" ||
+		(params.GetVisibilityQuery() == "" && len(params.GetExecutions()) == 0) {
+		return serviceerror.NewInvalidArgument("must provide required parameters: BatchType/Reason/Namespace/Query/Executions")
+	}
+
+	if len(params.GetJobId()) == 0 {
+		return serviceerror.NewInvalidArgument("JobId is not set on request.")
+	}
+	if len(params.GetNamespace()) == 0 {
+		return serviceerror.NewInvalidArgument("Namespace is not set on request.")
+	}
+	if len(params.GetVisibilityQuery()) == 0 && len(params.GetExecutions()) == 0 {
+		return serviceerror.NewInvalidArgument("VisibilityQuery or Executions must be set on request.")
+	}
+	if len(params.GetVisibilityQuery()) != 0 && len(params.GetExecutions()) != 0 {
+		return errors.New("batch query and executions are mutually exclusive")
+	}
+	if len(params.GetReason()) == 0 {
+		return serviceerror.NewInvalidArgument("Reason is not set on request.")
+	}
+	if params.GetOperation() == nil {
+		return serviceerror.NewInvalidArgument("Batch operation is not set on request.")
+	}
+
+	switch op := params.GetOperation().(type) {
+	case *workflowservice.StartBatchOperationRequest_SignalOperation:
+		if op.SignalOperation.GetSignal() == "" {
+			return errors.New("must provide signal name")
+		}
+		return nil
+	case *workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation:
+		if op.UpdateWorkflowOptionsOperation.GetWorkflowExecutionOptions() == nil {
+			return errors.New("must provide UpdateOptions")
+		}
+		if op.UpdateWorkflowOptionsOperation.GetUpdateMask() == nil {
+			return errors.New("must provide UpdateMask")
+		}
+		return worker_versioning.ValidateVersioningOverride(op.UpdateWorkflowOptionsOperation.GetWorkflowExecutionOptions().GetVersioningOverride())
+	case *workflowservice.StartBatchOperationRequest_CancellationOperation,
+		*workflowservice.StartBatchOperationRequest_TerminationOperation,
+		*workflowservice.StartBatchOperationRequest_DeletionOperation:
+		return nil
+	case *workflowservice.StartBatchOperationRequest_ResetOperation:
+		if op.ResetOperation == nil {
+			return serviceerror.NewInvalidArgument("reset operation is not set")
+		}
+		if op.ResetOperation.Options != nil {
+			if op.ResetOperation.Options.Target == nil {
+				return serviceerror.NewInvalidArgument("batch reset missing target")
+			}
+		} else {
+			//nolint:staticcheck // SA1019: GetResetType is deprecated but still needed for backward compatibility
+			resetType := op.ResetOperation.GetResetType()
+			if _, ok := enumspb.ResetType_name[int32(resetType)]; !ok || resetType == enumspb.RESET_TYPE_UNSPECIFIED {
+				return serviceerror.NewInvalidArgumentf("unknown batch reset type %v", resetType)
+			}
+		}
+	case *workflowservice.StartBatchOperationRequest_UnpauseActivitiesOperation:
+		if op.UnpauseActivitiesOperation == nil {
+			return serviceerror.NewInvalidArgument("unpause activities operation is not set")
+		}
+		if op.UnpauseActivitiesOperation.GetActivity() == nil {
+			return serviceerror.NewInvalidArgument("activity filter must be set")
+		}
+		switch a := op.UnpauseActivitiesOperation.GetActivity().(type) {
+		case *batchpb.BatchOperationUnpauseActivities_Type:
+			if len(a.Type) == 0 {
+				return serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
+			}
+		case *batchpb.BatchOperationUnpauseActivities_MatchAll:
+			if !a.MatchAll {
+				return serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
+			}
+		}
+		return nil
+	case *workflowservice.StartBatchOperationRequest_ResetActivitiesOperation:
+		if op.ResetActivitiesOperation == nil {
+			return serviceerror.NewInvalidArgument("reset activities operation is not set")
+		}
+		if op.ResetActivitiesOperation.GetActivity() == nil && !op.ResetActivitiesOperation.GetMatchAll() {
+			return serviceerror.NewInvalidArgument("must provide ActivityType or MatchAll")
+		}
+
+		switch a := op.ResetActivitiesOperation.GetActivity().(type) {
+		case *batchpb.BatchOperationResetActivities_Type:
+			if len(a.Type) == 0 {
+				return serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
+			}
+		case *batchpb.BatchOperationResetActivities_MatchAll:
+			if !a.MatchAll {
+				return serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
+			}
+		}
+		return nil
+	case *workflowservice.StartBatchOperationRequest_UpdateActivityOptionsOperation:
+		if op.UpdateActivityOptionsOperation == nil {
+			return serviceerror.NewInvalidArgument("update activity options operation is not set")
+		}
+		if op.UpdateActivityOptionsOperation.GetActivityOptions() != nil && op.UpdateActivityOptionsOperation.GetRestoreOriginal() {
+			return serviceerror.NewInvalidArgument("cannot set both activity options and restore original")
+		}
+		if op.UpdateActivityOptionsOperation.GetActivityOptions() == nil && !op.UpdateActivityOptionsOperation.GetRestoreOriginal() {
+			return serviceerror.NewInvalidArgument("Either activity type must be set, or restore original should be set to true")
+		}
+
+		switch a := op.UpdateActivityOptionsOperation.GetActivity().(type) {
+		case *batchpb.BatchOperationUpdateActivityOptions_Type:
+			if len(a.Type) == 0 {
+				return serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
+			}
+		case *batchpb.BatchOperationUpdateActivityOptions_MatchAll:
+			if !a.MatchAll {
+				return serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
+			}
+		}
+		return nil
+	default:
+		return fmt.Errorf("not supported batch type: %v", params.GetOperation())
+	}
+	return nil
+}
+
 func setDefaultParams(params BatchParams) BatchParams {
 	if params.AttemptsOnRetryableError <= 1 {
 		params.AttemptsOnRetryableError = defaultAttemptsOnRetryableError
@@ -342,3 +501,15 @@ func setDefaultParams(params BatchParams) BatchParams {
 	}
 	return params
 }
+
+func setDefaultParamsProtobuf(params *batchspb.BatchOperationInput) *batchspb.BatchOperationInput {
+	if params.GetAttemptsOnRetryableError() <= 1 {
+		params.AttemptsOnRetryableError = defaultAttemptsOnRetryableError
+	}
+	if params.GetActivityHeartbeatTimeout().AsDuration() <= 0 {
+		params.ActivityHeartbeatTimeout = &durationpb.Duration{
+			Seconds: int64(defaultActivityHeartBeatTimeout / time.Second),
+		}
+	}
+	return params
+}

service/worker/batcher/workflow_test.go

@@ -6,8 +6,12 @@ import (
 	"github.com/pborman/uuid"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
+	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/testsuite"
+	batchspb "go.temporal.io/server/api/batch/v1"
 	"go.uber.org/mock/gomock"
 )
 
@@ -26,6 +30,7 @@ func (s *batcherSuite) SetupTest() {
 	s.controller = gomock.NewController(s.T())
 	s.env = s.WorkflowTestSuite.NewTestWorkflowEnvironment()
 	s.env.RegisterWorkflow(BatchWorkflow)
+	s.env.RegisterWorkflow(BatchWorkflowProtobuf)
 }
 
 func (s *batcherSuite) TearDownTest() {
@@ -40,6 +45,13 @@ func (s *batcherSuite) TestBatchWorkflow_MissingParams() {
 	s.Contains(err.Error(), "must provide required parameters")
 }
 
+func (s *batcherSuite) TestBatchWorkflow_MissingParams_Protobuf() {
+	s.env.ExecuteWorkflow(BatchWorkflowProtobuf, &batchspb.BatchOperationInput{})
+	err := s.env.GetWorkflowError()
+	s.Require().Error(err)
+	s.Contains(err.Error(), "must provide required parameters")
+}
+
 func (s *batcherSuite) TestBatchWorkflow_ValidParams_Query() {
 	var ac *activities
 	s.env.OnActivity(ac.BatchActivity, mock.Anything, mock.Anything).Return(HeartBeatDetails{
@@ -66,6 +78,38 @@ func (s *batcherSuite) TestBatchWorkflow_ValidParams_Query() {
 	s.Require().NoError(err)
 }
 
+func (s *batcherSuite) TestBatchWorkflow_ValidParams_Query_Protobuf() {
+	var ac *activities
+	s.env.OnActivity(ac.BatchActivityWithProtobuf, mock.Anything, mock.Anything).Return(HeartBeatDetails{
+		SuccessCount: 42,
+		ErrorCount:   27,
+	}, nil)
+	s.env.OnUpsertMemo(mock.Anything).Run(func(args mock.Arguments) {
+		memo, ok := args.Get(0).(map[string]interface{})
+		s.Require().True(ok)
+		s.Equal(map[string]interface{}{
+			"batch_operation_stats": BatchOperationStats{
+				NumSuccess: 42,
+				NumFailure: 27,
+			},
+		}, memo)
+	}).Once()
+	s.env.ExecuteWorkflow(BatchWorkflowProtobuf, &batchspb.BatchOperationInput{
+		Request: &workflowservice.StartBatchOperationRequest{
+			JobId: uuid.New(),
+			Operation: &workflowservice.StartBatchOperationRequest_TerminationOperation{
+				TerminationOperation: &batchpb.BatchOperationTermination{},
+			},
+			Namespace:       "test-namespace",
+			Reason:          "test-reason",
+			VisibilityQuery: "test-query",
+		},
+		BatchType: enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+	})
+	err := s.env.GetWorkflowError()
+	s.Require().NoError(err)
+}
+
 func (s *batcherSuite) TestBatchWorkflow_ValidParams_Executions() {
 	var ac *activities
 	s.env.OnActivity(ac.BatchActivity, mock.Anything, mock.Anything).Return(HeartBeatDetails{
@@ -96,3 +140,40 @@ func (s *batcherSuite) TestBatchWorkflow_ValidParams_Executions() {
 	err := s.env.GetWorkflowError()
 	s.Require().NoError(err)
 }
+
+func (s *batcherSuite) TestBatchWorkflow_ValidParams_Executions_Protobuf() {
+	var ac *activities
+	s.env.OnActivity(ac.BatchActivityWithProtobuf, mock.Anything, mock.Anything).Return(HeartBeatDetails{
+		SuccessCount: 42,
+		ErrorCount:   27,
+	}, nil)
+	s.env.OnUpsertMemo(mock.Anything).Run(func(args mock.Arguments) {
+		memo, ok := args.Get(0).(map[string]interface{})
+		s.Require().True(ok)
+		s.Equal(map[string]interface{}{
+			"batch_operation_stats": BatchOperationStats{
+				NumSuccess: 42,
+				NumFailure: 27,
+			},
+		}, memo)
+	}).Once()
+	s.env.ExecuteWorkflow(BatchWorkflowProtobuf, &batchspb.BatchOperationInput{
+		Request: &workflowservice.StartBatchOperationRequest{
+			JobId: uuid.New(),
+			Operation: &workflowservice.StartBatchOperationRequest_TerminationOperation{
+				TerminationOperation: &batchpb.BatchOperationTermination{},
+			},
+			Executions: []*commonpb.WorkflowExecution{
+				{
+					WorkflowId: uuid.New(),
+					RunId:      uuid.New(),
+				},
+			},
+			Reason:    "test-reason",
+			Namespace: "test-namespace",
+		},
+		BatchType: enumspb.BATCH_OPERATION_TYPE_TERMINATE,
+	})
+	err := s.env.GetWorkflowError()
+	s.Require().NoError(err)
+}

tests/activity_api_batch_reset_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"github.com/temporalio/sqlparser"
 	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	"go.temporal.io/api/serviceerror"
@@ -96,7 +97,137 @@ func (s *ActivityApiBatchResetClientTestSuite) TestActivityBatchReset_Success()
 
 	// Make sure the activity is in visibility
 	var listResp *workflowservice.ListWorkflowExecutionsResponse
-	resetCause := fmt.Sprintf("%s = 'property:activityType=%s'", searchattribute.TemporalPauseInfo, activityTypeName)
+	searchValue := fmt.Sprintf("property:activityType=%s", activityTypeName)
+	escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
+	resetCause := fmt.Sprintf("%s = %s", searchattribute.TemporalPauseInfo, escapedSearchValue)
+	query := fmt.Sprintf("(WorkflowType='%s' AND %s)", workflowTypeName, resetCause)
+
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		listResp, err = s.FrontendClient().ListWorkflowExecutions(ctx, &workflowservice.ListWorkflowExecutionsRequest{
+			Namespace: s.Namespace().String(),
+			PageSize:  10,
+			Query:     query,
+		})
+		require.NoError(t, err)
+		require.NotNil(t, listResp)
+		require.Len(t, listResp.GetExecutions(), 2)
+		require.Greater(t, internalWorkflow.startedActivityCount.Load(), int32(0))
+	}, 5*time.Second, 500*time.Millisecond)
+
+	// reset the activities in both workflows with batch reset
+	_, err = s.SdkClient().WorkflowService().StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{
+		Namespace: s.Namespace().String(),
+		Operation: &workflowservice.StartBatchOperationRequest_ResetActivitiesOperation{
+			ResetActivitiesOperation: &batchpb.BatchOperationResetActivities{
+				Activity:   &batchpb.BatchOperationResetActivities_Type{Type: activityTypeName},
+				KeepPaused: true,
+			},
+		},
+		VisibilityQuery: fmt.Sprintf("WorkflowType='%s'", workflowTypeName),
+		JobId:           uuid.New(),
+		Reason:          "test",
+	})
+	s.NoError(err)
+
+	// make sure activities are restarted and still paused
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun1.GetID(), workflowRun1.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.PendingActivities, 1)
+		require.True(t, description.PendingActivities[0].Paused)
+		require.Equal(t, description.PendingActivities[0].Attempt, int32(1))
+
+		description, err = s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun2.GetID(), workflowRun2.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.PendingActivities, 1)
+		require.True(t, description.PendingActivities[0].Paused)
+		require.Equal(t, description.PendingActivities[0].Attempt, int32(1))
+	}, 5*time.Second, 100*time.Millisecond)
+
+	// let activities succeed
+	internalWorkflow.letActivitySucceed.Store(true)
+
+	// reset the activities in both workflows with batch reset
+	_, err = s.SdkClient().WorkflowService().StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{
+		Namespace: s.Namespace().String(),
+		Operation: &workflowservice.StartBatchOperationRequest_ResetActivitiesOperation{
+			ResetActivitiesOperation: &batchpb.BatchOperationResetActivities{
+				Activity:   &batchpb.BatchOperationResetActivities_Type{Type: activityTypeName},
+				KeepPaused: false,
+			},
+		},
+		VisibilityQuery: fmt.Sprintf("WorkflowType='%s'", workflowTypeName),
+		JobId:           uuid.New(),
+		Reason:          "test",
+	})
+	s.NoError(err)
+
+	var out string
+	err = workflowRun1.Get(ctx, &out)
+	s.NoError(err)
+
+	err = workflowRun2.Get(ctx, &out)
+	s.NoError(err)
+}
+
+func (s *ActivityApiBatchResetClientTestSuite) TestActivityBatchReset_Success_Protobuf() {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	internalWorkflow := newInternalWorkflow()
+
+	s.Worker().RegisterWorkflow(internalWorkflow.WorkflowFunc)
+	s.Worker().RegisterActivity(internalWorkflow.ActivityFunc)
+
+	workflowRun1 := s.createWorkflow(ctx, internalWorkflow.WorkflowFunc)
+	workflowRun2 := s.createWorkflow(ctx, internalWorkflow.WorkflowFunc)
+
+	// wait for activity to start in both workflows
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun1.GetID(), workflowRun1.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.Greater(t, internalWorkflow.startedActivityCount.Load(), int32(0))
+
+		description, err = s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun2.GetID(), workflowRun2.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.Greater(t, internalWorkflow.startedActivityCount.Load(), int32(0))
+	}, 5*time.Second, 100*time.Millisecond)
+
+	// pause activities in both workflows
+	pauseRequest := &workflowservice.PauseActivityRequest{
+		Namespace: s.Namespace().String(),
+		Execution: &commonpb.WorkflowExecution{},
+		Activity:  &workflowservice.PauseActivityRequest_Id{Id: "activity-id"},
+	}
+
+	pauseRequest.Execution.WorkflowId = workflowRun1.GetID()
+	resp, err := s.FrontendClient().PauseActivity(ctx, pauseRequest)
+	s.NoError(err)
+	s.NotNil(resp)
+
+	pauseRequest.Execution.WorkflowId = workflowRun2.GetID()
+	resp, err = s.FrontendClient().PauseActivity(ctx, pauseRequest)
+	s.NoError(err)
+	s.NotNil(resp)
+
+	// wait for activities to be paused
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun1.GetID(), workflowRun1.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.True(t, description.PendingActivities[0].Paused)
+	}, 5*time.Second, 100*time.Millisecond)
+
+	workflowTypeName := "WorkflowFunc"
+	activityTypeName := "ActivityFunc"
+
+	// Make sure the activity is in visibility
+	var listResp *workflowservice.ListWorkflowExecutionsResponse
+	searchValue := fmt.Sprintf("property:activityType=%s", activityTypeName)
+	escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
+	resetCause := fmt.Sprintf("%s = %s", searchattribute.TemporalPauseInfo, escapedSearchValue)
 	query := fmt.Sprintf("(WorkflowType='%s' AND %s)", workflowTypeName, resetCause)
 
 	s.EventuallyWithT(func(t *assert.CollectT) {
@@ -222,7 +353,9 @@ func (s *ActivityApiBatchResetClientTestSuite) TestActivityBatchReset_DontResetA
 
 	// Make sure the activity is in visibility
 	var listResp *workflowservice.ListWorkflowExecutionsResponse
-	resetCause := fmt.Sprintf("%s = 'property:activityType=%s'", searchattribute.TemporalPauseInfo, activityTypeName)
+	searchValue := fmt.Sprintf("property:activityType=%s", activityTypeName)
+	escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
+	resetCause := fmt.Sprintf("%s = %s", searchattribute.TemporalPauseInfo, escapedSearchValue)
 	query := fmt.Sprintf("(WorkflowType='%s' AND %s)", workflowTypeName, resetCause)
 
 	s.EventuallyWithT(func(t *assert.CollectT) {

tests/activity_api_batch_unpause_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"github.com/temporalio/sqlparser"
 	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	"go.temporal.io/api/serviceerror"
@@ -143,7 +144,9 @@ func (s *ActivityApiBatchUnpauseClientTestSuite) TestActivityBatchUnpause_Succes
 	activityTypeName := "ActivityFunc"
 	// Make sure the activity is in visibility
 	var listResp *workflowservice.ListWorkflowExecutionsResponse
-	unpauseCause := fmt.Sprintf("%s = 'property:activityType=%s'", searchattribute.TemporalPauseInfo, activityTypeName)
+	searchValue := fmt.Sprintf("property:activityType=%s", activityTypeName)
+	escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
+	unpauseCause := fmt.Sprintf("%s = %s", searchattribute.TemporalPauseInfo, escapedSearchValue)
 	query := fmt.Sprintf("(WorkflowType='%s' AND %s)", workflowTypeName, unpauseCause)
 
 	s.EventuallyWithT(func(t *assert.CollectT) {

tests/activity_api_batch_update_options_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"github.com/temporalio/sqlparser"
 	activitypb "go.temporal.io/api/activity/v1"
 	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
@@ -97,7 +98,9 @@ func (s *ActivityApiBatchUpdateOptionsClientTestSuite) TestActivityBatchUpdateOp
 	activityTypeName := "ActivityFunc"
 	// Make sure the activity is in visibility
 	var listResp *workflowservice.ListWorkflowExecutionsResponse
-	unpauseCause := fmt.Sprintf("%s = 'property:activityType=%s'", searchattribute.TemporalPauseInfo, activityTypeName)
+	searchValue := fmt.Sprintf("property:activityType=%s", activityTypeName)
+	escapedSearchValue := sqlparser.String(sqlparser.NewStrVal([]byte(searchValue)))
+	unpauseCause := fmt.Sprintf("%s = %s", searchattribute.TemporalPauseInfo, escapedSearchValue)
 	query := fmt.Sprintf("(WorkflowType='%s' AND %s)", workflowTypeName, unpauseCause)
 
 	s.EventuallyWithT(func(t *assert.CollectT) {
@@ -121,7 +124,7 @@ func (s *ActivityApiBatchUpdateOptionsClientTestSuite) TestActivityBatchUpdateOp
 					ScheduleToCloseTimeout: durationpb.New(10 * time.Second),
 				},
 				UpdateMask: &fieldmaskpb.FieldMask{
-					Paths: []string{"scheduleToCloseTimeout"},
+					Paths: []string{"schedule_to_close_timeout"},
 				},
 			},
 		},