package server

import (
	"context"
	"errors"
	"testing"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	bractionmocks "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br/mocks"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/validation"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
)

const (
	testTKRWSharedPoolID  = "testpool"
	testTKRWNamespace     = "tidbtestpool"
	testTKRWComponentID   = "default-comp"
	testTKRWTiKVGroupName = "db-default-comp"
	testTKRWWorkerName    = "db-default-comp-default"
	testTKRWCoprName      = "db-default-comp-coprocessor"
)

// mockTKRWTransactioner runs fn to completion and lets it own the returned
// error, matching the real Transactioner's semantics for these tests.
type mockTKRWTransactioner struct{}

func (m *mockTKRWTransactioner) Transaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}

// mockTKRWCommitFailTransactioner runs fn to completion (so any Submit call
// inside fn really happens) but then forces a commit-phase error regardless
// of fn's own outcome - simulating a DB commit failure after a successful
// DAG submission.
type mockTKRWCommitFailTransactioner struct {
	commitErr error
}

func (m *mockTKRWCommitFailTransactioner) Transaction(ctx context.Context, fn func(context.Context) error) error {
	if err := fn(ctx); err != nil {
		return err
	}
	return m.commitErr
}

func tkrwEnabledBoth(t *testing.T) []apiv1.ComponentType {
	t.Helper()
	normalized, err := entity.NormalizeEnabledTiKVRemoteWorkers([]apiv1.ComponentType{
		apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_COPROCESSOR_WORKER,
	})
	require.NoError(t, err)
	return normalized
}

func tkrwEnabledDisableOnly() []apiv1.ComponentType {
	return []apiv1.ComponentType{apiv1.ComponentType_TIKV_WORKER}
}

func tkrwKey(t *testing.T, enabled []apiv1.ComponentType) string {
	t.Helper()
	key, err := entity.EnabledTiKVRemoteWorkersToString(enabled)
	require.NoError(t, err)
	return key
}

// newTKRWSharedPool builds a minimal AVAILABLE, RCU-driven shared pool with a
// PD/TIKV/TIKV_WORKER component set sharing the same default component ID
// (so sharedPoolDefaultComponentMap/markComponentsPhase recognize them), and
// an optional coprocessor component when coprocessorNodeCount > 0.
func newTKRWSharedPool(enabled []apiv1.ComponentType, coprocessorNodeCount uint32) *entity.SharedPool {
	sp := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id:                       testTKRWSharedPoolID,
		Phase:                    apiv1.SharedPool_AVAILABLE,
		CapacityMode:             apiv1.CapacityMode_RCU_DRIVEN,
		EnabledTikvRemoteWorkers: enabled,
		ComponentGroups: []*apiv1.Component{
			{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_PD, Phase: apiv1.Component_AVAILABLE},
			{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_TIKV, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{NodeCount: 3}},
			{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_TIKV_WORKER, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{NodeCount: 3}},
		},
	}}
	if coprocessorNodeCount > 0 {
		sp.ComponentGroups = append(sp.ComponentGroups, &apiv1.Component{
			Id:            testTKRWComponentID,
			ComponentType: apiv1.ComponentType_COPROCESSOR_WORKER,
			Phase:         apiv1.Component_AVAILABLE,
			NodeQuota:     &apiv1.NodeQuota{NodeCount: coprocessorNodeCount},
		})
	}
	return sp
}

func tkrwFakeScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1.AddToScheme(scheme))
	require.NoError(t, corev1.AddToScheme(scheme))
	return scheme
}

// newTKRWDataplaneManager builds a dataplanemocks.Manager backed by a real
// fake controller-runtime client seeded with objs, following the same
// technique as domain/workflow/action/wait_coprocessor_worker_test.go.
func newTKRWDataplaneManager(t *testing.T, objs ...client.Object) *dataplanemocks.Manager {
	t.Helper()
	cli := fake.NewClientBuilder().WithScheme(tkrwFakeScheme(t)).WithObjects(objs...).Build()
	dm := dataplanemocks.NewManager(t)
	dm.On("GetClient", mock.Anything).Return(cli, nil).Maybe()
	return dm
}

func tkrwReadyTiKVGroup(coprocessorName string) *v1alpha1.TiKVGroup {
	return &v1alpha1.TiKVGroup{
		ObjectMeta: metav1.ObjectMeta{Name: testTKRWTiKVGroupName, Namespace: testTKRWNamespace, Generation: 1},
		Spec: v1alpha1.TiKVGroupSpec{
			Replicas: int32Ptr(3),
			Template: v1alpha1.TiKVTemplate{Spec: v1alpha1.TiKVTemplateSpec{
				RemoteWorkers: &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: testTKRWWorkerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprocessorName},
				},
			}},
		},
		Status: v1alpha1.TiKVGroupStatus{
			CommonStatus: v1alpha1.CommonStatus{
				ObservedGeneration: 1, CurrentRevision: "rev-1", UpdateRevision: "rev-1",
				Conditions: []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionTrue, ObservedGeneration: 1},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionTrue, ObservedGeneration: 1},
				},
			},
			GroupStatus: v1alpha1.GroupStatus{Replicas: 3, ReadyReplicas: 3, UpdatedReplicas: 3, CurrentReplicas: 3},
		},
	}
}

func tkrwReadyWorkerGroup(name string) *v1alpha1.TiKVWorkerGroup {
	return &v1alpha1.TiKVWorkerGroup{
		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testTKRWNamespace, Generation: 1},
		Spec:       v1alpha1.TiKVWorkerGroupSpec{Replicas: int32Ptr(3)},
		Status: v1alpha1.TiKVWorkerGroupStatus{
			CommonStatus: v1alpha1.CommonStatus{ObservedGeneration: 1, CurrentRevision: "rev-1", UpdateRevision: "rev-1"},
			GroupStatus:  v1alpha1.GroupStatus{Replicas: 3, ReadyReplicas: 3, UpdatedReplicas: 3, CurrentReplicas: 3},
		},
	}
}

// ---------------------------------------------------------------------------
// Pure-function / pure-method tests (no mocks required).
// ---------------------------------------------------------------------------

func TestHasSameTiKVRemoteWorkersTarget(t *testing.T) {
	runningDagWithKey := func(key string) executor.ExecTaskIdStatus {
		return executor.ExecTaskIdStatus{
			"task-1": &executor.ExecTaskStatusEx{
				ExecTaskStatus: executor.ExecTaskStatus{
					ExecTask: &executor.ExecutorTask{
						Args: actions.ActionArgs{
							workflow.EnabledTiKVRemoteWorkersKey: actions.ActionArgString(key),
						},
					},
				},
			},
		}
	}

	t.Run("empty running dags - false, no error", func(t *testing.T) {
		same, err := hasSameTiKVRemoteWorkersTarget(executor.NewExecTaskIdStatus(), "TIKV_WORKER")
		require.NoError(t, err)
		assert.False(t, same)
	})

	t.Run("single running dag matches target - true", func(t *testing.T) {
		same, err := hasSameTiKVRemoteWorkersTarget(runningDagWithKey("TIKV_WORKER"), "TIKV_WORKER")
		require.NoError(t, err)
		assert.True(t, same)
	})

	t.Run("single running dag targets something else - false", func(t *testing.T) {
		same, err := hasSameTiKVRemoteWorkersTarget(runningDagWithKey("TIKV_WORKER,COPROCESSOR_WORKER"), "TIKV_WORKER")
		require.NoError(t, err)
		assert.False(t, same)
	})

	t.Run("mixed running dags - one matches, one doesn't - unanimous match required, false", func(t *testing.T) {
		mixed := executor.ExecTaskIdStatus{
			"task-1": &executor.ExecTaskStatusEx{ExecTaskStatus: executor.ExecTaskStatus{ExecTask: &executor.ExecutorTask{
				Args: actions.ActionArgs{workflow.EnabledTiKVRemoteWorkersKey: actions.ActionArgString("TIKV_WORKER")},
			}}},
			"task-2": &executor.ExecTaskStatusEx{ExecTaskStatus: executor.ExecTaskStatus{ExecTask: &executor.ExecutorTask{
				Args: actions.ActionArgs{workflow.EnabledTiKVRemoteWorkersKey: actions.ActionArgString("TIKV_WORKER,COPROCESSOR_WORKER")},
			}}},
		}
		same, err := hasSameTiKVRemoteWorkersTarget(mixed, "TIKV_WORKER")
		require.NoError(t, err)
		assert.False(t, same, "a mixed in-flight state must never be trusted, even when one running dag happens to match")
	})

	t.Run("missing args key - error", func(t *testing.T) {
		broken := executor.ExecTaskIdStatus{
			"task-1": &executor.ExecTaskStatusEx{ExecTaskStatus: executor.ExecTaskStatus{ExecTask: &executor.ExecutorTask{
				Args: actions.ActionArgs{},
			}}},
		}
		_, err := hasSameTiKVRemoteWorkersTarget(broken, "TIKV_WORKER")
		require.Error(t, err)
	})
}

func TestBuildTiKVRemoteWorkersArgs(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	sharedPool.Phase = apiv1.SharedPool_MODIFYING

	args, err := buildTiKVRemoteWorkersArgs(context.Background(), sharedPool, "TIKV_WORKER")
	require.NoError(t, err)

	componentTypes, err := args.GetString(workflow.ComponentTypesKey)
	require.NoError(t, err)
	assert.Equal(t, entity.CoprocessorWorkerType, componentTypes,
		"the mutation target snapshot must only ever name the coprocessor worker as the component type being mutated, "+
			"using the runtime type name that CheckIsTargetComponent compares against entity.GetComponentTypeName")

	affected, err := args.GetString(workflow.AffectedComponentTypesKey)
	require.NoError(t, err)
	assert.Equal(t, "tikv,"+entity.CoprocessorWorkerType, affected,
		"affected types must cover both TIKV (routing) and COPROCESSOR_WORKER (lifecycle), "+
			"using the runtime type names that CheckIsAffectedComponent compares against entity.GetComponentTypeName")

	enabledKey, err := args.GetString(workflow.EnabledTiKVRemoteWorkersKey)
	require.NoError(t, err)
	assert.Equal(t, "TIKV_WORKER", enabledKey)

	sharedPoolIDArg, err := args.GetString(workflow.SharedPoolIdKey)
	require.NoError(t, err)
	assert.Equal(t, testTKRWSharedPoolID, sharedPoolIDArg)

	meteringType, err := args.GetString(workflow.MeteringTypeKey)
	require.NoError(t, err)
	assert.Equal(t, workflow.MeteringTypeValueLogicMetaBySharedPool, meteringType)

	componentIDs, err := args.GetString(workflow.ComponentIdsKey)
	require.NoError(t, err)
	assert.Equal(t, testTKRWComponentID, componentIDs)
}

// TestBuildTiKVRemoteWorkersArgs_ComponentTypeFilters pipes the real args
// builder output through the actual runtime filters. The filter functions are
// unit-tested in domain/workflow with hand-written runtime type names; this
// test guards the contract between the args builder and those filters, which
// is where the proto-enum-name vs runtime-name mismatch slipped through.
func TestBuildTiKVRemoteWorkersArgs_ComponentTypeFilters(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPool.Phase = apiv1.SharedPool_MODIFYING

	args, err := buildTiKVRemoteWorkersArgs(context.Background(), sharedPool, tkrwKey(t, tkrwEnabledDisableOnly()))
	require.NoError(t, err)

	ctx := context.Background()
	compByType := sharedPoolDefaultComponentMap(sharedPool)
	tikv := compByType[apiv1.ComponentType_TIKV]
	tikvWorker := compByType[apiv1.ComponentType_TIKV_WORKER]
	coprocessor := compByType[apiv1.ComponentType_COPROCESSOR_WORKER]
	require.NotNil(t, tikv)
	require.NotNil(t, tikvWorker)
	require.NotNil(t, coprocessor)
	tidb := &entity.Component{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_TIDB}

	// Mutation actions (apply-component-group, apply-component-hpa,
	// clean-up-cert, create-cert) must select the coprocessor worker only;
	// TIKV's remoteWorkers patch flows through the per-component update key.
	assert.True(t, workflow.CheckIsTargetComponent(ctx, coprocessor, args),
		"the coprocessor worker must pass the target filter or the DAG never creates/deletes its group")
	assert.False(t, workflow.CheckIsTargetComponent(ctx, tikv, args))
	assert.False(t, workflow.CheckIsTargetComponent(ctx, tikvWorker, args),
		"the default tikv worker must never be a mutation target of this DAG")
	assert.False(t, workflow.CheckIsTargetComponent(ctx, tidb, args))

	// The running-DAG overlay must expose both TIKV (routing switch) and
	// COPROCESSOR_WORKER (lifecycle) as in-flight.
	assert.True(t, workflow.CheckIsAffectedComponent(ctx, tikv, args))
	assert.True(t, workflow.CheckIsAffectedComponent(ctx, coprocessor, args))
	assert.False(t, workflow.CheckIsAffectedComponent(ctx, tikvWorker, args))
	assert.False(t, workflow.CheckIsAffectedComponent(ctx, tidb, args))
}

func TestBuildTiKVRemoteWorkersTargetSharedPool_DisableRepairsMissingCoprocessorMetadata(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	filtered := sharedPool.ComponentGroups[:0]
	for _, component := range sharedPool.ComponentGroups {
		if component.GetComponentType() != apiv1.ComponentType_COPROCESSOR_WORKER {
			filtered = append(filtered, component)
		}
	}
	sharedPool.ComponentGroups = filtered

	snapshot, err := (&serverImpl{}).buildTiKVRemoteWorkersTargetSharedPool(
		context.Background(), sharedPool, tkrwEnabledDisableOnly(), false,
	)
	require.NoError(t, err)

	coprocessor := sharedPoolDefaultComponentMap(snapshot)[apiv1.ComponentType_COPROCESSOR_WORKER]
	require.NotNil(t, coprocessor)
	assert.Equal(t, defaultSharedPoolComponentID(snapshot), coprocessor.GetId())
	assert.Equal(t, apiv1.Component_DELETING, coprocessor.GetPhase())
	assert.NotNil(t, coprocessor.GetNodeQuota(), "deleting render still reads node quota fields")
}

func TestBuildTiKVRemoteWorkersArgs_TiKVComponentMissing(t *testing.T) {
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{Id: testTKRWSharedPoolID}}
	_, err := buildTiKVRemoteWorkersArgs(context.Background(), sharedPool, "TIKV_WORKER")
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.Internal, statusErr.Code())
}

func TestTikvRemoteWorkersDAGPlansAndHelpers(t *testing.T) {
	assert.ElementsMatch(t, []string{dag.PlanEnableCoprocessorWorker, dag.PlanDisableCoprocessorWorker}, tikvRemoteWorkersDAGPlans())
	assert.Equal(t, dag.PlanEnableCoprocessorWorker, tikvRemoteWorkersPlanForTarget(true))
	assert.Equal(t, dag.PlanDisableCoprocessorWorker, tikvRemoteWorkersPlanForTarget(false))
	assert.ElementsMatch(t, []entity.ComponentType{apiv1.ComponentType_TIKV, apiv1.ComponentType_COPROCESSOR_WORKER}, tikvRemoteWorkersAffectedComponentTypes())
}

// ---------------------------------------------------------------------------
// isTiKVRemoteWorkersStableInTarget - direct method tests via fake dataplane.
// ---------------------------------------------------------------------------

func TestIsTiKVRemoteWorkersStableInTarget(t *testing.T) {
	tikvComp := &entity.Component{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_TIKV, NodeQuota: &apiv1.NodeQuota{NodeCount: 3}}

	t.Run("persisted key mismatch - not stable, no live call needed", func(t *testing.T) {
		sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
		dm := dataplanemocks.NewManager(t) // no .On(...) - must not be called
		s := &serverImpl{dataplaneManager: dm}

		enabled := tkrwEnabledBoth(t)
		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, enabled, tkrwKey(t, enabled), true)
		require.NoError(t, err)
		assert.False(t, stable)
	})

	t.Run("disable target - persisted matches, tikv group already routed to target, coprocessor CR gone - stable", func(t *testing.T) {
		disable := tkrwEnabledDisableOnly()
		sharedPool := newTKRWSharedPool(disable, 0)
		tikvGroup := tkrwReadyTiKVGroup(testTKRWWorkerName)
		dm := newTKRWDataplaneManager(t, tikvGroup)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, disable, tkrwKey(t, disable), false)
		require.NoError(t, err)
		assert.True(t, stable)
	})

	t.Run("disable target - persisted matches, tikv group routed, but coprocessor CR still present - not stable", func(t *testing.T) {
		disable := tkrwEnabledDisableOnly()
		sharedPool := newTKRWSharedPool(disable, 0)
		tikvGroup := tkrwReadyTiKVGroup(testTKRWWorkerName)
		coprGroup := tkrwReadyWorkerGroup(testTKRWCoprName)
		dm := newTKRWDataplaneManager(t, tikvGroup, coprGroup)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, disable, tkrwKey(t, disable), false)
		require.NoError(t, err)
		assert.False(t, stable, "cleanup of the dedicated coprocessor CR must complete before disable is considered stable")
	})

	t.Run("enable target - persisted matches, tikv group routed, coprocessor ready - stable", func(t *testing.T) {
		enabled := tkrwEnabledBoth(t)
		sharedPool := newTKRWSharedPool(enabled, 3)
		tikvGroup := tkrwReadyTiKVGroup(testTKRWCoprName)
		coprGroup := tkrwReadyWorkerGroup(testTKRWCoprName)
		dm := newTKRWDataplaneManager(t, tikvGroup, coprGroup)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, enabled, tkrwKey(t, enabled), true)
		require.NoError(t, err)
		assert.True(t, stable)
	})

	t.Run("enable target - coprocessor CR not ready yet - not stable", func(t *testing.T) {
		enabled := tkrwEnabledBoth(t)
		sharedPool := newTKRWSharedPool(enabled, 3)
		tikvGroup := tkrwReadyTiKVGroup(testTKRWCoprName)
		coprGroup := tkrwReadyWorkerGroup(testTKRWCoprName)
		coprGroup.Status.ReadyReplicas = 2
		dm := newTKRWDataplaneManager(t, tikvGroup, coprGroup)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, enabled, tkrwKey(t, enabled), true)
		require.NoError(t, err)
		assert.False(t, stable)
	})

	t.Run("tikv group rollout still pending (names not yet updated) - not stable, no error", func(t *testing.T) {
		disable := tkrwEnabledDisableOnly()
		sharedPool := newTKRWSharedPool(disable, 0)
		// Live CR still routed to the old (enabled) target rather than the
		// persisted-and-requested disable target.
		tikvGroup := tkrwReadyTiKVGroup(testTKRWCoprName)
		dm := newTKRWDataplaneManager(t, tikvGroup)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, disable, tkrwKey(t, disable), false)
		require.NoError(t, err)
		assert.False(t, stable)
	})

	t.Run("tikv group CR not found - not stable, no error (rollout NotFound is not CheckFailed)", func(t *testing.T) {
		disable := tkrwEnabledDisableOnly()
		sharedPool := newTKRWSharedPool(disable, 0)
		dm := newTKRWDataplaneManager(t) // no objects seeded
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, disable, tkrwKey(t, disable), false)
		require.NoError(t, err)
		assert.False(t, stable)
	})

	t.Run("tikv group client error - rollout check failed - error propagated", func(t *testing.T) {
		disable := tkrwEnabledDisableOnly()
		sharedPool := newTKRWSharedPool(disable, 0)
		cli := fake.NewClientBuilder().
			WithScheme(tkrwFakeScheme(t)).
			WithInterceptorFuncs(interceptor.Funcs{
				Get: func(context.Context, client.WithWatch, client.ObjectKey, client.Object, ...client.GetOption) error {
					return errors.New("kubernetes unavailable")
				},
			}).
			Build()
		dm := dataplanemocks.NewManager(t)
		dm.On("GetClient", mock.Anything).Return(cli, nil)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, disable, tkrwKey(t, disable), false)
		require.Error(t, err)
		assert.False(t, stable)
	})

	t.Run("coprocessor worker group client error - readiness check failed - error propagated", func(t *testing.T) {
		disable := tkrwEnabledDisableOnly()
		sharedPool := newTKRWSharedPool(disable, 0)
		tikvGroup := tkrwReadyTiKVGroup(testTKRWWorkerName)
		cli := fake.NewClientBuilder().
			WithScheme(tkrwFakeScheme(t)).
			WithObjects(tikvGroup).
			WithInterceptorFuncs(interceptor.Funcs{
				Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
					if _, ok := obj.(*v1alpha1.TiKVWorkerGroup); ok {
						return errors.New("kubernetes unavailable")
					}
					return c.Get(ctx, key, obj, opts...)
				},
			}).
			Build()
		dm := dataplanemocks.NewManager(t)
		dm.On("GetClient", mock.Anything).Return(cli, nil)
		s := &serverImpl{dataplaneManager: dm}

		stable, err := s.isTiKVRemoteWorkersStableInTarget(context.Background(), sharedPool, tikvComp, disable, tkrwKey(t, disable), false)
		require.Error(t, err)
		assert.False(t, stable)
	})
}

// ---------------------------------------------------------------------------
// SetSharedPoolTiKVRemoteWorkers - handler-level tests.
// ---------------------------------------------------------------------------

// newTKRWHandlerMocks wires the routine plumbing every handler test needs:
// Load returns sharedPool, Lock succeeds, the mergeSharedPoolStatusFromDagTask
// sweep and the handler's own running-dag lookup both see no running dags by
// default, and no scale decision is in flight. Callers layer scenario-specific
// expectations (a matching/conflicting running dag, Submit, Abort) on top -
// and since these are strict mockery mocks, any call nobody expected panics
// the test, so simply omitting an .On("Submit"/"Abort", ...) is itself the
// assertion that it must not be called.
func newTKRWHandlerMocks(t *testing.T, sharedPool *entity.SharedPool) (*repomocks.SharedPoolRepo, *repomocks.LockRepo, *executormocks.RemoteExecutor, *bractionmocks.ScaleManager) {
	t.Helper()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
	lockRepo := repomocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	scaleManager := bractionmocks.NewScaleManager(t)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
	return sharedPoolRepo, lockRepo, dagExecutor, scaleManager
}

// tkrwOtherDagPlans lists every dag plan mergeSharedPoolStatusFromDagTaskExceptRestore
// sweeps over besides the two tikv-remote-worker plans.
var tkrwOtherDagPlans = []string{
	dag.PlanCreateSharedPool, dag.PlanDeleteSharedPool,
	dag.PlanUpdateComponent, dag.PlanApplyComponent, dag.PlanCreateComponent, dag.PlanDeleteTiKVGroup,
	dag.PlanCreateTiDBCluster, dag.PlanUpdateTiDBCluster, dag.PlanDeleteTiDBCluster,
	dag.PlanCreateTiDBGroup, dag.PlanUpdateTiDBGroup, dag.PlanDeleteTiDBGroup,
	dag.PlanSetSharedPoolPDDeploymentModeToMonolithic, dag.PlanSetSharedPoolPDDeploymentModeToMicroservice,
}

// newTKRWHandlerMocksWithRunningDag is like newTKRWHandlerMocks, but wires
// dagExecutor with per-plan QueryRunnings expectations instead of one
// catch-all, so runningPlan can be given a distinct (non-empty) result from
// its sibling tikv-remote-worker plan. A single catch-all could not do this:
// testify's mock resolves QueryRunnings calls in .On(...) registration order
// and returns the first argument-matching, non-exhausted expectation, so a
// wildcard registered before a same-signature specific one would always win
// and the specific one would never be reached.
func newTKRWHandlerMocksWithRunningDag(
	t *testing.T, sharedPool *entity.SharedPool, runningPlan string, runningStatus executor.ExecTaskIdStatus,
) (*repomocks.SharedPoolRepo, *repomocks.LockRepo, *executormocks.RemoteExecutor, *bractionmocks.ScaleManager) {
	t.Helper()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
	lockRepo := repomocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

	dagExecutor := executormocks.NewRemoteExecutor(t)
	for _, planName := range tkrwOtherDagPlans {
		dagExecutor.On("QueryRunnings", mock.Anything, planName, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	}
	for _, planName := range tikvRemoteWorkersDAGPlans() {
		if planName == runningPlan {
			dagExecutor.On("QueryRunnings", mock.Anything, planName, mock.Anything).Return(runningStatus, nil)
		} else {
			dagExecutor.On("QueryRunnings", mock.Anything, planName, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		}
	}

	scaleManager := bractionmocks.NewScaleManager(t)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
	return sharedPoolRepo, lockRepo, dagExecutor, scaleManager
}

func TestSetSharedPoolTiKVRemoteWorkers_Validation(t *testing.T) {
	s := &serverImpl{}

	t.Run("empty sharedPoolID", func(t *testing.T) {
		_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), "", tkrwEnabledDisableOnly())
		require.Error(t, err)
		statusErr, ok := status.FromError(err)
		require.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, statusErr.Code())
	})

	t.Run("empty target workers", func(t *testing.T) {
		_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, nil)
		require.Error(t, err)
		statusErr, ok := status.FromError(err)
		require.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, statusErr.Code())
	})

	t.Run("target missing TIKV_WORKER", func(t *testing.T) {
		_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID,
			[]apiv1.ComponentType{apiv1.ComponentType_COPROCESSOR_WORKER})
		require.Error(t, err)
		statusErr, ok := status.FromError(err)
		require.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, statusErr.Code())
	})

	t.Run("unsupported worker type", func(t *testing.T) {
		_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID,
			[]apiv1.ComponentType{apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_TIDB})
		require.Error(t, err)
		statusErr, ok := status.FromError(err)
		require.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, statusErr.Code())
	})
}

func TestSetSharedPoolTiKVRemoteWorkers_SharedPoolNotFound(t *testing.T) {
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).
		Return(nil, status.Errorf(codes.NotFound, "sharedPool %s not found", testTKRWSharedPoolID))
	lockRepo := repomocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		transactioner:  &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.NotFound, statusErr.Code())
}

func TestSetSharedPoolTiKVRemoteWorkers_NotRCUDriven(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	sharedPool.CapacityMode = apiv1.CapacityMode_TOPOLOGY_DRIVEN
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

// TestGuardTiKVRemoteWorkersTopologyStable_NonRCUNoOp verifies the guard
// itself (not the SetSharedPoolTiKVRemoteWorkers RPC) is a no-op for
// non-RCU-driven shared pools - unlike the RPC, which explicitly rejects
// non-RCU pools, callers of the guard (create/update/scale) must be allowed
// to proceed untouched since there is no tikv-remote-worker topology to
// protect on a topology-driven/dedicated pool.
func TestGuardTiKVRemoteWorkersTopologyStable_NonRCUNoOp(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	sharedPool.CapacityMode = apiv1.CapacityMode_TOPOLOGY_DRIVEN

	// No mocks wired on any dependency - the guard must return before making
	// any call (getRunningDagIns, sharedPoolDefaultComponentMap lookups, or
	// the live-CR drift check), since serverImpl{} here has every repo/client
	// field left nil.
	s := &serverImpl{}
	err := s.guardTiKVRemoteWorkersTopologyStable(context.Background(), sharedPool)
	require.NoError(t, err)
}

// TestGuardTiKVRemoteWorkersTopologyStable_CreatingSkipsLiveDriftCheck verifies
// that a shared pool still in SharedPool_CREATING is never rejected for live
// topology drift, even though its metadata already has a TIKV component (as
// generatedSharedPoolComponent(..., force=true) populates during creation)
// while no live CR exists yet. Only the running-plan race check applies
// during CREATING; dataplaneManager is left completely unmocked so any call
// into the live-drift path (which would need it) fails the test outright.
func TestGuardTiKVRemoteWorkersTopologyStable_CreatingSkipsLiveDriftCheck(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPool.Phase = apiv1.SharedPool_CREATING

	dagExecutor := executormocks.NewRemoteExecutor(t)
	setupAllDAGMocks(dagExecutor)

	s := &serverImpl{dagExecutor: dagExecutor}
	err := s.guardTiKVRemoteWorkersTopologyStable(context.Background(), sharedPool)
	require.NoError(t, err)
}

// TestGuardTiKVRemoteWorkersTopologyStable_CreatingStillBlocksOnRunningPlan
// verifies that the CREATING-phase live-drift skip does not also skip the
// running-plan race check - an in-flight tikv-remote-worker routing switch
// must still block, regardless of shared pool phase.
func TestGuardTiKVRemoteWorkersTopologyStable_CreatingStillBlocksOnRunningPlan(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPool.Phase = apiv1.SharedPool_CREATING

	dagExecutor := executormocks.NewRemoteExecutor(t)
	setupAllDAGMocksWithConflict(dagExecutor, dag.PlanDisableCoprocessorWorker, runningDagStatus(tkrwKey(t, tkrwEnabledBoth(t))))

	s := &serverImpl{dagExecutor: dagExecutor}
	err := s.guardTiKVRemoteWorkersTopologyStable(context.Background(), sharedPool)
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}

func TestSetSharedPoolTiKVRemoteWorkers_TiKVComponentMissing(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	sharedPool.ComponentGroups = []*apiv1.Component{
		{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_PD, Phase: apiv1.Component_AVAILABLE},
	}
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

func TestSetSharedPoolTiKVRemoteWorkers_SharedPoolNotAvailable(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	sharedPool.Phase = apiv1.SharedPool_MODIFYING
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

// runningDagStatus builds a QueryRunnings result whose single task carries
// EnabledTiKVRemoteWorkersKey=key, matching common_test.go's fixture style.
func runningDagStatus(key string) executor.ExecTaskIdStatus {
	return executor.ExecTaskIdStatus{
		"running-task": &executor.ExecTaskStatusEx{
			ExecTaskStatus: executor.ExecTaskStatus{
				ExecTask: &executor.ExecutorTask{
					Args: actions.ActionArgs{
						workflow.EnabledTiKVRemoteWorkersKey: actions.ActionArgString(key),
					},
				},
			},
		},
	}
}

func TestSetSharedPoolTiKVRemoteWorkers_RunningDagMatchesTarget(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	target := tkrwEnabledDisableOnly()
	targetKey := tkrwKey(t, target)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocksWithRunningDag(
		t, sharedPool, dag.PlanDisableCoprocessorWorker, runningDagStatus(targetKey),
	)

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockTKRWTransactioner{},
	}
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, target)
	require.NoError(t, err, "Submit/Abort are not stubbed on dagExecutor at all - a call to either would panic the mock, so a clean pass here proves neither was invoked")
	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
	assert.Equal(t, target, result.GetEnabledTikvRemoteWorkers())
}

func TestSetSharedPoolTiKVRemoteWorkers_RunningDagConflict(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	target := tkrwEnabledDisableOnly()
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocksWithRunningDag(
		t, sharedPool, dag.PlanEnableCoprocessorWorker, runningDagStatus(tkrwKey(t, tkrwEnabledBoth(t))),
	)

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, target)
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

// TestSetSharedPoolTiKVRemoteWorkers_StableNoOp verifies Tech-Leader's 1st
// mandated assertion: a stable no-op must not call Submit. dagExecutor has no
// Submit/Abort expectation registered at all, so either call would panic.
func TestSetSharedPoolTiKVRemoteWorkers_StableNoOp(t *testing.T) {
	disable := tkrwEnabledDisableOnly()
	sharedPool := newTKRWSharedPool(disable, 0)
	tikvGroup := tkrwReadyTiKVGroup(testTKRWWorkerName)
	dm := newTKRWDataplaneManager(t, tikvGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, disable)
	require.NoError(t, err)
	assert.Equal(t, apiv1.SharedPool_AVAILABLE, result.GetPhase())
	assert.Equal(t, disable, result.GetEnabledTikvRemoteWorkers())
}

func TestSetSharedPoolTiKVRemoteWorkers_DisableDefaultWorkerMissing(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPool.ComponentGroups = []*apiv1.Component{
		{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_PD, Phase: apiv1.Component_AVAILABLE},
		{Id: testTKRWComponentID, ComponentType: apiv1.ComponentType_TIKV, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{NodeCount: 3}},
	}
	dm := newTKRWDataplaneManager(t)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

func TestSetSharedPoolTiKVRemoteWorkers_DisableDefaultWorkerNodeCountZero(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	for _, comp := range sharedPool.ComponentGroups {
		if comp.GetComponentType() == apiv1.ComponentType_TIKV_WORKER {
			comp.NodeQuota = &apiv1.NodeQuota{NodeCount: 0}
		}
	}
	dm := newTKRWDataplaneManager(t)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

func TestSetSharedPoolTiKVRemoteWorkers_DisableDefaultWorkerNotReady(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	dm := newTKRWDataplaneManager(t) // no TiKVWorkerGroup CR seeded - NotFound
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.FailedPrecondition, statusErr.Code())
}

// TestSetSharedPoolTiKVRemoteWorkers_DisableSubmitSuccess drives a full real
// disable-direction rollout: persisted state (both enabled) does not match
// the disable target, so isTiKVRemoteWorkersStableInTarget short-circuits
// false on the persisted-key check alone (no live TiKV CR needed), the
// default worker is live-ready, and Submit is called exactly once with the
// disable plan name and a target snapshot naming only COPROCESSOR_WORKER.
func TestSetSharedPoolTiKVRemoteWorkers_DisableSubmitSuccess(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	workerGroup := tkrwReadyWorkerGroup(testTKRWWorkerName)
	dm := newTKRWDataplaneManager(t, workerGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()

	var submittedPlan string
	var submittedArgs actions.ActionArgs
	var submittedExecID string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			submittedPlan = args.String(1)
			submittedArgs = args.Get(2).(actions.ActionArgs)
			submittedExecID = args.String(3)
		}).
		Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	target := tkrwEnabledDisableOnly()
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, target)
	require.NoError(t, err)

	assert.Equal(t, dag.PlanDisableCoprocessorWorker, submittedPlan)
	assert.Empty(t, validation.IsDNS1123Subdomain(submittedExecID))
	assert.Contains(t, submittedExecID, dag.PlanDisableCoprocessorWorker)
	componentTypes, argErr := submittedArgs.GetString(workflow.ComponentTypesKey)
	require.NoError(t, argErr)
	assert.Equal(t, entity.CoprocessorWorkerType, componentTypes)
	enabledKey, argErr := submittedArgs.GetString(workflow.EnabledTiKVRemoteWorkersKey)
	require.NoError(t, argErr)
	assert.Equal(t, tkrwKey(t, target), enabledKey)

	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
	for _, comp := range result.GetComponentGroups() {
		switch comp.GetComponentType() {
		case apiv1.ComponentType_TIKV:
			assert.Equal(t, apiv1.Component_MODIFYING, comp.GetPhase())
		case apiv1.ComponentType_COPROCESSOR_WORKER:
			assert.Equal(t, apiv1.Component_DELETING, comp.GetPhase())
		}
	}
}

func TestSetSharedPoolTiKVRemoteWorkers_SubmissionExecIDIsUniquePerRequest(t *testing.T) {
	traceID := trace.TraceID{1}
	ctxWithTrace := trace.ContextWithSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{TraceID: traceID}))

	for _, tt := range []struct {
		name string
		ctx  context.Context
	}{
		{name: "no trace", ctx: context.Background()},
		{name: "same trace", ctx: ctxWithTrace},
	} {
		t.Run(tt.name, func(t *testing.T) {
			execIDs := make([]string, 0, 2)
			for range 2 {
				sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
				workerGroup := tkrwReadyWorkerGroup(testTKRWWorkerName)
				dm := newTKRWDataplaneManager(t, workerGroup)
				sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
				sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
				dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
					Run(func(args mock.Arguments) { execIDs = append(execIDs, args.String(3)) }).
					Return(nil).Once()

				s := &serverImpl{
					sharedPoolRepo:   sharedPoolRepo,
					lockRepo:         lockRepo,
					dagExecutor:      dagExecutor,
					scaleManager:     scaleManager,
					dataplaneManager: dm,
					transactioner:    &mockTKRWTransactioner{},
				}
				_, err := s.SetSharedPoolTiKVRemoteWorkers(tt.ctx, testTKRWSharedPoolID, tkrwEnabledDisableOnly())
				require.NoError(t, err)
			}

			require.Len(t, execIDs, 2)
			assert.NotEqual(t, execIDs[0], execIDs[1], "completed or aborted same-direction switches must not reuse executor IDs")
		})
	}
}

// TestSetSharedPoolTiKVRemoteWorkers_SubmitFailure verifies Tech-Leader's 2nd
// mandated assertion: when Submit fails, the DAG was never successfully
// created, so Abort must not be called. No Abort expectation is registered on
// dagExecutor at all, so a call to it would panic the mock.
func TestSetSharedPoolTiKVRemoteWorkers_SubmitFailure(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	workerGroup := tkrwReadyWorkerGroup(testTKRWWorkerName)
	dm := newTKRWDataplaneManager(t, workerGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Return(errors.New("submit unavailable")).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err, "Abort is not stubbed on dagExecutor at all - a call to it would panic the mock, so a clean error return here proves it was never invoked")
	statusErr, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.Internal, statusErr.Code())
}

// TestSetSharedPoolTiKVRemoteWorkers_CommitFailureAfterSubmitAborts verifies
// Tech-Leader's 3rd mandated assertion: once Submit has succeeded, a later
// transaction-commit failure must trigger exactly one Abort call carrying the
// same execID that was submitted.
func TestSetSharedPoolTiKVRemoteWorkers_CommitFailureAfterSubmitAborts(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	workerGroup := tkrwReadyWorkerGroup(testTKRWWorkerName)
	dm := newTKRWDataplaneManager(t, workerGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()

	var submittedExecID string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) { submittedExecID = args.String(3) }).
		Return(nil).Once()
	dagExecutor.On("Abort", mock.Anything, mock.MatchedBy(func(execID string) bool {
		return execID == submittedExecID && execID != ""
	})).Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWCommitFailTransactioner{commitErr: errors.New("commit failed")},
	}
	_, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledDisableOnly())
	require.Error(t, err)
	assert.Contains(t, err.Error(), "commit failed")
}

// ---------------------------------------------------------------------------
// Enable-direction real submission - dependency fixtures.
//
// The enable direction is the only path through buildTiKVRemoteWorkersTargetSharedPool
// that calls genDefaultTiKVWorkerComponent, which needs a ruConfigRepo,
// tidbClusterRepo (for total RU) and dataplaneClient (for node specs, since
// the fixture shared pool has no DataplaneName so genNodeSpecMap falls back
// to ListNodePoolConfigs) wired up.
// ---------------------------------------------------------------------------

// tkrwCoprocessorOnlyRuConfig gives the coprocessor worker a single 8C16G
// RU tier covering RU 0 (the fixture's zero-tidb-cluster total RU), with no
// explicit default node count and no max node count override.
func tkrwCoprocessorOnlyRuConfig() *entity.RuConfig {
	return &entity.RuConfig{RuConfig: apiv1.RuConfig{
		Components: map[string]*apiv1.RuConfig_ComponentsMap{
			"coprocessor_worker": {
				NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
					"8c16g": {Min: 0, Max: 100000000},
				},
			},
		},
	}}
}

// tkrwNodePoolConfigsResponse supplies a single dedicated 8C16G node spec,
// matching tkrwCoprocessorOnlyRuConfig's tier.
func tkrwNodePoolConfigsResponse() *dplanemgrv1.ListNodePoolConfigsResponse {
	return &dplanemgrv1.ListNodePoolConfigsResponse{
		NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{
			{
				CpuCore:               8,
				MemoryGi:              16,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     15000,
				NodeType:              dplanemgrv1.NodeType_Dedicated,
			},
		},
	}
}

// newTKRWEnableDeps wires the extra repos/clients the enable direction needs
// on top of newTKRWHandlerMocks/newTKRWHandlerMocksWithRunningDag's plumbing.
func newTKRWEnableDeps(t *testing.T) (*repomocks.RuConfigRepo, *repomocks.TiDBClusterRepo, *dataplanemgrmocks.Client) {
	t.Helper()
	ruConfigRepo := repomocks.NewRuConfigRepo(t)
	ruConfigRepo.On("GetRuConfig", mock.Anything, testTKRWSharedPoolID, apiv1.RuConfig_SHARED_POOL).
		Return(tkrwCoprocessorOnlyRuConfig(), nil)
	tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
	tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return(nil, nil)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(tkrwNodePoolConfigsResponse(), nil)
	return ruConfigRepo, tidbClusterRepo, dataplaneClient
}

// TestSetSharedPoolTiKVRemoteWorkers_EnableSubmitSuccess drives a full real
// enable-direction rollout from a from-scratch coprocessor component (no
// prior COPROCESSOR_WORKER entry) and asserts the resulting snapshot in
// detail: the new component carries the generated quota/node-spec/storage
// defaults, TIKV only has its Phase changed (its own node quota is
// untouched), and TIKV_WORKER/PD are left completely unchanged.
func TestSetSharedPoolTiKVRemoteWorkers_EnableSubmitSuccess(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)

	var origPD, origTiKV, origTiKVWorker *apiv1.Component
	for _, comp := range sharedPool.GetComponentGroups() {
		switch comp.GetComponentType() {
		case apiv1.ComponentType_PD:
			origPD = proto.Clone(comp).(*apiv1.Component)
		case apiv1.ComponentType_TIKV:
			origTiKV = proto.Clone(comp).(*apiv1.Component)
		case apiv1.ComponentType_TIKV_WORKER:
			origTiKVWorker = proto.Clone(comp).(*apiv1.Component)
		}
	}
	require.NotNil(t, origPD)
	require.NotNil(t, origTiKV)
	require.NotNil(t, origTiKVWorker)

	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	var savedSharedPool *entity.SharedPool
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) { savedSharedPool = args.Get(1).(*entity.SharedPool) }).
		Return(nil).Once()
	ruConfigRepo, tidbClusterRepo, dataplaneClient := newTKRWEnableDeps(t)

	var submittedPlan string
	var submittedArgs actions.ActionArgs
	var submittedExecID string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			submittedPlan = args.String(1)
			submittedArgs = args.Get(2).(actions.ActionArgs)
			submittedExecID = args.String(3)
		}).
		Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:  sharedPoolRepo,
		lockRepo:        lockRepo,
		dagExecutor:     dagExecutor,
		scaleManager:    scaleManager,
		ruConfigRepo:    ruConfigRepo,
		tidbClusterRepo: tidbClusterRepo,
		dataplaneClient: dataplaneClient,
		transactioner:   &mockTKRWTransactioner{},
	}
	target := tkrwEnabledBoth(t)
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, target)
	require.NoError(t, err)

	assert.Equal(t, dag.PlanEnableCoprocessorWorker, submittedPlan)
	assert.Empty(t, validation.IsDNS1123Subdomain(submittedExecID))
	assert.Contains(t, submittedExecID, dag.PlanEnableCoprocessorWorker)
	componentTypes, argErr := submittedArgs.GetString(workflow.ComponentTypesKey)
	require.NoError(t, argErr)
	assert.Equal(t, entity.CoprocessorWorkerType, componentTypes)
	enabledKey, argErr := submittedArgs.GetString(workflow.EnabledTiKVRemoteWorkersKey)
	require.NoError(t, argErr)
	assert.Equal(t, tkrwKey(t, target), enabledKey)

	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())

	var coprComp, tikvComp, tikvWorkerComp, pdComp *apiv1.Component
	for _, comp := range result.GetComponentGroups() {
		switch comp.GetComponentType() {
		case apiv1.ComponentType_COPROCESSOR_WORKER:
			coprComp = comp
		case apiv1.ComponentType_TIKV:
			tikvComp = comp
		case apiv1.ComponentType_TIKV_WORKER:
			tikvWorkerComp = comp
		case apiv1.ComponentType_PD:
			pdComp = comp
		}
	}

	require.NotNil(t, coprComp, "a new COPROCESSOR_WORKER component must be generated")
	assert.Equal(t, apiv1.Component_CREATING, coprComp.GetPhase())
	require.NotNil(t, coprComp.GetNodeQuota())
	assert.Equal(t, uint32(1), coprComp.GetNodeQuota().GetNodeCount())
	assert.Equal(t, uint32(1), coprComp.GetNodeQuota().GetMinNodeCount())
	assert.Equal(t, uint32(60), coprComp.GetNodeQuota().GetMaxNodeCount(), "falls back to entity.DefaultNodeCountQuota since the ru config sets no explicit max")
	require.NotNil(t, coprComp.GetNodeQuota().GetNodeSpec())
	assert.Equal(t, uint32(8), coprComp.GetNodeQuota().GetNodeSpec().GetCpu())
	assert.Equal(t, uint32(16), coprComp.GetNodeQuota().GetNodeSpec().GetMemoryGi())
	require.Len(t, coprComp.GetStorages(), 1, "a CREATING component with no prior storages must get the default coprocessor storage")
	assert.Equal(t, apiv1.Component_Storage_DATA, coprComp.GetStorages()[0].GetName())
	assert.Equal(t, uint32(200), coprComp.GetStorages()[0].GetSizeGi())
	assert.Equal(t, entity.GP3, coprComp.GetStorages()[0].GetType(), "unset provider type falls back to the AWS/GP3 default")

	require.NotNil(t, savedSharedPool, "the repository save must receive the generated enable metadata, not the pre-enable pool")
	savedCoprocessor := sharedPoolDefaultComponentMap(savedSharedPool)[apiv1.ComponentType_COPROCESSOR_WORKER]
	require.NotNil(t, savedCoprocessor)
	assert.True(t, proto.Equal(coprComp, savedCoprocessor), "the persisted metadata must preserve the generated CREATING quota and storage snapshot")
	for _, operation := range []string{"SetHPASpec", "SetScaleOutConfig", "SetHPAScaleBehavior"} {
		require.NoError(t, ensureCoprocessorWorkerComponentExists(savedSharedPool), "%s must not reject the post-overlay persisted component as missing", operation)
	}

	require.NotNil(t, tikvComp)
	wantTiKV := proto.Clone(origTiKV).(*apiv1.Component)
	wantTiKV.Phase = apiv1.Component_MODIFYING
	assert.True(t, proto.Equal(wantTiKV, tikvComp), "tikv component must have only its Phase changed to MODIFYING, everything else byte-for-byte unchanged")

	require.NotNil(t, tikvWorkerComp)
	assert.True(t, proto.Equal(origTiKVWorker, tikvWorkerComp), "tikv_worker component must remain byte-for-byte unchanged")

	require.NotNil(t, pdComp)
	assert.True(t, proto.Equal(origPD, pdComp), "pd component must remain byte-for-byte unchanged")
}

// TestSetSharedPoolTiKVRemoteWorkers_EnablePreservesConfigCenterMetaProfileId
// guards against buildTiKVRemoteWorkersTargetSharedPool dropping the
// non-protobuf ConfigCenterMetaProfileId field, which proto.Merge cannot
// copy since it lives on the entity.SharedPool Go wrapper, not the embedded
// apiv1.SharedPool message - matching the same field explicitly threaded
// through by buildPDDeploymentModeTargetSharedPool.
func TestSetSharedPoolTiKVRemoteWorkers_EnablePreservesConfigCenterMetaProfileId(t *testing.T) {
	const profileID = "cc-meta-profile-123"
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	sharedPool.ConfigCenterMetaProfileId = profileID

	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	ruConfigRepo, tidbClusterRepo, dataplaneClient := newTKRWEnableDeps(t)
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:  sharedPoolRepo,
		lockRepo:        lockRepo,
		dagExecutor:     dagExecutor,
		scaleManager:    scaleManager,
		ruConfigRepo:    ruConfigRepo,
		tidbClusterRepo: tidbClusterRepo,
		dataplaneClient: dataplaneClient,
		transactioner:   &mockTKRWTransactioner{},
	}
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledBoth(t))
	require.NoError(t, err)
	assert.Equal(t, profileID, result.ConfigCenterMetaProfileId, "the frozen DAG snapshot must preserve ConfigCenterMetaProfileId unchanged")
}

// TestSetSharedPoolTiKVRemoteWorkers_EnableReusesRetainedCoprocessorComponentAsCreating
// guards against a regression where re-enabling coprocessor routing after a
// disable that retained COPROCESSOR_WORKER component metadata could silently
// skip marking it CREATING in the frozen DAG snapshot: if the recomputed
// quota happens to equal the retained quota, genDefaultTiKVWorkerComponent's
// Phase-transition logic (gated on a quota diff) would otherwise leave Phase
// untouched at whatever the retained component holds, breaking the enable
// snapshot contract that ApplyComponentGroupAction relies on to recreate the
// already-deleted coprocessor CR.
func TestSetSharedPoolTiKVRemoteWorkers_EnableReusesRetainedCoprocessorComponentAsCreating(t *testing.T) {
	sharedPool := newTKRWSharedPool(tkrwEnabledDisableOnly(), 0)
	retainedQuota := &apiv1.NodeQuota{
		NodeCount:    1,
		MinNodeCount: 1,
		MaxNodeCount: 60,
		NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
	}
	retainedStorages := []*apiv1.Component_Storage{
		{Name: apiv1.Component_Storage_DATA, SizeGi: 200, Type: entity.GP3},
	}
	sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, &apiv1.Component{
		Id:            testTKRWComponentID,
		ComponentType: apiv1.ComponentType_COPROCESSOR_WORKER,
		// AVAILABLE phase with quota/storages already matching what the
		// generator would independently compute is exactly the scenario that
		// used to silently skip the CREATING transition.
		Phase:     apiv1.Component_AVAILABLE,
		NodeQuota: proto.Clone(retainedQuota).(*apiv1.NodeQuota),
		Storages:  retainedStorages,
	})

	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	var savedSharedPool *entity.SharedPool
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) { savedSharedPool = args.Get(1).(*entity.SharedPool) }).
		Return(nil).Once()
	ruConfigRepo, tidbClusterRepo, dataplaneClient := newTKRWEnableDeps(t)

	var submittedPlan string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) { submittedPlan = args.String(1) }).
		Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:  sharedPoolRepo,
		lockRepo:        lockRepo,
		dagExecutor:     dagExecutor,
		scaleManager:    scaleManager,
		ruConfigRepo:    ruConfigRepo,
		tidbClusterRepo: tidbClusterRepo,
		dataplaneClient: dataplaneClient,
		transactioner:   &mockTKRWTransactioner{},
	}
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, tkrwEnabledBoth(t))
	require.NoError(t, err)

	assert.Equal(t, dag.PlanEnableCoprocessorWorker, submittedPlan)
	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())

	var coprComps []*apiv1.Component
	for _, comp := range result.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
			coprComps = append(coprComps, comp)
		}
	}
	require.Len(t, coprComps, 1, "the retained component must be reused in place, not duplicated")
	coprComp := coprComps[0]

	assert.Equal(t, apiv1.Component_CREATING, coprComp.GetPhase(),
		"a re-enable must always snapshot the retained COPROCESSOR_WORKER component as CREATING, regardless of whether its recomputed quota changed")
	assert.True(t, proto.Equal(retainedQuota, coprComp.GetNodeQuota()), "the fix must not corrupt the generator's quota computation")
	require.Len(t, coprComp.GetStorages(), 1)
	assert.Equal(t, apiv1.Component_Storage_DATA, coprComp.GetStorages()[0].GetName())
	assert.Equal(t, uint32(200), coprComp.GetStorages()[0].GetSizeGi())
	assert.Equal(t, entity.GP3, coprComp.GetStorages()[0].GetType())

	require.NotNil(t, savedSharedPool)
	var savedCoprocessorComps []*apiv1.Component
	for _, comp := range savedSharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
			savedCoprocessorComps = append(savedCoprocessorComps, comp)
		}
	}
	require.Len(t, savedCoprocessorComps, 1, "persisting a retained re-enable must not duplicate COPROCESSOR_WORKER metadata")
	assert.True(t, proto.Equal(coprComp, savedCoprocessorComps[0]))
}

// TestSetSharedPoolTiKVRemoteWorkers_LegacyEmptyPersistedDisable verifies
// Tech-Leader's 2nd mandated assertion: a legacy/unset persisted
// EnabledTikvRemoteWorkers field normalizes to "both enabled" for the
// stability comparison, so an explicit disable request against it must not
// be short-circuited as a stable no-op - Save must persist an EXPLICIT
// [TIKV_WORKER] value (not leave the field unset), and the disable plan must
// actually be submitted.
func TestSetSharedPoolTiKVRemoteWorkers_LegacyEmptyPersistedDisable(t *testing.T) {
	sharedPool := newTKRWSharedPool(nil, 3)
	workerGroup := tkrwReadyWorkerGroup(testTKRWWorkerName)
	dm := newTKRWDataplaneManager(t, workerGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)

	var saveCalls int
	var savedEnabled []apiv1.ComponentType
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			saveCalls++
			savedEnabled = args.Get(1).(*entity.SharedPool).GetEnabledTikvRemoteWorkers()
		}).
		Return(nil).Once()

	var submittedPlan string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) { submittedPlan = args.String(1) }).
		Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	target := tkrwEnabledDisableOnly()
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, target)
	require.NoError(t, err, "a legacy-empty persisted state must not be treated as a stable no-op against an explicit disable request")

	assert.Equal(t, 1, saveCalls, "Save must actually be called, not skipped as a no-op")
	assert.Equal(t, target, savedEnabled, "Save must persist an explicit [TIKV_WORKER] value rather than leaving the legacy-empty field untouched")
	assert.Equal(t, dag.PlanDisableCoprocessorWorker, submittedPlan)
	assert.Equal(t, target, result.GetEnabledTikvRemoteWorkers())
	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
}

// TestSetSharedPoolTiKVRemoteWorkers_DisableRepairSubmission verifies Tech-
// Leader's 3rd mandated assertion for the disable direction at the full
// handler level (not merely via the isolated isTiKVRemoteWorkersStableInTarget
// helper): persisted state already equals the disable target and the TiKV
// group is already routed correctly, but the dedicated coprocessor CR has
// not been cleaned up yet, so isTiKVRemoteWorkersStableInTarget reports not
// stable - the handler must still submit a real disable DAG.
func TestSetSharedPoolTiKVRemoteWorkers_DisableRepairSubmission(t *testing.T) {
	disable := tkrwEnabledDisableOnly()
	sharedPool := newTKRWSharedPool(disable, 0)
	tikvGroup := tkrwReadyTiKVGroup(testTKRWWorkerName)
	workerGroup := tkrwReadyWorkerGroup(testTKRWWorkerName)
	coprGroup := tkrwReadyWorkerGroup(testTKRWCoprName) // dedicated coprocessor CR still present - cleanup outstanding
	dm := newTKRWDataplaneManager(t, tikvGroup, workerGroup, coprGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()

	var submitted bool
	var submittedPlan string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			submitted = true
			submittedPlan = args.String(1)
		}).
		Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		transactioner:    &mockTKRWTransactioner{},
	}
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, disable)
	require.NoError(t, err)

	assert.True(t, submitted, "persisted-target-matches-but-live-not-converged must still trigger a real DAG submission, not a no-op")
	assert.Equal(t, dag.PlanDisableCoprocessorWorker, submittedPlan)
	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
}

// TestSetSharedPoolTiKVRemoteWorkers_EnableRepairSubmission verifies Tech-
// Leader's 3rd mandated assertion for the enable direction at the full
// handler level: persisted state already equals the enable target and the
// TiKV group is already routed correctly, but the dedicated coprocessor CR
// has not finished rolling out (ReadyReplicas short of Replicas), so
// isTiKVRemoteWorkersStableInTarget reports not stable - the handler must
// still submit a real enable DAG.
func TestSetSharedPoolTiKVRemoteWorkers_EnableRepairSubmission(t *testing.T) {
	enabled := tkrwEnabledBoth(t)
	sharedPool := newTKRWSharedPool(enabled, 3)
	tikvGroup := tkrwReadyTiKVGroup(testTKRWCoprName)
	coprGroup := tkrwReadyWorkerGroup(testTKRWCoprName)
	coprGroup.Status.ReadyReplicas = 2 // rollout not finished
	dm := newTKRWDataplaneManager(t, tikvGroup, coprGroup)
	sharedPoolRepo, lockRepo, dagExecutor, scaleManager := newTKRWHandlerMocks(t, sharedPool)
	sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	ruConfigRepo, tidbClusterRepo, dataplaneClient := newTKRWEnableDeps(t)

	var submitted bool
	var submittedPlan string
	dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			submitted = true
			submittedPlan = args.String(1)
		}).
		Return(nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: dm,
		ruConfigRepo:     ruConfigRepo,
		tidbClusterRepo:  tidbClusterRepo,
		dataplaneClient:  dataplaneClient,
		transactioner:    &mockTKRWTransactioner{},
	}
	result, err := s.SetSharedPoolTiKVRemoteWorkers(context.Background(), testTKRWSharedPoolID, enabled)
	require.NoError(t, err)

	assert.True(t, submitted, "persisted-target-matches-but-live-not-converged must still trigger a real DAG submission, not a no-op")
	assert.Equal(t, dag.PlanEnableCoprocessorWorker, submittedPlan)
	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
}