package cmd

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"os"
	"os/exec"
	pathpkg "path"
	"path/filepath"
	"reflect"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/innerr/ticat/pkg/core/model"
	"k8s.io/apimachinery/pkg/api/resource"

	"github.com/tidbcloud/controlplane9/internal/artifactsync"
	cpaws "github.com/tidbcloud/controlplane9/internal/aws"
	"github.com/tidbcloud/controlplane9/internal/cluster"
	"github.com/tidbcloud/controlplane9/internal/config"
	cp9errors "github.com/tidbcloud/controlplane9/internal/errors"
	"github.com/tidbcloud/controlplane9/internal/host"
	"github.com/tidbcloud/controlplane9/internal/lifecycle"
	"github.com/tidbcloud/controlplane9/internal/proof"
	"github.com/tidbcloud/controlplane9/internal/sourcebacked"
	cpterraform "github.com/tidbcloud/controlplane9/internal/terraform"
	cptilt "github.com/tidbcloud/controlplane9/internal/tilt"
	"github.com/tidbcloud/controlplane9/internal/verbose"
)

const (
	starterDefaultRemoteEKSKubernetesVersion = "1.36"
	starterPDMSNodeGroupLabel                = "cp9.tidbcloud.com/starter-pdms"
	starterPDMSModeEnabled                   = "enabled"
	starterPDMSModeDisabled                  = "disabled"
	starterRemoteRegionalNamespace           = "tidb-admin"
	starterContextIndexFileName              = "context.json"
	starterGatewayDNSAnnotation              = "gateway.tidbcloud.com/dnsName"
	starterImagePullDefaultAccountID         = "385595570414"
	starterImagePullDefaultRegion            = "us-west-2"
	starterImagePullDefaultProfile           = "sso"
	starterFixedDataplaneStagingRegistry     = starterImagePullDefaultAccountID + ".dkr.ecr." + starterImagePullDefaultRegion + ".amazonaws.com"
	starterFixedRegionalStagingRegistry      = starterImagePullDefaultAccountID + ".dkr.ecr.us-east-1.amazonaws.com"
	starterSelectedImageSnapshotFileName     = "starter-selected-image-snapshot.json"
	starterStatusEvidenceFileName            = "starter-status.json"
	starterActivationWaitEvidenceFileName    = "starter-activation-waits.json"
	starterPreApplyPlanFileName              = "starter-pre-apply-plan.json"
	starterParityContractFileName            = "starter-parity-contract.json"
	starterDBAASProxyHostFileName            = "starter-dbaas-proxy-host.txt"
	starterGatewayHostFileName               = "starter-tidb-gateway-host.txt"
	starterRegionalServiceURLFileName        = "starter-serverless-regional-service-url.txt"
	starterOpsServiceHostFileName            = "starter-ops-service-host.txt"
	starterAIGatewayHostFileName             = "starter-ai-gateway-host.txt"
	starterAIGatewayDNSReceiptFileName       = "starter-ai-gateway-route53.json"
	starterRemoteWorkloadPlaceholderImage    = "public.ecr.aws/docker/library/nginx:1.27-alpine"
	starterGatewayImageRepositoryPlaceholder = "__CP9_STARTER_TIDB_GATEWAY_IMAGE_REPOSITORY__"
	starterGatewayImagePullSecretPlaceholder = "__CP9_STARTER_TIDB_GATEWAY_IMAGE_PULL_SECRET__"
	starterGatewayVersionPlaceholder         = "__CP9_STARTER_TIDB_GATEWAY_VERSION__"
	starterO11yAgentControlServerArtifact    = "starter-o11y-agent-control-server"
	starterPrivateLinkSubnetTagKey           = "type"
	starterPrivateLinkSubnetTagValue         = "private"
	starterPrivateLinkEndpointSGPrefix       = "plc-endpoint-sg-"
	starterCatalogServicePlan                = "Starter"
	starterSelectedServingLane               = "tidb-pool"
	starterSeedOrgID                         = uint64(1440002)
	// Account requires a non-zero cluster ID but does not use it when ensuring the shared dev-tier key.
	starterAccountJWKSEnsureClusterID = uint64(1)

	// The session PCA is valid for 7 days; request less so issuance after CA creation stays within the parent validity.
	starterDbaasProxyPCACertificateDays = 6

	starterRemoteWorkloadImageSourceEnvOverride = "env-override"
	starterImageSourceOwnerSnapshot             = "owner-artifact-snapshot"
	starterRemoteWorkloadImageSourceSummary     = "source-summary"
	starterRemoteWorkloadImageSourceStaging     = "staging"
	starterImageProvenanceReviewedOverride      = "reviewed-override"
	starterImageProvenanceSourceBuild           = "source-build"
	starterRemoteSelectedImageDeploymentKind    = "kubernetes-deployment"
	starterRemoteSelectedImageStatefulSetKind   = "kubernetes-statefulset"
	starterRemoteSelectedImageDaemonSetKind     = "kubernetes-daemonset"
	starterSelectedImageSnapshotAPIVersion      = "cp9.starter.selected-images/v1alpha1"
	starterSelectedImageSnapshotKind            = "StarterSelectedImageSnapshot"
	starterStatusEvidenceAPIVersion             = "cp9.starter.status/v1alpha1"
	starterStatusEvidenceKind                   = "StarterStatus"
	starterActivationWaitAPIVersion             = "cp9.starter.activation-waits/v1alpha1"
	starterActivationWaitKind                   = "StarterActivationWaitEvidence"
	starterParityContractAPIVersion             = "cp9.starter.parity/v1alpha1"
	starterParityContractKind                   = "StarterParityContract"
	starterContractStubVersion                  = "cp9.starter.contract-stub/v1"
	starterAWSSharedCDRevision                  = "57fa96f337548ce1b5c61834a33b38ef1c0ed4c0"
	starterAWSSharedProviderRevision            = "e49dba3de19c3461a6dc01eba4cdce3475c8de37"
	starterAppDeliveryRevision                  = "3a2cc887002b0a4f8086f54dc36416dd220ba8d8"
)

type starterStatusResult struct {
	State            string               `json:"state"`
	Session          string               `json:"session"`
	Persona          string               `json:"persona"`
	StateDir         string               `json:"state_dir"`
	RemoteEKSCluster string               `json:"remote_eks_cluster,omitempty"`
	PDMode           string               `json:"pd_mode"`
	Message          string               `json:"message"`
	Checks           []starterStatusCheck `json:"checks"`
	Activation       *activationRef       `json:"activation,omitempty"`
}

type starterStatusCheck = lifecycle.StarterReadinessCheck

type activationRef struct {
	Command string `json:"command"`
}

type starterStatusEvidence struct {
	APIVersion     string               `json:"apiVersion"`
	Kind           string               `json:"kind"`
	ArtifactSHA256 map[string]string    `json:"artifact_sha256,omitempty"`
	Result         *starterStatusResult `json:"result"`
}

type starterParityContractEvidence struct {
	APIVersion           string            `json:"apiVersion"`
	Kind                 string            `json:"kind"`
	AuthoritativeSource  map[string]string `json:"authoritative_source"`
	InventoryPath        string            `json:"inventory_path"`
	InventorySHA256      string            `json:"inventory_sha256"`
	SourceEvidencePath   string            `json:"source_evidence_path"`
	SourceEvidenceSHA256 string            `json:"source_evidence_sha256"`
	ContractStubVersion  string            `json:"contract_stub_version"`
	LiveStagingRole      string            `json:"live_staging_role"`
}

func (r *starterStatusResult) String() string {
	if r == nil {
		return "starter status: <nil>"
	}
	var b strings.Builder
	fmt.Fprintf(&b, "starter %s: %s\n", r.State, r.Message)
	if r.RemoteEKSCluster != "" {
		fmt.Fprintf(&b, "remote eks: %s\n", r.RemoteEKSCluster)
	}
	fmt.Fprintf(&b, "pd mode: %s\n", r.PDMode)
	for i := range r.Checks {
		check := &r.Checks[i]
		if check.Message != "" {
			fmt.Fprintf(&b, "- %s: %s (%s)\n", check.Name, check.Status, check.Message)
		} else {
			fmt.Fprintf(&b, "- %s: %s\n", check.Name, check.Status)
		}
	}
	return strings.TrimRight(b.String(), "\n")
}

type starterCommandRunner func(ctx context.Context, env []string, name string, args ...string) ([]byte, error)

type starterLiveChecker func(ctx context.Context, input starterLiveCheckInput) []starterStatusCheck

type starterPrerequisiteWaiter func(context.Context, *host.Session) ([]starterStatusCheck, error)

type starterPrerequisiteChecker func(context.Context, *host.Session) []starterStatusCheck

type starterSubstrateActivateFunc func(ctx context.Context, stateDir string, session *config.SessionConfig) error

type starterWorkloadActivateFunc func(ctx context.Context, stateDir string, baseline *host.Session, session *config.SessionConfig) error

type starterLiveCheckInput struct {
	Session    *config.SessionConfig
	Baseline   *host.Session
	StateDir   string
	RemoteEKS  *cpterraform.RemoteEKSOutput
	Kubeconfig string
}

var (
	starterStatusCommandRunner           starterCommandRunner          = runStarterStatusCommand
	starterImageRegistryChecker          starterImageRegistryCheckFunc = checkStarterImageRegistry
	starterStatusLiveChecker             starterLiveChecker            = runStarterLiveChecks
	starterSubstrateActivator            starterSubstrateActivateFunc  = runStarterSubstrateActivation
	starterWorkloadActivator             starterWorkloadActivateFunc   = runStarterWorkloadActivation
	starterActivationPrerequisiteWaiter  starterPrerequisiteWaiter     = waitStarterActivationPrerequisites
	starterActivationPrerequisiteChecker starterPrerequisiteChecker    = checkStarterActivationPrerequisites
	starterActivationPlanReporter                                      = reportStarterActivationPreApplyPlan
	starterKubeAPITunnelEnsurer                                        = cptilt.EnsureKubeAPITunnel
	starterActivationWaitDeadline                                      = 60 * time.Minute
	starterActivationWaitPollInterval                                  = 5 * time.Second
)

func starterActivateCmd() model.NormalCmd {
	return func(args model.ArgVals, cc *model.Cli, env *model.Env, _ []model.ParsedCmd) error {
		syncVerbose(env)
		result, err := runStarterActivate(context.Background(), args.GetRaw("config"))
		if result != nil {
			if outputErr := model.Output(cc, env, result); outputErr != nil {
				return outputErr
			}
		}
		if err != nil {
			return err
		}
		return nil
	}
}

func starterStatusCmd() model.NormalCmd {
	return func(args model.ArgVals, cc *model.Cli, env *model.Env, _ []model.ParsedCmd) error {
		syncVerbose(env)
		result, err := runStarterStatus(context.Background(), args.GetRaw("config"))
		if err != nil {
			return err
		}
		return model.Output(cc, env, result)
	}
}

func runStarterActivate(ctx context.Context, cfgPath string) (*starterStatusResult, error) {
	return runStarterActivateWithPlan(ctx, cfgPath, true)
}

func runStarterActivateAfterBaseline(ctx context.Context, cfgPath string) (*starterStatusResult, error) {
	return runStarterActivateWithPlan(ctx, cfgPath, false)
}

func runStarterActivateWithPlan(ctx context.Context, cfgPath string, emitPreApplyPlan bool) (*starterStatusResult, error) {
	resolved, selected, err := resolveDevFullSession(cfgPath, "cp9 starter")
	if err != nil {
		return nil, err
	}
	if err := validateStarterAIFTSInputs(resolved); err != nil {
		return nil, err
	}
	if err := requireStarterBaselineHost(selected); err != nil {
		return nil, err
	}
	repoRoot, err := sessionRepoRoot(resolved)
	if err != nil {
		return nil, err
	}
	if err := writeStarterParityContractEvidence(selected.StateDir, repoRoot); err != nil {
		return nil, err
	}
	if _, err := ensureStarterBaselineKubeAPITunnel(ctx, selected); err != nil {
		return nil, fmt.Errorf("repair Dev full baseline kube API tunnel before Starter activation: %w", err)
	}
	activation := starterActivationSession(resolved)
	if err := validateStarterMinimalNetworkCapacity(activation.Target.AWS.RemoteEKS.NodeGroups); err != nil {
		return nil, err
	}
	if emitPreApplyPlan {
		preApplyPlan := starterActivationPreApplyPlan(activation)
		if err := emitStarterActivationPreApplyPlan(selected.StateDir, preApplyPlan); err != nil {
			return nil, err
		}
	}
	artifactReady, err := starterRemoteEKSArtifactMatches(selected.StateDir, activation)
	if err != nil {
		return nil, err
	}
	if !artifactReady {
		if err := starterSubstrateActivator(ctx, selected.StateDir, activation); err != nil {
			return nil, fmt.Errorf("activate starter substrate: %w", err)
		}
	}
	checks, waitErr := starterActivationPrerequisiteWaiter(ctx, selected)
	checks = withStarterStatusConfig(checks, cfgPath)
	if waitErr != nil {
		result := starterPrerequisiteStatusResult(activation, selected, checks, cfgPath)
		return result, errors.New(withStarterStatusCommand(waitErr.Error(), cfgPath))
	}
	if err := starterWorkloadActivator(ctx, selected.StateDir, selected, activation); err != nil {
		var readinessErr *starterReadinessError
		if errors.As(err, &readinessErr) {
			result := starterPrerequisiteStatusResult(activation, selected, readinessErr.Checks, cfgPath)
			if writeErr := writeStarterStatusEvidence(result); writeErr != nil {
				return result, errors.Join(err, writeErr)
			}
			return result, err
		}
		return nil, fmt.Errorf("activate starter workloads: %w", err)
	}
	result, err := runStarterStatusResolved(ctx, activation, selected, cfgPath)
	if err != nil {
		return nil, err
	}
	if result.State != "ready" {
		return result, cp9errors.Newf(cp9errors.CategoryResource, "starter activation is not ready: %s", result.Message)
	}
	return result, nil
}

func starterActivationPreApplyPlan(activation *config.SessionConfig) *upDryRunStarterActivation {
	if activation == nil {
		return nil
	}
	reportSession := *activation
	reportSession.Starter.Enabled = true
	plan := upDryRunStarterActivationPlan(&reportSession)
	if plan != nil {
		plan.Trigger = "cp9 starter activate"
	}
	return plan
}

func writeStarterActivationPreApplyPlan(stateDir string, plan *upDryRunStarterActivation) error {
	if plan == nil {
		return cp9errors.New(cp9errors.CategoryConfig, "Starter pre-apply plan is required")
	}
	path := filepath.Join(stateDir, "reports", "evidence", starterPreApplyPlanFileName)
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return fmt.Errorf("create Starter pre-apply plan directory: %w", err)
	}
	data, err := json.MarshalIndent(plan, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal Starter pre-apply plan: %w", err)
	}
	if err := os.WriteFile(path, data, 0o600); err != nil {
		return fmt.Errorf("write Starter pre-apply plan: %w", err)
	}
	return nil
}

func writeStarterParityContractEvidence(stateDir, repoRoot string) error {
	inventoryPath := filepath.Join(
		repoRoot,
		"personas", "app-delivery", "envs", "dev-full", "starter-regional", "PARITY.md",
	)
	digest, err := sha256File(inventoryPath)
	if err != nil {
		return fmt.Errorf("hash Starter parity inventory: %w", err)
	}
	sourceEvidencePath := filepath.Join(
		repoRoot,
		"personas", "app-delivery", "envs", "dev-full", "starter-regional", "SOURCE_EVIDENCE.yaml",
	)
	sourceEvidenceDigest, err := sha256File(sourceEvidencePath)
	if err != nil {
		return fmt.Errorf("hash Starter source evidence: %w", err)
	}
	evidence := starterParityContractEvidence{
		APIVersion: starterParityContractAPIVersion,
		Kind:       starterParityContractKind,
		AuthoritativeSource: map[string]string{
			"aws-shared-cd":       starterAWSSharedCDRevision,
			"aws-shared-provider": starterAWSSharedProviderRevision,
			"app-delivery":        starterAppDeliveryRevision,
		},
		InventoryPath:   filepath.ToSlash(filepath.Join("personas", "app-delivery", "envs", "dev-full", "starter-regional", "PARITY.md")),
		InventorySHA256: digest,
		SourceEvidencePath: filepath.ToSlash(
			filepath.Join("personas", "app-delivery", "envs", "dev-full", "starter-regional", "SOURCE_EVIDENCE.yaml"),
		),
		SourceEvidenceSHA256: sourceEvidenceDigest,
		ContractStubVersion:  starterContractStubVersion,
		LiveStagingRole:      "drift-observation-only-unverified",
	}
	data, err := json.MarshalIndent(evidence, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal Starter parity contract evidence: %w", err)
	}
	path := filepath.Join(stateDir, "reports", "evidence", starterParityContractFileName)
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return fmt.Errorf("create Starter parity contract evidence directory: %w", err)
	}
	if err := os.WriteFile(path, append(data, '\n'), 0o600); err != nil {
		return fmt.Errorf("write Starter parity contract evidence: %w", err)
	}
	return nil
}

func emitStarterActivationPreApplyPlan(stateDir string, plan *upDryRunStarterActivation) error {
	if err := writeStarterActivationPreApplyPlan(stateDir, plan); err != nil {
		return err
	}
	starterActivationPlanReporter(plan)
	return nil
}

func reportStarterActivationPreApplyPlan(plan *upDryRunStarterActivation) {
	summary := formatStarterActivationPreApplyPlan(plan)
	if summary == "" {
		return
	}
	fmt.Fprintln(os.Stderr, summary)
}

func formatStarterActivationPreApplyPlan(plan *upDryRunStarterActivation) string {
	if plan == nil {
		return ""
	}
	roles := make([]string, 0, len(plan.Topology))
	for _, group := range plan.Topology {
		roles = append(roles, fmt.Sprintf("%s=%s×%d", group.Role, group.InstanceType, group.Nodes))
	}
	return fmt.Sprintf(
		"starter activation pre-apply: remote_eks=%s topology=%s selected_lane=%s tidb_pool_pods=%d expected_cost_window=%s source=%s as_of=%s",
		plan.RemoteEKSCluster,
		strings.Join(roles, ","),
		plan.SelectedServingLane,
		plan.TiDBPoolPods,
		plan.Cost.ExpectedCostWindow,
		plan.Cost.Source,
		plan.Cost.AsOf,
	)
}

func runStarterStatus(ctx context.Context, cfgPath string) (*starterStatusResult, error) {
	resolved, selected, err := resolveDevFullSession(cfgPath, "cp9 starter")
	if err != nil {
		return nil, err
	}
	return runStarterStatusResolved(ctx, starterActivationSession(resolved), selected, cfgPath)
}

func resolveDevFullSession(cfgPath, commandGroup string) (*config.SessionConfig, *host.Session, error) {
	session, err := config.LoadSession(cfgPath)
	if err != nil {
		return nil, nil, err
	}
	resolved, err := config.ResolveSession(session, cfgPath)
	if err != nil {
		return nil, nil, err
	}
	persona := config.SessionPersona(resolved)
	if persona != "dev-full" {
		return nil, nil, cp9errors.Newf(cp9errors.CategoryConfig,
			"%s commands require a dev-full session, got %q", commandGroup, persona)
	}
	selected := host.ResolveSessionConfig(resolved)
	if loaded, lerr := host.Load(selected.StateDir); lerr == nil {
		selected = loaded
	} else if !errors.Is(lerr, os.ErrNotExist) {
		return nil, nil, fmt.Errorf("load selected host: %w", lerr)
	} else {
		selected.Mode = ""
	}
	return resolved, selected, nil
}

func starterActivationSession(resolved *config.SessionConfig) *config.SessionConfig {
	activation := *resolved
	remote := activation.Target.AWS.RemoteEKS
	remote.Enabled = true
	remote.ClusterName = upDryRunRegionalSmokeClusterName(activation.Name, remote.ClusterName)
	if strings.TrimSpace(remote.KubernetesVersion) == "" {
		remote.KubernetesVersion = starterDefaultRemoteEKSKubernetesVersion
	}
	remote.NodeInstanceType = ""
	remote.NodeCount = 0
	topology := config.ResolvedStarterTopology(&activation)
	remote.NodeGroups = starterMinimalNodeGroupsForPDMode(&topology, config.ResolvedStarterPDMSEnabled(&activation))
	activation.Target.AWS.RemoteEKS = remote
	return &activation
}

func starterMinimalNodeGroups(topology *config.SessionStarterTopologyConfig) []config.RemoteEKSNodeGroupConfig {
	return starterMinimalNodeGroupsForPDMode(topology, true)
}

func starterMinimalNodeGroupsForPDMode(
	topology *config.SessionStarterTopologyConfig,
	pdmsEnabled bool,
) []config.RemoteEKSNodeGroupConfig {
	tidbNodes := 0
	if topology.TiDBServing.Nodes != nil {
		tidbNodes = *topology.TiDBServing.Nodes
	}
	groups := []config.RemoteEKSNodeGroupConfig{
		starterMinimalNodeGroup("starter-pd", "pd", topology.PD.InstanceType, 3, true),
		starterMinimalNodeGroup("starter-tikv", "tikv", topology.TiKV.InstanceType, 5, true),
		starterMinimalNodeGroup("starter-tikv-hot", "tikv-hot", "r8g.2xlarge", 0, true),
		starterMinimalNodeGroup("starter-keyspace-etcd", "keyspace-etcd", topology.KeyspaceETCD.InstanceType, 3, true),
		starterMinimalNodeGroup("starter-gateway", "gateway", topology.Gateway.InstanceType, 4, true),
		starterMinimalNodeGroup("starter-regional-control-plane", "regional-control-plane", topology.RegionalControlPlane.InstanceType, 3, false),
		starterMinimalNodeGroup("starter-tidb-serving", "tidb-serving", topology.TiDBServing.InstanceType, tidbNodes, true),
		starterMinimalNodeGroup("starter-import-pool", "import-pool", "r6g.2xlarge", 0, false),
		starterMinimalNodeGroup("starter-dedicated-x16xl", "dedicated-x16xl", "m6g.16xlarge", 0, false),
		starterMinimalNodeGroup("starter-general", "general", "t4g.2xlarge", 0, false),
	}
	for i := range groups {
		switch groups[i].Name {
		case "starter-pd":
			groups[i].Labels[starterPDMSNodeGroupLabel] = starterPDMSModeEnabled
			if !pdmsEnabled {
				groups[i].Labels[starterPDMSNodeGroupLabel] = starterPDMSModeDisabled
			}
		case "starter-tikv":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "tikv"
			groups[i].Labels["serverless.tidbcloud.com/tier"] = "standard"
			groups[i].Labels["serverless.tidbcloud.com/zone"] = "starter"
		case "starter-tikv-hot":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "tikv"
			groups[i].Labels["serverless.tidbcloud.com/tier"] = "hot"
			groups[i].Labels["serverless.tidbcloud.com/zone"] = "starter"
		case "starter-tidb-serving":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "tidb"
			groups[i].Labels["serverless.tidbcloud.com/nodegroup"] = "starter"
			groups[i].Labels["serverless.tidbcloud.com/tier"] = "standard"
			groups[i].Labels["serverless.tidbcloud.com/zone"] = "starter"
		case "starter-import-pool":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "import-pool"
			groups[i].Taints = []config.RemoteEKSNodeTaint{{
				Key: "serverless.tidbcloud.com/node", Value: "import-pool", Effect: "NO_SCHEDULE",
			}}
		case "starter-dedicated-x16xl":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "dedicated"
			groups[i].Labels["dedicated-node-cpu-num"] = "64"
			groups[i].Taints = []config.RemoteEKSNodeTaint{{
				Key: "serverless.tidbcloud.com/node", Value: "dedicated", Effect: "NO_SCHEDULE",
			}}
		case "starter-general":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "general"
		case "starter-gateway":
			groups[i].Labels["serverless.tidbcloud.com/node"] = "critical-services"
			groups[i].Labels["serverless.tidbcloud.com/zone"] = "primary"
			groups[i].Labels["serverless.tidbcloud.com/kubelet-version"] = "1.31.7-20250627"
			groups[i].Taints = []config.RemoteEKSNodeTaint{{
				Key: "serverless.tidbcloud.com/node", Value: "critical-services", Effect: "NO_SCHEDULE",
			}}
		}
	}
	return groups
}

func starterMinimalNodeGroup(name, role, instanceType string, size int, tainted bool) config.RemoteEKSNodeGroupConfig {
	architecture, amiType, _ := config.StarterInstancePlatform(instanceType)
	group := config.RemoteEKSNodeGroupConfig{
		Name:         name,
		InstanceType: instanceType,
		Architecture: architecture,
		AMIType:      amiType,
		DesiredSize:  size,
		MinSize:      size,
		MaxSize:      size + 1,
		Labels: map[string]string{
			"cp9.tidbcloud.com/starter-role": role,
		},
	}
	if tainted {
		group.Taints = []config.RemoteEKSNodeTaint{{
			Key: "cp9.tidbcloud.com/starter-role", Value: role, Effect: "NO_SCHEDULE",
		}}
	}
	return group
}

func starterRemoteEKSArtifactMatches(stateDir string, activation *config.SessionConfig) (bool, error) {
	remote, err := loadStarterRemoteEKSArtifact(stateDir)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return false, nil
		}
		return false, fmt.Errorf("check starter remote EKS artifact: %w", err)
	}
	want := ""
	if activation != nil {
		want = strings.TrimSpace(activation.Target.AWS.RemoteEKS.ClusterName)
	}
	got := strings.TrimSpace(remote.ClusterName)
	if want == "" {
		return false, cp9errors.New(cp9errors.CategoryConfig, "starter activation remote EKS cluster name is empty")
	}
	if got != want {
		return false, cp9errors.Newf(
			cp9errors.CategoryConfig,
			"starter remote EKS artifact cluster %q does not match activation cluster %q; remove the stale artifact or use the matching session",
			got,
			want,
		)
	}
	wantNodeGroups := cpterraform.RemoteEKSNodeGroupInputs(activation.Target.AWS.RemoteEKS.NodeGroups)
	gotNodeGroups := slices.Clone(remote.NodeGroups)
	slices.SortFunc(wantNodeGroups, func(a, b cpterraform.RemoteEKSNodeGroupInput) int {
		return strings.Compare(a.Name, b.Name)
	})
	slices.SortFunc(gotNodeGroups, func(a, b cpterraform.RemoteEKSNodeGroupInput) int {
		return strings.Compare(a.Name, b.Name)
	})
	wantPDMSMode := starterRemoteEKSNodeGroupPDMSMode(wantNodeGroups)
	gotPDMSMode := starterRemoteEKSNodeGroupPDMSMode(gotNodeGroups)
	if wantPDMSMode != gotPDMSMode {
		return false, cp9errors.Newf(
			cp9errors.CategoryConfig,
			"starter.activation.pdms.enabled is immutable for an existing Starter substrate (artifact mode %s, requested mode %s); run cp9 destroy before recreating the session with the other PD mode",
			gotPDMSMode,
			wantPDMSMode,
		)
	}
	if !reflect.DeepEqual(gotNodeGroups, wantNodeGroups) {
		return false, nil
	}
	discovery := remote.PrivateLinkDiscovery
	if discovery == nil ||
		strings.TrimSpace(discovery.SubnetTagKey) != starterPrivateLinkSubnetTagKey ||
		strings.TrimSpace(discovery.SubnetTagValue) != starterPrivateLinkSubnetTagValue ||
		strings.TrimSpace(discovery.EndpointSecurityGroupName) != starterPrivateLinkEndpointSGPrefix+want {
		return false, nil
	}
	return true, nil
}

func starterRemoteEKSNodeGroupPDMSMode(groups []cpterraform.RemoteEKSNodeGroupInput) string {
	for _, group := range groups {
		if group.Name != "starter-pd" {
			continue
		}
		if group.Labels[starterPDMSNodeGroupLabel] == starterPDMSModeDisabled {
			return starterPDMSModeDisabled
		}
		return starterPDMSModeEnabled
	}
	return starterPDMSModeEnabled
}

func runStarterStatusResolved(
	ctx context.Context,
	resolved *config.SessionConfig,
	selected *host.Session,
	cfgPath string,
) (result *starterStatusResult, err error) {
	result = &starterStatusResult{
		State:            "not_ready",
		Session:          resolved.Name,
		Persona:          config.SessionPersona(resolved),
		StateDir:         selected.StateDir,
		RemoteEKSCluster: resolved.Target.AWS.RemoteEKS.ClusterName,
		PDMode:           starterResolvedPDMode(resolved),
		Activation:       &activationRef{Command: "cp9 starter activate"},
	}
	defer func() {
		if err != nil {
			return
		}
		if writeErr := writeStarterStatusEvidence(result); writeErr != nil {
			err = writeErr
		}
	}()

	baselineCheck := starterBaselineHostCheck(selected)
	result.Checks = append(result.Checks, baselineCheck)
	if baselineCheck.Status != "passed" {
		result.Message = baselineCheck.Message
		return result, nil
	}
	tunnelMessage, tunnelErr := ensureStarterBaselineKubeAPITunnel(ctx, selected)
	if tunnelErr != nil {
		check := starterStatusCheck{
			Name:    "baseline-kube-api-tunnel",
			Status:  "not_ready",
			Message: "repair Dev full baseline kube API tunnel: " + tunnelErr.Error(),
		}
		result.Checks = append(result.Checks, check)
		result.Message = check.Message
		return result, nil
	}
	if strings.TrimSpace(tunnelMessage) != "" {
		result.Checks = append(result.Checks, starterStatusCheck{
			Name:    "baseline-kube-api-tunnel",
			Status:  "passed",
			Message: tunnelMessage,
		})
	}

	remote, err := loadStarterRemoteEKSArtifact(selected.StateDir)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			check := starterStatusCheck{
				Name:    "remote-eks-artifact",
				Status:  "not_ready",
				Message: "no Starter remote EKS artifact found; run cp9 starter activate after cp9 up",
			}
			result.Checks = append(result.Checks, check)
			result.Message = check.Message
			return result, nil
		}
		return nil, err
	}
	if strings.TrimSpace(result.RemoteEKSCluster) == "" {
		result.RemoteEKSCluster = remote.ClusterName
	}
	result.Checks = append(result.Checks, starterStatusCheck{
		Name:    "remote-eks-artifact",
		Status:  "passed",
		Message: "session-owned remote EKS artifact exists",
	})
	prerequisites := starterActivationPrerequisiteChecker(ctx, selected)
	prerequisites = withStarterStatusConfig(prerequisites, cfgPath)
	result.Checks = append(result.Checks, prerequisites...)
	for i := range prerequisites {
		check := &prerequisites[i]
		if check.Status != "passed" {
			result.Message = check.Message
			if result.Message == "" {
				result.Message = check.Name + " is not ready; run " + starterStatusCommand(cfgPath)
			}
			return result, nil
		}
	}

	kubeconfig := filepath.Join(selected.StateDir, "remote-eks", "kubeconfig.yaml")
	liveChecks := starterStatusLiveChecker(ctx, starterLiveCheckInput{
		Session:    resolved,
		Baseline:   selected,
		StateDir:   selected.StateDir,
		RemoteEKS:  remote,
		Kubeconfig: kubeconfig,
	})
	result.Checks = append(result.Checks, liveChecks...)
	for i := range result.Checks {
		check := &result.Checks[i]
		if check.Status != "passed" {
			result.State = "not_ready"
			result.Message = check.Message
			if result.Message == "" {
				result.Message = check.Name + " is not ready"
			}
			return result, nil
		}
	}
	result.State = "ready"
	result.Message = "Starter regional substrate and workloads are active and reachable for this Dev full session"
	result.Activation = nil
	return result, nil
}

func starterResolvedPDMode(session *config.SessionConfig) string {
	return string(config.ResolvedStarterPDTopology(session).Mode)
}

func ensureStarterBaselineKubeAPITunnel(ctx context.Context, selected *host.Session) (string, error) {
	var lifecycleState *cluster.State
	if loaded, err := cluster.LoadState(selected.StateDir); err == nil {
		lifecycleState = loaded
	} else if !errors.Is(err, os.ErrNotExist) {
		return "", fmt.Errorf("load lifecycle state for baseline kube API tunnel: %w", err)
	}
	tiltStatusPath := cptilt.PathsForState(selected.StateDir, lifecycleState).Status
	return starterKubeAPITunnelEnsurer(ctx, tiltStatusPath)
}

func waitStarterActivationPrerequisites(ctx context.Context, selected *host.Session) ([]starterStatusCheck, error) {
	return lifecycle.WaitStarterActivationPrerequisites(
		ctx,
		selected,
		starterActivationWaitDeadline,
		starterActivationWaitPollInterval,
		nil,
	)
}

func checkStarterActivationPrerequisites(ctx context.Context, selected *host.Session) []starterStatusCheck {
	return lifecycle.CheckStarterActivationPrerequisites(ctx, selected, nil)
}

func starterPrerequisiteStatusResult(
	session *config.SessionConfig,
	selected *host.Session,
	checks []starterStatusCheck,
	cfgPath string,
) *starterStatusResult {
	message := "Starter activation prerequisites are not ready; run " + starterStatusCommand(cfgPath)
	for i := range checks {
		check := &checks[i]
		if check.Status != "passed" && strings.TrimSpace(check.Message) != "" {
			message = check.Message
			break
		}
	}
	return &starterStatusResult{
		State:            "not_ready",
		Session:          session.Name,
		Persona:          config.SessionPersona(session),
		StateDir:         selected.StateDir,
		RemoteEKSCluster: session.Target.AWS.RemoteEKS.ClusterName,
		Message:          message,
		Checks:           checks,
		Activation:       &activationRef{Command: "cp9 starter activate"},
	}
}

func withStarterStatusConfig(checks []starterStatusCheck, cfgPath string) []starterStatusCheck {
	for i := range checks {
		checks[i].Message = withStarterStatusCommand(checks[i].Message, cfgPath)
	}
	return checks
}

func withStarterStatusCommand(message, cfgPath string) string {
	const marker = "cp9 starter status config="
	prefix, _, found := strings.Cut(message, marker)
	if !found {
		return message
	}
	return prefix + starterStatusCommand(cfgPath)
}

func starterStatusCommand(cfgPath string) string {
	cfgPath = strings.TrimSpace(cfgPath)
	if cfgPath == "" {
		cfgPath = config.DefaultConfigFile
	}
	return "cp9 starter status config=" + shellQuote(cfgPath)
}

func starterBaselineHostCheck(selected *host.Session) starterStatusCheck {
	if selected == nil {
		return starterStatusCheck{Name: "baseline-host", Status: "not_ready", Message: "Dev full baseline host is missing; run cp9 up first"}
	}
	switch selected.Mode {
	case "":
		return starterStatusCheck{
			Name:    "baseline-host",
			Status:  "not_ready",
			Message: "Dev full baseline host is missing; run cp9 up first",
		}
	case config.HostModeLocalK3d, config.HostModeRemoteEC2K3d:
		return starterStatusCheck{
			Name:    "baseline-host",
			Status:  "passed",
			Message: "Dev full k3d control-plane host remains selected",
		}
	case config.HostModeRemoteEKS:
		return starterStatusCheck{
			Name:    "baseline-host",
			Status:  "not_ready",
			Message: "Starter activation must not replace the Dev full k3d host with remote EKS",
		}
	default:
		return starterStatusCheck{
			Name:    "baseline-host",
			Status:  "not_ready",
			Message: fmt.Sprintf("unsupported Dev full baseline host mode %q", selected.Mode),
		}
	}
}

func requireStarterBaselineHost(selected *host.Session) error {
	check := starterBaselineHostCheck(selected)
	if check.Status == "passed" {
		return nil
	}
	return cp9errors.New(cp9errors.CategoryConfig, check.Message)
}

func loadStarterRemoteEKSArtifact(stateDir string) (*cpterraform.RemoteEKSOutput, error) {
	path := filepath.Join(stateDir, "remote-eks", "remote-eks.json")
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var remote cpterraform.RemoteEKSOutput
	if err := json.Unmarshal(data, &remote); err != nil {
		return nil, fmt.Errorf("parse remote EKS artifact: %w", err)
	}
	if !remote.Enabled {
		return nil, cp9errors.New(cp9errors.CategoryConfig, "remote EKS artifact is disabled")
	}
	return &remote, nil
}

func runStarterLiveChecks(ctx context.Context, input starterLiveCheckInput) []starterStatusCheck {
	remote := input.RemoteEKS
	if remote == nil {
		return []starterStatusCheck{{Name: "remote-eks-artifact", Status: "not_ready", Message: "remote EKS artifact missing"}}
	}
	session := input.Session
	profile := ""
	region := strings.TrimSpace(remote.Region)
	if session != nil {
		profile = strings.TrimSpace(session.Target.AWS.IAM.Profile)
		if region == "" {
			region = strings.TrimSpace(session.Target.Region)
		}
	}
	clusterName := strings.TrimSpace(remote.ClusterName)
	kubeContext := strings.TrimSpace(remote.KubeconfigContext)
	if kubeContext == "" {
		kubeContext = clusterName
	}
	kubeconfig := strings.TrimSpace(input.Kubeconfig)
	if kubeconfig == "" {
		kubeconfig = filepath.Join(input.StateDir, "remote-eks", "kubeconfig.yaml")
	}
	execEnv := starterAWSExecEnv(profile, region)
	access := starterKubeAccess{ExecEnv: execEnv, Kubeconfig: kubeconfig, KubeContext: kubeContext}
	if input.Baseline != nil && input.Baseline.IsRemote() {
		state, err := cluster.LoadState(input.StateDir)
		if err != nil {
			return []starterStatusCheck{{Name: "starter-remote-status-access", Status: "not_ready", Message: err.Error()}}
		}
		tiltStatus, err := cptilt.LoadStatusForState(input.StateDir, state)
		if err != nil {
			return []starterStatusCheck{{Name: "starter-remote-status-access", Status: "not_ready", Message: err.Error()}}
		}
		access.Remote = input.Baseline
		access.Kubeconfig = pathpkg.Join(tiltStatus.RemoteRoot, "starter", "kubeconfig.yaml")
	}

	var checks []starterStatusCheck
	eksArgs := []string{"eks", "describe-cluster", "--name", clusterName, "--query", "cluster.status", "--output", "text"}
	if profile != "" {
		eksArgs = append(eksArgs, "--profile", profile)
	}
	if region != "" {
		eksArgs = append(eksArgs, "--region", region)
	}
	eksOut, err := starterStatusCommandRunner(ctx, execEnv, "aws", eksArgs...)
	switch {
	case err != nil:
		checks = append(
			checks,
			starterStatusCheck{Name: "aws-eks-active", Status: "not_ready", Message: trimmedCommandError("describe remote EKS cluster", eksOut, err)},
		)
	case strings.TrimSpace(string(eksOut)) != "ACTIVE":
		checks = append(
			checks,
			starterStatusCheck{
				Name:    "aws-eks-active",
				Status:  "not_ready",
				Message: "remote EKS cluster status is " + strings.TrimSpace(string(eksOut)),
			},
		)
	default:
		checks = append(checks, starterStatusCheck{Name: "aws-eks-active", Status: "passed", Message: "remote EKS cluster is ACTIVE"})
	}

	readyOut, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", kubeContext,
		"get", "--raw=/readyz",
	)
	switch {
	case err != nil:
		checks = append(
			checks,
			starterStatusCheck{
				Name:    "remote-kube-readyz",
				Status:  "not_ready",
				Message: trimmedCommandError("query remote EKS readyz", readyOut, err),
			},
		)
	case !strings.Contains(strings.ToLower(string(readyOut)), "ok"):
		checks = append(checks, starterStatusCheck{Name: "remote-kube-readyz", Status: "not_ready", Message: "remote EKS readyz did not report ok"})
	default:
		checks = append(checks, starterStatusCheck{Name: "remote-kube-readyz", Status: "passed", Message: "remote Kubernetes API is reachable"})
	}

	checks = append(checks, starterSelectedImageStatusChecks(ctx, input, access)...)
	if session != nil && strings.TrimSpace(session.Name) != "" {
		checks = append(checks, starterMinimalTopologyStatusChecks(ctx, input, access)...)
	}

	for _, rollout := range starterRemoteWorkloadRollouts(session) {
		rolloutOut, err := access.run(ctx,
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"-n", rollout.namespace,
			"rollout", "status", rollout.resource,
			"--timeout=5s",
		)
		if err != nil {
			checks = append(
				checks,
				starterStatusCheck{
					Name:    rollout.checkName,
					Status:  "not_ready",
					Message: trimmedCommandError("check "+rollout.resource+" rollout", rolloutOut, err),
				},
			)
			continue
		}
		checks = append(
			checks,
			starterStatusCheck{
				Name:    rollout.checkName,
				Status:  "passed",
				Message: rollout.resource + " rollout is available",
			},
		)
	}
	if config.StarterAIFTSEnabled(session) {
		checks = append(checks, starterAIFTSCronJobStatusCheck(ctx, access))
	}
	if starterObservabilityEnabled(session) {
		checks = append(checks, starterObservabilityStatusChecks(ctx, input, access)...)
		checks = append(checks, starterDashboardAccessStatusCheck(ctx, input, access))
	}
	checks = append(checks, starterRetainedWorkloadGraphStatusCheck(ctx, input, access))
	checks = append(checks, starterLoadBalancerStatusChecks(ctx, input, access)...)
	checks = append(
		checks,
		starterTLSSecretStatusCheck(ctx, session, access),
		starterContractStubStatusCheck(ctx, access),
	)
	checks = append(checks, starterBaselineBridgeStatusChecks(ctx, input)...)
	checks = append(checks, starterGatewayDNSAnnotationCheck(ctx, access))
	regionalURL, endpointCheck := starterRegionalServiceEndpointCheck(ctx, access)
	checks = append(checks, endpointCheck)
	checks = append(checks, starterGlobalCatalogCheck(ctx, input, regionalURL)...)
	return checks
}

func starterAIFTSCronJobTarget(kubeconfig, kubeContext string) starterActivationWaitTarget {
	return starterActivationWaitTarget{
		Name:       "starter-fts-update-index-cronjob",
		Kind:       starterActivationWaitCronJob,
		Kubeconfig: kubeconfig,
		Context:    kubeContext,
		Namespace:  starterRemoteRegionalNamespace,
		Resource:   "cronjob/fts-update-index",
	}
}

func starterAIFTSCronJobStatusCheck(ctx context.Context, access starterKubeAccess) starterStatusCheck {
	target := starterAIFTSCronJobTarget(access.Kubeconfig, access.KubeContext)
	target.Remote = access.Remote
	_, passed, message := target.probe(ctx, access.ExecEnv)
	status := "not_ready"
	if passed {
		status = "passed"
	}
	return starterStatusCheck{Name: target.Name, Status: status, Message: message}
}

func starterRetainedWorkloadGraphStatusCheck(
	ctx context.Context,
	input starterLiveCheckInput,
	access starterKubeAccess,
) starterStatusCheck {
	check := starterStatusCheck{Name: "starter-retained-workload-graph", Status: "not_ready"}
	out, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"get", "deployments,statefulsets,daemonsets,gateways.gateway.serverless.pingcap.com", "--all-namespaces", "-o", "json",
	)
	if err != nil {
		check.Message = trimmedCommandError("read retained Starter workloads", out, err)
		return check
	}
	var list starterRuntimeWorkloadList
	if err := json.Unmarshal(out, &list); err != nil {
		check.Message = "parse retained Starter workloads: " + err.Error()
		return check
	}
	namespaces := map[string]bool{
		"calico-system":   true,
		"ingress-nginx":   true,
		"tigera-operator": true,
	}
	for _, namespace := range starterRemoteWorkloadNamespaces(input.Session) {
		namespaces[namespace] = true
	}
	activeGateways := make(map[string]starterActiveGateway)
	for i := range list.Items {
		workload := &list.Items[i]
		if workload.Kind == "Gateway" && workload.Spec.Version != "" {
			key := workload.Metadata.Namespace + "/" + workload.Metadata.Name
			activeGateways[key] = starterActiveGateway{
				Deployment: workload.Metadata.Name + "-" + workload.Spec.Version,
				UID:        workload.Metadata.UID,
			}
		}
	}
	for i := range list.Items {
		workload := &list.Items[i]
		key := workload.Metadata.Namespace + "/" + workload.Metadata.Labels["app"]
		active := activeGateways[key]
		if workload.Metadata.Name == active.Deployment &&
			starterGatewayDeploymentOwnedBy(workload, workload.Metadata.Labels["app"], active.UID) {
			active.Present = true
			activeGateways[key] = active
		}
	}
	checked := 0
	var pending []string
	for i := range list.Items {
		workload := &list.Items[i]
		if !namespaces[workload.Metadata.Namespace] {
			continue
		}
		if workload.Kind == "Gateway" {
			key := workload.Metadata.Namespace + "/" + workload.Metadata.Name
			active := activeGateways[key]
			if active.Deployment != "" && !active.Present {
				pending = append(pending, workload.Metadata.Namespace+"/deployment/"+active.Deployment+"=missing")
			}
			continue
		}
		if workload.Kind == "Deployment" && workload.Metadata.Labels["control-plane"] == "gateway-controller" {
			key := workload.Metadata.Namespace + "/" + workload.Metadata.Labels["app"]
			active := activeGateways[key]
			if active.Present && starterGatewayDeploymentOwnedBy(workload, workload.Metadata.Labels["app"], active.UID) &&
				workload.Metadata.Name != active.Deployment {
				continue
			}
		}
		var desired, ready int
		switch workload.Kind {
		case "Deployment":
			desired = 1
			if workload.Spec.Replicas != nil {
				desired = *workload.Spec.Replicas
			}
			ready = workload.Status.AvailableReplicas
		case "StatefulSet":
			desired = 1
			if workload.Spec.Replicas != nil {
				desired = *workload.Spec.Replicas
			}
			ready = workload.Status.ReadyReplicas
		case "DaemonSet":
			desired = workload.Status.DesiredNumberScheduled
			ready = workload.Status.NumberReady
		default:
			continue
		}
		if desired == 0 {
			continue
		}
		checked++
		if ready < desired {
			pending = append(pending, fmt.Sprintf(
				"%s/%s/%s=%d/%d",
				workload.Metadata.Namespace,
				strings.ToLower(workload.Kind),
				workload.Metadata.Name,
				ready,
				desired,
			))
		}
	}
	if checked == 0 {
		check.Message = "no positive-replica retained workloads were found"
		return check
	}
	if len(pending) > 0 {
		check.Message = fmt.Sprintf("retained workloads checked=%d; unavailable=%s", checked, strings.Join(pending, ","))
		return check
	}
	check.Status = "passed"
	check.Message = fmt.Sprintf("all %d positive-replica retained Deployments, StatefulSets, and DaemonSets are ready", checked)
	return check
}

func starterGatewayDeploymentOwnedBy(workload *starterRuntimeWorkload, gatewayName, gatewayUID string) bool {
	if workload.Kind != "Deployment" || workload.Metadata.Labels["control-plane"] != "gateway-controller" ||
		workload.Metadata.Labels["version"] != workload.Metadata.Name {
		return false
	}
	return slices.ContainsFunc(workload.Metadata.OwnerReferences, func(owner starterKubernetesOwnerReference) bool {
		return owner.Controller && strings.HasPrefix(owner.APIVersion, "gateway.serverless.pingcap.com/") &&
			owner.Kind == "Gateway" && owner.Name == gatewayName && owner.UID == gatewayUID
	})
}

func starterLoadBalancerStatusChecks(
	ctx context.Context,
	input starterLiveCheckInput,
	access starterKubeAccess,
) []starterStatusCheck {
	services := []struct {
		name      string
		namespace string
	}{
		{name: "dbaas-proxy", namespace: starterRemoteRegionalNamespace},
		{name: "tidb-gateway-new", namespace: starterRemoteRegionalNamespace},
		{name: "serverless-regional-service", namespace: starterRemoteRegionalNamespace},
		{name: "ops-service-svc", namespace: starterRemoteRegionalNamespace},
		{name: "ingress-nginx-controller", namespace: "ingress-nginx"},
	}
	if input.Session != nil {
		services = append(services, struct {
			name      string
			namespace string
		}{name: input.Session.Name + "-privatelink", namespace: starterRemoteRegionalNamespace})
	}
	if config.StarterAIFTSEnabled(input.Session) {
		services = append(services, struct {
			name      string
			namespace string
		}{name: "ai-gateway", namespace: starterRemoteRegionalNamespace})
	}
	const (
		nameAnnotation = "service.beta.kubernetes.io/aws-load-balancer-name"
		tagAnnotation  = "service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags"
	)
	var checks []starterStatusCheck
	for _, expected := range services {
		check := starterStatusCheck{Name: "starter-load-balancer-" + expected.name, Status: "not_ready"}
		out, err := access.run(ctx,
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"--namespace", expected.namespace,
			"get", "service/"+expected.name, "-o", "json",
		)
		if err != nil {
			check.Message = trimmedCommandError("read Starter LoadBalancer service/"+expected.name, out, err)
			checks = append(checks, check)
			continue
		}
		var service starterService
		if err := json.Unmarshal(out, &service); err != nil {
			check.Message = "parse Starter LoadBalancer service/" + expected.name + ": " + err.Error()
			checks = append(checks, check)
			continue
		}
		lbName := strings.TrimSpace(service.Metadata.Annotations[nameAnnotation])
		tags := service.Metadata.Annotations[tagAnnotation]
		lbHost := ""
		if len(service.Status.LoadBalancer.Ingress) > 0 {
			lbHost = firstNonEmpty(service.Status.LoadBalancer.Ingress[0].Hostname, service.Status.LoadBalancer.Ingress[0].IP)
		}
		var missing []string
		if service.Spec.Type != "LoadBalancer" {
			missing = append(missing, "type=LoadBalancer")
		}
		if input.Session == nil || !strings.HasPrefix(lbName, input.Session.Name+"-") {
			missing = append(missing, "session-prefixed-name")
		}
		for _, fragment := range []string{"cp9-owner=", "cp9-managed=true", "cp9-ttl="} {
			if !strings.Contains(tags, fragment) {
				missing = append(missing, fragment)
			}
		}
		if input.Session == nil || !strings.Contains(tags, "cp9-session="+input.Session.Name) {
			missing = append(missing, "cp9-session")
		}
		if lbHost == "" {
			missing = append(missing, "load-balancer-host")
		}
		if len(missing) > 0 {
			check.Message = "missing/invalid " + strings.Join(missing, ",")
			checks = append(checks, check)
			continue
		}
		check.Status = "passed"
		check.Message = fmt.Sprintf("%s is session-owned and ready at %s", lbName, lbHost)
		checks = append(checks, check)
	}
	return checks
}

func starterTLSSecretStatusCheck(
	ctx context.Context,
	session *config.SessionConfig,
	access starterKubeAccess,
) starterStatusCheck {
	check := starterStatusCheck{Name: "starter-required-tls-secrets", Status: "not_ready"}
	contracts := []struct {
		namespace string
		name      string
		keys      []string
	}{
		{namespace: "tidb-admin", name: "proxy-certificate-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-admin", name: "ops-service-cluster-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-admin", name: "manager-server-cert", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-admin", name: "gateway-letsencrypt-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-admin", name: "gateway-cert-inner-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-admin", name: "tidb-gateway-cluster-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-admin", name: "meta-store-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}},
		{namespace: "tidb-serverless", name: "serverless-cluster-pd-cluster-secret", keys: []string{"ca.crt", "ca.key", "tls.crt", "tls.key"}},
	}
	if config.StarterAIFTSEnabled(session) {
		contracts = append(contracts, struct {
			namespace string
			name      string
			keys      []string
		}{namespace: "tidb-admin", name: "ai-gateway-cluster-secret", keys: []string{"ca.crt", "tls.crt", "tls.key"}})
	}
	var missing []string
	for _, contract := range contracts {
		out, err := access.run(ctx,
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"--namespace", contract.namespace,
			"get", "secret/"+contract.name, "-o", "json",
		)
		if err != nil {
			missing = append(missing, contract.namespace+"/"+contract.name)
			continue
		}
		var secret starterSecret
		if err := json.Unmarshal(out, &secret); err != nil {
			missing = append(missing, contract.namespace+"/"+contract.name+":invalid-json")
			continue
		}
		for _, key := range contract.keys {
			if strings.TrimSpace(secret.Data[key]) == "" {
				missing = append(missing, contract.namespace+"/"+contract.name+":"+key)
			}
		}
	}
	if len(missing) > 0 {
		check.Message = "required TLS material missing: " + strings.Join(missing, ",")
		return check
	}
	check.Status = "passed"
	check.Message = fmt.Sprintf("all %d required TLS Secrets contain CA, certificate, and key material", len(contracts))
	return check
}

func starterContractStubStatusCheck(ctx context.Context, access starterKubeAccess) starterStatusCheck {
	check := starterStatusCheck{Name: "starter-contract-stub-health", Status: "not_ready"}
	script := `import json, urllib.request
version = "cp9.starter.contract-stub/v1"
for port in (9090, 9091):
    value = json.load(urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz", timeout=2))
    assert value == {"status": "ok", "contract_version": version}, value
rows = [json.loads(line) for line in urllib.request.urlopen("http://127.0.0.1:9090/ledger", timeout=2) if line.strip()]
rejected = [row for row in rows if not row.get("accepted")]
assert not rejected, f"rejected contract calls={len(rejected)}"
print(f"{version} accepted={len(rows)} rejected=0")`
	out, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", starterRemoteRegionalNamespace,
		"exec", "deployment/starter-contract-stub", "--", "python3", "-c", script,
	)
	if err != nil {
		check.Message = trimmedCommandError("verify Starter contract stub", out, err)
		return check
	}
	message := strings.TrimSpace(string(out))
	if !strings.Contains(message, starterContractStubVersion) {
		check.Message = "Starter contract stub did not report the frozen contract version"
		return check
	}
	check.Status = "passed"
	check.Message = message
	return check
}

func starterBaselineBridgeStatusChecks(ctx context.Context, input starterLiveCheckInput) []starterStatusCheck {
	if input.Baseline == nil {
		return []starterStatusCheck{{Name: "starter-baseline-bridges", Status: "not_ready", Message: "Dev full baseline is missing"}}
	}
	access := starterKubeAccess{Kubeconfig: input.Baseline.KubeconfigPath, KubeContext: input.Baseline.Context}
	bridges := []struct {
		name         string
		namespace    string
		expectedPath string
	}{
		{name: "dbaas-proxy", namespace: "tidb-admin", expectedPath: starterDBAASProxyHostPath(input.StateDir)},
		{name: "ops-service", namespace: "cp9-global", expectedPath: starterOpsServiceHostPath(input.StateDir)},
	}
	var checks []starterStatusCheck
	for _, bridge := range bridges {
		check := starterStatusCheck{Name: "starter-local-bridge-" + bridge.name, Status: "not_ready"}
		expectedData, err := os.ReadFile(bridge.expectedPath)
		if err != nil {
			check.Message = "read persisted remote host: " + err.Error()
			checks = append(checks, check)
			continue
		}
		out, err := starterStatusCommandRunner(ctx, nil, "kubectl",
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"--namespace", bridge.namespace,
			"get", "service/"+bridge.name, "-o", "json",
		)
		if err != nil {
			check.Message = trimmedCommandError("read local bridge service/"+bridge.name, out, err)
			checks = append(checks, check)
			continue
		}
		var service starterService
		if err := json.Unmarshal(out, &service); err != nil {
			check.Message = "parse local bridge service/" + bridge.name + ": " + err.Error()
			checks = append(checks, check)
			continue
		}
		expectedHost := strings.TrimSpace(string(expectedData))
		if service.Spec.Type != "ExternalName" || service.Spec.ExternalName != expectedHost || expectedHost == "" {
			check.Message = fmt.Sprintf(
				"expected ExternalName %q, got type=%s externalName=%q",
				expectedHost,
				service.Spec.Type,
				service.Spec.ExternalName,
			)
			checks = append(checks, check)
			continue
		}
		check.Status = "passed"
		check.Message = bridge.namespace + "/" + bridge.name + " targets " + expectedHost
		checks = append(checks, check)
	}
	for _, deployment := range []string{"serverless-service", "serverless-svc-grpc-server", "serverless-service-worker", "serverless-svc-br-worker"} {
		check := starterStatusCheck{Name: "starter-global-bridge-" + deployment, Status: "not_ready"}
		out, err := starterStatusCommandRunner(ctx, nil, "kubectl",
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"--namespace", "cp9-global",
			"rollout", "status", "deployment/"+deployment, "--timeout=5s",
		)
		if err != nil {
			check.Message = trimmedCommandError("check global bridge deployment/"+deployment, out, err)
		} else {
			check.Status = "passed"
			check.Message = "deployment/" + deployment + " is available"
		}
		checks = append(checks, check)
	}
	return checks
}

var starterServingLanes = []string{
	"tidb-pool",
	"vip-tidb-pool",
	"canary-tidb-pool",
	"canary-vip-tidb-pool",
	"export-tidb-pool",
	"super-vip-tidb-pool",
	"export-super-vip-tidb-pool",
	"export-vip-tidb-pool",
	"canary-super-vip-tidb-pool",
}

type starterKubernetesMetadata struct {
	Name      string            `json:"name"`
	Namespace string            `json:"namespace"`
	Labels    map[string]string `json:"labels"`
}

type starterKubernetesOwnerReference struct {
	APIVersion string `json:"apiVersion"`
	Kind       string `json:"kind"`
	Name       string `json:"name"`
	UID        string `json:"uid"`
	Controller bool   `json:"controller"`
}

type starterRuntimeWorkloadMetadata struct {
	Name            string                            `json:"name"`
	Namespace       string                            `json:"namespace"`
	UID             string                            `json:"uid"`
	Labels          map[string]string                 `json:"labels"`
	OwnerReferences []starterKubernetesOwnerReference `json:"ownerReferences"`
}

type starterActiveGateway struct {
	Deployment string
	UID        string
	Present    bool
}

type starterRuntimeWorkloadSpec struct {
	Replicas *int   `json:"replicas"`
	Version  string `json:"version"`
}

type starterRuntimeWorkloadStatus struct {
	AvailableReplicas      int `json:"availableReplicas"`
	ReadyReplicas          int `json:"readyReplicas"`
	DesiredNumberScheduled int `json:"desiredNumberScheduled"`
	NumberReady            int `json:"numberReady"`
}

type starterRuntimeWorkload struct {
	Kind     string                         `json:"kind"`
	Metadata starterRuntimeWorkloadMetadata `json:"metadata"`
	Spec     starterRuntimeWorkloadSpec     `json:"spec"`
	Status   starterRuntimeWorkloadStatus   `json:"status"`
}

type starterRuntimeWorkloadList struct {
	Items []starterRuntimeWorkload `json:"items"`
}

type starterServiceMetadata struct {
	Name        string            `json:"name"`
	Namespace   string            `json:"namespace"`
	Annotations map[string]string `json:"annotations"`
}

type starterServiceSpec struct {
	Type         string `json:"type"`
	ExternalName string `json:"externalName"`
}

type starterServiceIngress struct {
	Hostname string `json:"hostname"`
	IP       string `json:"ip"`
}

type starterServiceLoadBalancer struct {
	Ingress []starterServiceIngress `json:"ingress"`
}

type starterServiceStatus struct {
	LoadBalancer starterServiceLoadBalancer `json:"loadBalancer"`
}

type starterService struct {
	Metadata starterServiceMetadata `json:"metadata"`
	Spec     starterServiceSpec     `json:"spec"`
	Status   starterServiceStatus   `json:"status"`
}

type starterSecret struct {
	Data map[string]string `json:"data"`
}

type starterKubernetesCondition struct {
	Type   string `json:"type"`
	Status string `json:"status"`
}

type starterKubernetesStatus struct {
	Phase       string                       `json:"phase"`
	Conditions  []starterKubernetesCondition `json:"conditions"`
	Allocatable map[string]string            `json:"allocatable"`
}

type starterKubernetesNode struct {
	Metadata starterKubernetesMetadata `json:"metadata"`
	Status   starterKubernetesStatus   `json:"status"`
}

type starterKubernetesNodeList struct {
	Items []starterKubernetesNode `json:"items"`
}

type starterTiKVStorageRequest struct {
	Storage string `json:"storage"`
}

type starterTiKVStorageVolume struct {
	Name             string `json:"name"`
	StorageClassName string `json:"storageClassName"`
	StorageSize      string `json:"storageSize"`
}

type starterTiKVSpec struct {
	Replicas         int                        `json:"replicas"`
	StorageClassName string                     `json:"storageClassName"`
	Requests         starterTiKVStorageRequest  `json:"requests"`
	StorageVolumes   []starterTiKVStorageVolume `json:"storageVolumes"`
}

type starterTiDBClusterSpec struct {
	TiKV starterTiKVSpec `json:"tikv"`
}

type starterTiDBCluster struct {
	Metadata starterKubernetesMetadata `json:"metadata"`
	Spec     starterTiDBClusterSpec    `json:"spec"`
}

type starterTiDBClusterList struct {
	Items []starterTiDBCluster `json:"items"`
}

type starterPDReplicaSpec struct {
	Replicas int `json:"replicas"`
}

type starterPDResourceValues struct {
	CPU    string `json:"cpu"`
	Memory string `json:"memory"`
}

type starterPDWorkloadSpec struct {
	Replicas int                     `json:"replicas"`
	Requests starterPDResourceValues `json:"requests"`
	Limits   starterPDResourceValues `json:"limits"`
}

type starterRootPDClusterSpec struct {
	PD starterPDWorkloadSpec `json:"pd"`
}

type starterRootPDCluster struct {
	Spec starterRootPDClusterSpec `json:"spec"`
}

type starterStatefulSetStatus struct {
	ReadyReplicas int `json:"readyReplicas"`
}

type starterStatefulSet struct {
	Kind     string                    `json:"kind"`
	Metadata starterKubernetesMetadata `json:"metadata"`
	Spec     starterPDReplicaSpec      `json:"spec"`
	Status   starterStatefulSetStatus  `json:"status"`
}

type starterStatefulSetList struct {
	Items []starterStatefulSet `json:"items"`
}

type starterPodSpec struct {
	NodeName     string            `json:"nodeName"`
	NodeSelector map[string]string `json:"nodeSelector"`
}

type starterPod struct {
	Metadata starterKubernetesMetadata `json:"metadata"`
	Spec     starterPodSpec            `json:"spec"`
	Status   starterKubernetesStatus   `json:"status"`
}

type starterPodList struct {
	Items []starterPod `json:"items"`
}

type starterKeyspaceStatus struct {
	InPool      *bool  `json:"inPool"`
	BindCluster string `json:"bindCluster"`
	Phase       string `json:"phase"`
}

type starterKeyspace struct {
	Metadata starterKubernetesMetadata `json:"metadata"`
	Status   starterKeyspaceStatus     `json:"status"`
}

type starterKeyspaceList struct {
	Items []starterKeyspace `json:"items"`
}

func starterMinimalTopologyStatusChecks(
	ctx context.Context,
	input starterLiveCheckInput,
	access starterKubeAccess,
) []starterStatusCheck {
	topology := config.ResolvedStarterTopology(input.Session)
	groups := starterMinimalNodeGroups(&topology)
	var liveNodes *starterKubernetesNodeList
	nodeOut, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"get", "nodes",
		"-l", "cp9.tidbcloud.com/starter-role",
		"-o", "json",
	)
	checks := make([]starterStatusCheck, 0, len(groups)+2)
	if err != nil {
		message := trimmedCommandError("read Starter role nodes", nodeOut, err)
		for _, group := range groups {
			role := group.Labels["cp9.tidbcloud.com/starter-role"]
			checks = append(checks, starterStatusCheck{Name: "starter-nodegroup-" + role, Status: "not_ready", Message: message})
		}
	} else {
		var nodes starterKubernetesNodeList
		if err := json.Unmarshal(nodeOut, &nodes); err != nil {
			message := "parse Starter role nodes: " + err.Error()
			for _, group := range groups {
				role := group.Labels["cp9.tidbcloud.com/starter-role"]
				checks = append(checks, starterStatusCheck{Name: "starter-nodegroup-" + role, Status: "not_ready", Message: message})
			}
		} else {
			liveNodes = &nodes
			pd := config.ResolvedStarterPDTopology(input.Session)
			checks = append(checks, starterNodeGroupStatusChecks(
				groups,
				&nodes,
				&pd,
			)...)
		}
	}
	if liveNodes == nil {
		checks = append(checks, starterStatusCheck{
			Name:    "starter-workload-placement",
			Status:  "not_ready",
			Message: "Starter role nodes are unavailable, so workload placement cannot be verified",
		})
	} else {
		checks = append(checks, starterWorkloadPlacementStatusCheck(
			ctx,
			access,
			liveNodes,
		))
	}
	checks = append(checks,
		starterPDWorkloadTopologyStatusCheck(ctx, input.Session, access),
		starterTiKVStorageStatusCheck(ctx, access),
		starterSelectedLaneStatusCheck(ctx, input.Session, access),
	)
	return checks
}

func starterPDWorkloadTopologyStatusCheck(
	ctx context.Context,
	session *config.SessionConfig,
	access starterKubeAccess,
) starterStatusCheck {
	check := starterStatusCheck{Name: "starter-pd-workload-topology", Status: "not_ready"}
	rootOut, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", "tidb-serverless",
		"get", "tidbcluster", "serverless-cluster",
		"-o", "json",
	)
	if err != nil {
		check.Message = trimmedCommandError("read root PD workload contract", rootOut, err)
		return check
	}
	var root starterRootPDCluster
	if err := json.Unmarshal(rootOut, &root); err != nil {
		check.Message = "parse root PD workload contract: " + err.Error()
		return check
	}
	podOut, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", "tidb-serverless",
		"get", "pods",
		"-l", "app.kubernetes.io/component=pd,app.kubernetes.io/instance=serverless-cluster",
		"-o", "json",
	)
	if err != nil {
		check.Message = trimmedCommandError("read root PD Pods", podOut, err)
		return check
	}
	var pdPods starterPodList
	if err := json.Unmarshal(podOut, &pdPods); err != nil {
		check.Message = "parse root PD Pods: " + err.Error()
		return check
	}
	readyRootPD := 0
	pdNodes := map[string]struct{}{}
	for _, pod := range pdPods.Items {
		if node := strings.TrimSpace(pod.Spec.NodeName); node != "" {
			pdNodes[node] = struct{}{}
		}
		for _, condition := range pod.Status.Conditions {
			if condition.Type == "Ready" && condition.Status == "True" {
				readyRootPD++
				break
			}
		}
	}
	statefulOut, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", "tidb-serverless",
		"get", "statefulsets,services",
		"-o", "json",
	)
	if err != nil {
		check.Message = trimmedCommandError("read PD microservice workload contract", statefulOut, err)
		return check
	}
	var statefulSets starterStatefulSetList
	if err := json.Unmarshal(statefulOut, &statefulSets); err != nil {
		check.Message = "parse PD microservice workload contract: " + err.Error()
		return check
	}
	tsoReplicas := 0
	schedulerReplicas := 0
	resourceManagerReplicas := 0
	tsoReady := 0
	schedulerReady := 0
	resourceManagerReady := 0
	tsoService := false
	schedulerService := false
	resourceManagerService := false
	for _, statefulSet := range statefulSets.Items {
		switch statefulSet.Metadata.Name {
		case "pd-tso-server":
			tsoReplicas = statefulSet.Spec.Replicas
			tsoReady = statefulSet.Status.ReadyReplicas
		case "pd-sche-server":
			schedulerReplicas = statefulSet.Spec.Replicas
			schedulerReady = statefulSet.Status.ReadyReplicas
		case "pd-resource-manager-server":
			resourceManagerReplicas = statefulSet.Spec.Replicas
			resourceManagerReady = statefulSet.Status.ReadyReplicas
		case "tso-service":
			tsoService = true
		case "sche-service":
			schedulerService = true
		case "resource-manager-service":
			resourceManagerService = true
		}
	}
	pd := config.ResolvedStarterPDTopology(session)
	pdmsEnabled := pd.Mode == config.StarterPDModePDMS
	services := "none"
	if tsoService || schedulerService || resourceManagerService {
		var names []string
		if tsoService {
			names = append(names, "tso")
		}
		if schedulerService {
			names = append(names, "scheduler")
		}
		if resourceManagerService {
			names = append(names, "resource-manager")
		}
		services = strings.Join(names, ",")
	}
	check.Message = fmt.Sprintf(
		"mode=%s pd_nodes=3 instance_type=%s root_pd=%d/%d distinct_nodes=%d requests=%s/%s limits=%s/%s tso=%d/%d scheduler=%d/%d resource_manager=%d/%d services=%s",
		pd.Mode,
		config.ResolvedStarterTopology(session).PD.InstanceType,
		readyRootPD,
		root.Spec.PD.Replicas,
		len(pdNodes),
		root.Spec.PD.Requests.CPU,
		root.Spec.PD.Requests.Memory,
		root.Spec.PD.Limits.CPU,
		root.Spec.PD.Limits.Memory,
		tsoReady,
		tsoReplicas,
		schedulerReady,
		schedulerReplicas,
		resourceManagerReady,
		resourceManagerReplicas,
		services,
	)
	if root.Spec.PD.Replicas != pd.RootPDReplicas || tsoReplicas != pd.TSOReplicas || schedulerReplicas != pd.SchedulerReplicas ||
		resourceManagerReplicas != pd.ResourceManagerReplicas || tsoService != pdmsEnabled || schedulerService != pdmsEnabled ||
		resourceManagerService != pdmsEnabled {
		check.Message += fmt.Sprintf(
			"; hybrid or mismatched PD topology is forbidden (want root_pd=%d tso=%d scheduler=%d resource_manager=%d); run cp9 destroy before recreating with the selected starter.activation.pdms.enabled mode",
			pd.RootPDReplicas,
			pd.TSOReplicas,
			pd.SchedulerReplicas,
			pd.ResourceManagerReplicas,
		)
		return check
	}
	if readyRootPD != pd.RootPDReplicas || len(pdNodes) != pd.RootPDReplicas || tsoReady != pd.TSOReplicas ||
		schedulerReady != pd.SchedulerReplicas ||
		resourceManagerReady != pd.ResourceManagerReplicas {
		check.Message += "; PD workloads are not all Ready"
		return check
	}
	if root.Spec.PD.Requests.CPU != pd.RequestCPU || root.Spec.PD.Requests.Memory != pd.RequestMemory ||
		root.Spec.PD.Limits.CPU != pd.LimitCPU || root.Spec.PD.Limits.Memory != pd.LimitMemory {
		check.Message += fmt.Sprintf("; PD resources must match the selected mode contract %s/%s requests and %s/%s limits",
			pd.RequestCPU, pd.RequestMemory, pd.LimitCPU, pd.LimitMemory)
		return check
	}
	check.Status = "passed"
	return check
}

func starterWorkloadPlacementStatusCheck(
	ctx context.Context,
	access starterKubeAccess,
	nodes *starterKubernetesNodeList,
) starterStatusCheck {
	out, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"get", "pods",
		"--all-namespaces",
		"-o", "json",
	)
	if err != nil {
		return starterStatusCheck{
			Name:    "starter-workload-placement",
			Status:  "not_ready",
			Message: trimmedCommandError("read Starter workload Pods", out, err),
		}
	}
	var pods starterPodList
	if err := json.Unmarshal(out, &pods); err != nil {
		return starterStatusCheck{
			Name:    "starter-workload-placement",
			Status:  "not_ready",
			Message: "parse Starter workload Pods: " + err.Error(),
		}
	}
	nodeRoles := make(map[string]string, len(nodes.Items))
	for _, node := range nodes.Items {
		nodeRoles[node.Metadata.Name] = node.Metadata.Labels["cp9.tidbcloud.com/starter-role"]
	}
	requiredRoles := []string{"pd", "tikv", "tikv-hot", "keyspace-etcd", "gateway", "regional-control-plane", "tidb-serving"}
	roleCounts := make(map[string]int, len(requiredRoles))
	var mismatches []string
	for _, pod := range pods.Items {
		expectedRole := strings.TrimSpace(pod.Spec.NodeSelector["cp9.tidbcloud.com/starter-role"])
		if expectedRole == "" {
			continue
		}
		roleCounts[expectedRole]++
		podID := strings.Trim(strings.TrimSpace(pod.Metadata.Namespace)+"/"+strings.TrimSpace(pod.Metadata.Name), "/")
		actualRole := nodeRoles[strings.TrimSpace(pod.Spec.NodeName)]
		if strings.TrimSpace(pod.Spec.NodeName) == "" {
			mismatches = append(mismatches, podID+"=unscheduled(expected="+expectedRole+")")
			continue
		}
		if actualRole != expectedRole {
			mismatches = append(mismatches, fmt.Sprintf(
				"%s=%s(expected=%s,node=%s)",
				podID,
				firstNonEmpty(actualRole, "unlabelled"),
				expectedRole,
				pod.Spec.NodeName,
			))
		}
	}
	roleParts := make([]string, 0, len(requiredRoles))
	for _, role := range requiredRoles {
		roleParts = append(roleParts, fmt.Sprintf("%s=%d", role, roleCounts[role]))
		if roleCounts[role] == 0 && role != "tidb-serving" && role != "tikv-hot" {
			mismatches = append(mismatches, role+"=no-role-selected-pods")
		}
	}
	slices.Sort(mismatches)
	message := "role-selected Pods " + strings.Join(roleParts, ",")
	if len(mismatches) > 0 {
		return starterStatusCheck{
			Name:    "starter-workload-placement",
			Status:  "not_ready",
			Message: message + "; mismatches=" + strings.Join(mismatches, ","),
		}
	}
	return starterStatusCheck{
		Name:    "starter-workload-placement",
		Status:  "passed",
		Message: message + "; every selected Pod runs on the intended role-labelled node",
	}
}

func starterNodeGroupStatusChecks(
	groups []config.RemoteEKSNodeGroupConfig,
	nodes *starterKubernetesNodeList,
	pd *config.StarterPDTopologyContract,
) []starterStatusCheck {
	checks := make([]starterStatusCheck, 0, len(groups))
	for _, group := range groups {
		role := group.Labels["cp9.tidbcloud.com/starter-role"]
		observed := 0
		ready := 0
		architectures := map[string]struct{}{}
		instanceTypes := map[string]struct{}{}
		allocatableCPU := map[string]struct{}{}
		allocatableMemory := map[string]struct{}{}
		pdAllocatableFits := true
		if nodes != nil {
			for _, node := range nodes.Items {
				if node.Metadata.Labels["cp9.tidbcloud.com/starter-role"] != role {
					continue
				}
				observed++
				architectures[node.Metadata.Labels["kubernetes.io/arch"]] = struct{}{}
				instanceTypes[node.Metadata.Labels["node.kubernetes.io/instance-type"]] = struct{}{}
				if role == "pd" {
					cpu := node.Status.Allocatable["cpu"]
					memory := node.Status.Allocatable["memory"]
					allocatableCPU[cpu] = struct{}{}
					allocatableMemory[memory] = struct{}{}
					cpuQuantity, cpuErr := resource.ParseQuantity(cpu)
					memoryQuantity, memoryErr := resource.ParseQuantity(memory)
					requestCPU, requestCPUErr := resource.ParseQuantity(pd.RequestCPU)
					requestMemory, requestMemoryErr := resource.ParseQuantity(pd.RequestMemory)
					if cpuErr != nil || memoryErr != nil || requestCPUErr != nil || requestMemoryErr != nil ||
						cpuQuantity.Cmp(requestCPU) < 0 || memoryQuantity.Cmp(requestMemory) < 0 {
						pdAllocatableFits = false
					}
				}
				for _, condition := range node.Status.Conditions {
					if condition.Type == "Ready" && condition.Status == "True" {
						ready++
						break
					}
				}
			}
		}
		architectureList := make([]string, 0, len(architectures))
		for architecture := range architectures {
			architectureList = append(architectureList, architecture)
		}
		slices.Sort(architectureList)
		instanceTypeList := make([]string, 0, len(instanceTypes))
		for instanceType := range instanceTypes {
			instanceTypeList = append(instanceTypeList, instanceType)
		}
		slices.Sort(instanceTypeList)
		allocatableCPUList := slices.Sorted(maps.Keys(allocatableCPU))
		allocatableMemoryList := slices.Sorted(maps.Keys(allocatableMemory))
		message := fmt.Sprintf("desired=%d min=%d max=%d observed=%d ready=%d arch=%s expected_arch=%s instance_type=%s expected_instance_type=%s",
			group.DesiredSize, group.MinSize, group.MaxSize, observed, ready, strings.Join(architectureList, ","), group.Architecture,
			strings.Join(instanceTypeList, ","), group.InstanceType)
		if role == "pd" {
			message += fmt.Sprintf(" allocatable_cpu=%s allocatable_memory=%s pd_requests=%s/%s",
				strings.Join(allocatableCPUList, ","), strings.Join(allocatableMemoryList, ","), pd.RequestCPU, pd.RequestMemory)
		}
		status := "passed"
		if observed < group.MinSize || observed > group.MaxSize || ready != observed ||
			(observed > 0 && (len(architectureList) != 1 || architectureList[0] != group.Architecture ||
				len(instanceTypeList) != 1 || instanceTypeList[0] != group.InstanceType)) || !pdAllocatableFits {
			status = "not_ready"
		}
		checks = append(checks, starterStatusCheck{Name: "starter-nodegroup-" + role, Status: status, Message: message})
	}
	return checks
}

func starterTiKVStorageStatusCheck(ctx context.Context, access starterKubeAccess) starterStatusCheck {
	out, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", "tidb-serverless",
		"get", "tidbclusters.pingcap.com",
		"-o", "json",
	)
	if err != nil {
		return starterStatusCheck{
			Name:    "starter-tikv-storage",
			Status:  "not_ready",
			Message: trimmedCommandError("read TiKV storage contract", out, err),
		}
	}
	var clusters starterTiDBClusterList
	if err := json.Unmarshal(out, &clusters); err != nil {
		return starterStatusCheck{Name: "starter-tikv-storage", Status: "not_ready", Message: "parse TiKV storage contract: " + err.Error()}
	}
	want := map[string]struct {
		replicas int
		storage  string
	}{
		"tikv-r8gxl-34v260403-p0":      {replicas: 1, storage: "1Ti"},
		"tikv-r8gxl-34v260403-p1":      {replicas: 1, storage: "1Ti"},
		"tikv-r8gxl-34v260403-p2":      {replicas: 1, storage: "1Ti"},
		"tikv-r8gxl-34v260403-standby": {replicas: 1, storage: "1Ti"},
		"tikv-r8gxl-34v260403-witness": {replicas: 1, storage: "1Ti"},
		"tikv-r8g2xl-34v260403-p3":     {replicas: 0, storage: "500Gi"},
	}
	ready := 0
	var mismatches []string
	for _, tidbCluster := range clusters.Items {
		expected, ok := want[tidbCluster.Metadata.Name]
		if !ok {
			continue
		}
		volumes := map[string]string{}
		for _, volume := range tidbCluster.Spec.TiKV.StorageVolumes {
			volumes[volume.Name] = volume.StorageClassName + "/" + volume.StorageSize
		}
		if tidbCluster.Spec.TiKV.Replicas == expected.replicas && tidbCluster.Spec.TiKV.Requests.Storage == expected.storage &&
			tidbCluster.Spec.TiKV.StorageClassName == "gp3-500" &&
			volumes["raft-db"] == "ebs-sc/30Gi" && volumes["raft-wal"] == "ebs-sc/5Gi" {
			ready++
			continue
		}
		mismatches = append(
			mismatches,
			fmt.Sprintf("%s=replicas:%d primary:%s/%s raft-db:%s raft-wal:%s", tidbCluster.Metadata.Name,
				tidbCluster.Spec.TiKV.Replicas, tidbCluster.Spec.TiKV.StorageClassName,
				tidbCluster.Spec.TiKV.Requests.Storage, volumes["raft-db"], volumes["raft-wal"]),
		)
	}
	if ready != len(want) {
		return starterStatusCheck{
			Name:    "starter-tikv-storage",
			Status:  "not_ready",
			Message: fmt.Sprintf("TiKV storage ready=%d/%d; mismatches=%s", ready, len(want), strings.Join(mismatches, ",")),
		}
	}
	return starterStatusCheck{
		Name:    "starter-tikv-storage",
		Status:  "passed",
		Message: "five one-replica standard partitions use gp3-500 1Ti and the zero-replica hot template uses gp3-500 500Gi; all retain 30Gi raft-db and 5Gi raft-wal volumes",
	}
}

func starterSelectedLaneStatusCheck(
	ctx context.Context,
	session *config.SessionConfig,
	access starterKubeAccess,
) starterStatusCheck {
	readPods := func(namespace string) (*starterPodList, error) {
		out, err := access.run(ctx,
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"--namespace", namespace,
			"get", "pods",
			"-l", "app.kubernetes.io/component=serverless-tidb",
			"-o", "json",
		)
		if err != nil {
			return nil, fmt.Errorf("%s", trimmedCommandError("read "+namespace+" TiDB Pods", out, err))
		}
		var pods starterPodList
		if err := json.Unmarshal(out, &pods); err != nil {
			return nil, fmt.Errorf("parse %s TiDB Pods: %w", namespace, err)
		}
		return &pods, nil
	}
	lanePods := make(map[string]*starterPodList, len(starterServingLanes))
	for _, lane := range starterServingLanes {
		pods, err := readPods(lane)
		if err != nil {
			return starterStatusCheck{Name: "starter-selected-serving-lane", Status: "not_ready", Message: err.Error()}
		}
		lanePods[lane] = pods
	}
	keyspaceOut, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"get", "keyspaces.serverless.pingcap.com",
		"--all-namespaces",
		"-o", "json",
	)
	if err != nil {
		return starterStatusCheck{
			Name:    "starter-selected-serving-lane",
			Status:  "not_ready",
			Message: trimmedCommandError("read Starter Keyspaces", keyspaceOut, err),
		}
	}
	var keyspaces starterKeyspaceList
	if err := json.Unmarshal(keyspaceOut, &keyspaces); err != nil {
		return starterStatusCheck{Name: "starter-selected-serving-lane", Status: "not_ready", Message: "parse Starter Keyspaces: " + err.Error()}
	}
	target := config.ResolvedStarterTiDBPoolPods(session)
	selectedPods := lanePods[starterSelectedServingLane]
	if selectedPods == nil {
		return starterStatusCheck{
			Name:    "starter-selected-serving-lane",
			Status:  "not_ready",
			Message: "selected Starter serving lane is missing from the reviewed product policy",
		}
	}
	selectedLive := len(selectedPods.Items)
	selectedReady := 0
	selectedPending := 0
	laneCounts := make([]string, 0, len(starterServingLanes))
	distribution := map[string]int{}
	for _, lane := range starterServingLanes {
		pods := lanePods[lane]
		laneCounts = append(laneCounts, fmt.Sprintf("%s=%d", lane, len(pods.Items)))
		if lane != starterSelectedServingLane {
			continue
		}
		for _, pod := range pods.Items {
			podReady := false
			for _, condition := range pod.Status.Conditions {
				if condition.Type == "Ready" && condition.Status == "True" {
					podReady = true
					break
				}
			}
			if podReady {
				selectedReady++
			} else {
				selectedPending++
			}
			if pod.Spec.NodeName != "" {
				distribution[pod.Spec.NodeName]++
			}
		}
	}
	distributionParts := make([]string, 0, len(distribution))
	for node, count := range distribution {
		distributionParts = append(distributionParts, fmt.Sprintf("%s=%d", node, count))
	}
	slices.Sort(distributionParts)
	allocatedKeyspaces := 0
	availableKeyspaces := 0
	for _, keyspace := range keyspaces.Items {
		if keyspace.Status.InPool == nil || !*keyspace.Status.InPool || strings.TrimSpace(keyspace.Status.BindCluster) != "" {
			allocatedKeyspaces++
		}
		if keyspace.Status.InPool != nil && *keyspace.Status.InPool &&
			strings.TrimSpace(keyspace.Status.BindCluster) == "" && keyspace.Status.Phase == "Available" {
			if keyspace.Metadata.Namespace == "keyspace-pool" {
				availableKeyspaces++
			}
		}
	}
	message := fmt.Sprintf(
		"policy=%s selected=%s configured_target=%d lanes=%s selected_live=%d selected_ready=%d selected_pending=%d keyspaces=%d allocated=%d available=keyspace-pool:%d pod_to_node=%s",
		starterCatalogServicePlan,
		starterSelectedServingLane,
		target,
		strings.Join(laneCounts, ","),
		selectedLive,
		selectedReady,
		selectedPending,
		len(keyspaces.Items),
		allocatedKeyspaces,
		availableKeyspaces,
		strings.Join(distributionParts, ","),
	)
	if selectedLive < target {
		return starterStatusCheck{
			Name:    "starter-selected-serving-lane",
			Status:  "not_ready",
			Message: message + "; selected lane has fewer live TiDB Pods than the configured initial/minimum target",
		}
	}
	if selectedReady < target || selectedPending > 0 {
		return starterStatusCheck{
			Name:    "starter-selected-serving-lane",
			Status:  "not_ready",
			Message: message + "; selected-lane TiDB serving Pods are not all Ready",
		}
	}
	if availableKeyspaces == 0 {
		return starterStatusCheck{
			Name:    "starter-selected-serving-lane",
			Status:  "not_ready",
			Message: message + "; Starter product keyspace pool has no available keyspace",
		}
	}
	return starterStatusCheck{Name: "starter-selected-serving-lane", Status: "passed", Message: message}
}

func starterSelectedImageStatusChecks(
	ctx context.Context,
	input starterLiveCheckInput,
	access starterKubeAccess,
) []starterStatusCheck {
	var state *cluster.State
	if strings.TrimSpace(input.StateDir) != "" {
		if loaded, lerr := cluster.LoadState(input.StateDir); lerr == nil {
			state = loaded
		}
	}
	snapshot, err := loadStarterRemoteSelectedImageSnapshot(input.StateDir, state)
	if err != nil {
		return []starterStatusCheck{{
			Name:       "starter-selected-image-snapshot",
			Status:     "not_ready",
			ReasonCode: lifecycle.StarterReasonInvalidImagePin,
			Message:    err.Error(),
		}}
	}
	if snapshot == nil || !snapshot.SourceSummaryLoaded || len(snapshot.SelectedSourceImages) == 0 {
		return nil
	}
	var checks []starterStatusCheck
	for _, selected := range snapshot.SelectedSourceImages {
		checkName := "starter-" + selected.Service + "-source-image"
		jsonPath := "jsonpath={.spec.template.spec.containers[?(@.name=='" + selected.Container + "')].image}"
		workloadResource := selected.resourceKind() + "/" + selected.Deployment
		out, err := access.run(ctx,
			"--kubeconfig", access.Kubeconfig,
			"--context", access.KubeContext,
			"--namespace", selected.Namespace,
			"get", workloadResource,
			"-o", jsonPath,
		)
		currentImage := strings.TrimSpace(string(out))
		if err != nil {
			checks = append(checks, starterStatusCheck{
				Name:    checkName,
				Status:  "not_ready",
				Service: selected.Service,
				Pin:     selected.Image,
				Message: "selected source-backed image for " + selected.Service +
					" waits for activation/substrate/workload readiness prerequisite: " +
					trimmedCommandError("read "+selected.Namespace+"/"+selected.Deployment+" image", out, err),
			})
			continue
		}
		if currentImage == "" {
			checks = append(checks, starterStatusCheck{
				Name:    checkName,
				Status:  "not_ready",
				Service: selected.Service,
				Pin:     selected.Image,
				Message: "selected source-backed image for " + selected.Service + " waits for activation/substrate/workload readiness prerequisite: container " + selected.Container + " is absent on " + selected.Namespace + "/" + selected.Deployment,
			})
			continue
		}
		if currentImage != selected.Image {
			checks = append(checks, starterStatusCheck{
				Name:       checkName,
				Status:     "not_ready",
				ReasonCode: lifecycle.StarterReasonInvalidImagePin,
				Service:    selected.Service,
				Pin:        selected.Image,
				Message:    "selected source-backed image mismatch for " + selected.Service + ": running " + currentImage + ", selected " + selected.Image + " from " + snapshot.SourceSummaryPath,
			})
			continue
		}
		checks = append(checks, starterStatusCheck{
			Name:    checkName,
			Status:  "passed",
			Message: selected.Service + " runs selected source-backed image " + selected.Image,
		})
	}
	return checks
}

func starterGatewayDNSAnnotationCheck(ctx context.Context, access starterKubeAccess) starterStatusCheck {
	if access.Kubeconfig == "" || access.KubeContext == "" {
		return starterStatusCheck{
			Name:    "starter-tidb-gateway-dns-annotation",
			Status:  "not_ready",
			Message: "remote EKS kubeconfig/context is missing",
		}
	}
	out, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", starterRemoteRegionalNamespace,
		"get", "service/tidb-gateway-new",
		"-o", "jsonpath={.metadata.annotations.gateway\\.tidbcloud\\.com/dnsName}",
	)
	if err != nil {
		return starterStatusCheck{
			Name:    "starter-tidb-gateway-dns-annotation",
			Status:  "not_ready",
			Message: trimmedCommandError("read tidb-gateway-new DNS annotation", out, err),
		}
	}
	dnsName := strings.TrimSpace(string(out))
	if dnsName == "" {
		return starterStatusCheck{
			Name:    "starter-tidb-gateway-dns-annotation",
			Status:  "not_ready",
			Message: "tidb-gateway-new is missing " + starterGatewayDNSAnnotation + "; run cp9 starter activate",
		}
	}
	return starterStatusCheck{
		Name:    "starter-tidb-gateway-dns-annotation",
		Status:  "passed",
		Message: "tidb-gateway-new " + starterGatewayDNSAnnotation + "=" + dnsName,
	}
}

func starterRegionalServiceEndpointCheck(ctx context.Context, access starterKubeAccess) (string, starterStatusCheck) {
	if access.Kubeconfig == "" || access.KubeContext == "" {
		return "", starterStatusCheck{
			Name:    "starter-regional-service-endpoint",
			Status:  "not_ready",
			Message: "remote EKS kubeconfig/context is missing",
		}
	}
	out, err := access.run(ctx,
		"--kubeconfig", access.Kubeconfig,
		"--context", access.KubeContext,
		"--namespace", starterRemoteRegionalNamespace,
		"get", "service/serverless-regional-service",
		"-o", "jsonpath={.status.loadBalancer.ingress[0].hostname}{.status.loadBalancer.ingress[0].ip}",
	)
	if err != nil {
		return "", starterStatusCheck{
			Name:    "starter-regional-service-endpoint",
			Status:  "not_ready",
			Message: trimmedCommandError("read Starter regional service endpoint", out, err),
		}
	}
	endpointHost := strings.TrimSpace(string(out))
	if endpointHost == "" {
		return "", starterStatusCheck{
			Name:    "starter-regional-service-endpoint",
			Status:  "not_ready",
			Message: "serverless-regional-service has no LoadBalancer host/IP",
		}
	}
	url := "http://" + endpointHost
	return url, starterStatusCheck{
		Name:    "starter-regional-service-endpoint",
		Status:  "passed",
		Message: url,
	}
}

func starterGlobalCatalogCheck(ctx context.Context, input starterLiveCheckInput, regionalURL string) []starterStatusCheck {
	regionalURL = strings.TrimSpace(regionalURL)
	if regionalURL == "" {
		return []starterStatusCheck{{
			Name:    "starter-global-catalog",
			Status:  "not_ready",
			Message: "Starter regional service URL is missing",
		}}
	}
	baseline := input.Baseline
	if baseline == nil || strings.TrimSpace(baseline.KubeconfigPath) == "" || strings.TrimSpace(baseline.Context) == "" {
		return []starterStatusCheck{{
			Name:    "starter-global-catalog",
			Status:  "not_ready",
			Message: "Dev full baseline kubeconfig is missing; run cp9 up first",
		}}
	}
	if _, err := ensureStarterBaselineKubeAPITunnel(ctx, baseline); err != nil {
		return []starterStatusCheck{{
			Name:    "starter-global-catalog",
			Status:  "not_ready",
			Message: "repair Dev full baseline kube API tunnel before Starter global catalog query: " + err.Error(),
		}}
	}
	query := starterGlobalCatalogQuery(regionalURL)
	out, err := starterStatusCommandRunner(ctx, nil, "kubectl",
		"--kubeconfig", baseline.KubeconfigPath,
		"--context", baseline.Context,
		"--namespace", "cp9-system",
		"exec", "deploy/premium-metadb", "--",
		"mysql", "-uroot", "-pcp9-local", "-Nse", query,
	)
	if err != nil {
		return []starterStatusCheck{{
			Name:    "starter-global-catalog",
			Status:  "not_ready",
			Message: trimmedCommandError("query Starter global catalog", out, err),
		}}
	}
	state, err := parseStarterCatalogState(out)
	if err != nil {
		return []starterStatusCheck{
			{
				Name:    "starter-global-catalog",
				Status:  "not_ready",
				Message: "parse Starter global catalog state: " + err.Error(),
			},
		}
	}
	if !state.matchesProductActivation() {
		return []starterStatusCheck{{
			Name:   "starter-global-catalog",
			Status: "not_ready",
			Message: fmt.Sprintf(
				"Starter global catalog state does not match product activation: active Starter endpoint rows=%d active Essential v1 endpoint rows=%d total Essential v1 rows=%d Essential v2 flag bindings=%d",
				state.ActiveStarterRowsAtEndpoint,
				state.ActiveEssentialV1RowsAtEndpoint,
				state.TotalEssentialV1Rows,
				state.EssentialV2FlagBindings,
			),
		}}
	}
	message := "Starter and Essential v1 global catalog routes select " + starterSelectedServingLane + " at " + regionalURL
	if state.EssentialV2FlagBindings == 1 {
		message = "Starter and Essential v1 remain selected while Essential v2 is enabled for the alternate organization at " + regionalURL
	}
	return []starterStatusCheck{
		{
			Name:    "starter-global-catalog",
			Status:  "passed",
			Message: message,
		},
	}
}

type starterCatalogState struct {
	ActiveStarterRowsAtEndpoint     int
	ActiveEssentialV1RowsAtEndpoint int
	TotalEssentialV1Rows            int
	EssentialV2FlagBindings         int
}

func (s starterCatalogState) matchesProductActivation() bool {
	return s.ActiveStarterRowsAtEndpoint == 1 &&
		s.ActiveEssentialV1RowsAtEndpoint == 1 &&
		s.TotalEssentialV1Rows == 1 &&
		s.EssentialV2FlagBindings <= 1
}

func starterGlobalCatalogQuery(regionalURL string) string {
	regionalURL = sqlStringLiteral(regionalURL)
	regionCount := func(plan, suffix string) string {
		return "(SELECT COUNT(*) FROM serverless_global_control_dev.providers p " +
			"JOIN serverless_global_control_dev.regions r ON r.provider_id=p.id " +
			"WHERE p.name='aws' AND r.name='us-west-2' AND r.service_plan='" + plan + "'" + suffix + ")"
	}
	activeAtStarter := " AND r.status='active' AND r.is_public=1 AND r.service_address = '" + regionalURL + "'"
	return "SELECT " +
		regionCount("Starter", activeAtStarter) + ", " +
		regionCount("Essential", activeAtStarter) + ", " +
		regionCount("Essential", "") + ", " +
		"(SELECT COUNT(*) FROM account.feature_flag_orgs o " +
		"JOIN account.feature_flags f ON f.id=o.flag_id " +
		"WHERE f.flag_name='essentialv2_enable' AND o.org_id=" + config.EssentialV2UIOrganizationID + ");"
}

func parseStarterCatalogState(output []byte) (starterCatalogState, error) {
	lines := strings.Split(strings.TrimSpace(string(output)), "\n")
	for _, value := range slices.Backward(lines) {
		fields := strings.Fields(value)
		if len(fields) != 4 {
			continue
		}
		counts := [4]int{}
		valid := true
		for i, field := range fields {
			count, err := strconv.Atoi(field)
			if err != nil {
				valid = false
				break
			}
			counts[i] = count
		}
		if valid {
			return starterCatalogState{
				ActiveStarterRowsAtEndpoint:     counts[0],
				ActiveEssentialV1RowsAtEndpoint: counts[1],
				TotalEssentialV1Rows:            counts[2],
				EssentialV2FlagBindings:         counts[3],
			}, nil
		}
	}
	return starterCatalogState{}, fmt.Errorf("expected four integer columns, got %q", strings.TrimSpace(string(output)))
}

func sqlStringLiteral(value string) string {
	return strings.ReplaceAll(value, "'", "''")
}

type starterRemoteWorkloadRollout struct {
	checkName string
	namespace string
	resource  string
}

func starterRemoteWorkloadRollouts(session *config.SessionConfig) []starterRemoteWorkloadRollout {
	rollouts := []starterRemoteWorkloadRollout{
		{checkName: "starter-contract-stub-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/starter-contract-stub"},
		{checkName: "starter-cluster-proxy-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/dbaas-proxy"},
		{checkName: "starter-gateway-controller-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/gateway-controller"},
		{checkName: "starter-tidb-http-gateway-rollout", namespace: starterRemoteRegionalNamespace, resource: "statefulset/tidb-http-gateway"},
		{checkName: "starter-meta-service-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/serverless-meta-server"},
		{checkName: "starter-meta-store-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/meta-store"},
		{checkName: "starter-vpa-agent-rollout", namespace: "tidb-vpa", resource: "daemonset/vpa-agent"},
		{checkName: "starter-manager-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/manager"},
		{checkName: "starter-ops-service-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/ops-service"},
		{
			checkName: "starter-regional-service-rollout",
			namespace: starterRemoteRegionalNamespace,
			resource:  "deployment/serverless-regional-service",
		},
		{checkName: "starter-tidb-gateway-inner-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/tidb-gateway-inner"},
		{checkName: "starter-tidb-worker-scaler-rollout", namespace: "tidb-worker", resource: "deployment/scaler"},
		{checkName: "starter-tikv-api-rollout", namespace: "tidb-serverless", resource: "statefulset/tikv-api"},
	}
	if config.StarterAIFTSEnabled(session) {
		rollouts = append(rollouts, starterRemoteAIFTSWorkloadRollouts()...)
	}
	return rollouts
}

func starterRemoteAIFTSWorkloadRollouts() []starterRemoteWorkloadRollout {
	return []starterRemoteWorkloadRollout{
		{checkName: "starter-ai-gateway-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/ai-gateway"},
		{checkName: "starter-qdrant-snapshot-rollout", namespace: starterRemoteRegionalNamespace, resource: "deployment/qdrant4snapshot"},
		{checkName: "starter-fts-server-rollout", namespace: starterRemoteRegionalNamespace, resource: "statefulset/fts-server"},
	}
}

func starterAWSExecEnv(profile, region string) []string {
	var env []string
	if strings.TrimSpace(profile) != "" {
		env = append(env, "AWS_PROFILE="+strings.TrimSpace(profile))
	}
	if strings.TrimSpace(region) != "" {
		region = strings.TrimSpace(region)
		env = append(env, "AWS_REGION="+region, "AWS_DEFAULT_REGION="+region)
	}
	return env
}

func trimmedCommandError(action string, output []byte, err error) string {
	msg := strings.TrimSpace(string(output))
	if msg == "" && err != nil {
		msg = err.Error()
	}
	if msg == "" {
		msg = "unknown error"
	}
	return action + ": " + msg
}

func runStarterStatusCommand(ctx context.Context, env []string, name string, args ...string) ([]byte, error) {
	cmd := exec.CommandContext(ctx, name, args...)
	if len(env) > 0 {
		cmd.Env = append(os.Environ(), env...)
	}
	return verbose.RunCommand(cmd)
}

func runStarterSubstrateActivation(ctx context.Context, stateDir string, session *config.SessionConfig) error {
	if session == nil {
		return cp9errors.New(cp9errors.CategoryConfig, "starter activation session is required")
	}
	if err := validateStarterAIFTSInputs(session); err != nil {
		return err
	}
	state, err := cluster.LoadState(stateDir)
	if err != nil {
		return fmt.Errorf("load lifecycle state: %w", err)
	}
	if strings.TrimSpace(state.Owner) == "" {
		if awsState, aerr := cpaws.LoadSessionState(stateDir); aerr == nil {
			state.Owner = strings.TrimSpace(awsState.Owner)
		}
	}
	if err := recoverTaintedStarterRemoteEKS(ctx, stateDir, session); err != nil {
		return err
	}
	runner := cpterraform.Runner{}
	for _, step := range []cluster.StateAwareStep{
		&lifecycle.SubstrateInitStep{StateDir: stateDir, Session: session, Runner: runner},
		&lifecycle.SubstratePlanStep{StateDir: stateDir, Session: session, Runner: runner},
		&lifecycle.SubstrateApplyStep{StateDir: stateDir, Session: session, Runner: runner},
	} {
		if err := step.RunWithState(ctx, state); err != nil {
			return err
		}
	}
	if err := cluster.SaveState(stateDir, state); err != nil {
		return fmt.Errorf("persist lifecycle state: %w", err)
	}
	return nil
}

type starterTerraformState struct {
	Resources []starterTerraformStateResource `json:"resources"`
}

type starterTerraformStateResource struct {
	Type      string                          `json:"type"`
	Name      string                          `json:"name"`
	Instances []starterTerraformStateInstance `json:"instances"`
}

type starterTerraformStateInstance struct {
	Status     string                          `json:"status"`
	Attributes starterTerraformStateAttributes `json:"attributes"`
}

type starterTerraformStateAttributes struct {
	ID string `json:"id"`
}

func recoverTaintedStarterRemoteEKS(ctx context.Context, stateDir string, session *config.SessionConfig) error {
	remote := session.Target.AWS.RemoteEKS
	if !remote.Enabled || strings.TrimSpace(remote.ClusterName) == "" {
		return nil
	}
	statePath, err := filepath.Abs(filepath.Join(stateDir, "terraform", "terraform.tfstate"))
	if err != nil {
		return cp9errors.Newf(cp9errors.CategoryResource, "resolve Terraform state before Starter resume: %v", err)
	}
	data, err := os.ReadFile(statePath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return cp9errors.Newf(cp9errors.CategoryResource, "read Terraform state before Starter resume: %v", err)
	}
	var state starterTerraformState
	if err := json.Unmarshal(data, &state); err != nil {
		return cp9errors.Newf(cp9errors.CategoryResource, "parse Terraform state before Starter resume: %v", err)
	}
	tainted := false
	for _, entry := range state.Resources {
		if entry.Type != "aws_eks_cluster" || entry.Name != "starter_remote" {
			continue
		}
		for _, instance := range entry.Instances {
			if instance.Status == "tainted" && instance.Attributes.ID == remote.ClusterName {
				tainted = true
			}
		}
	}
	if !tainted {
		return nil
	}
	profile := firstNonEmpty(session.Target.AWS.IAM.Profile, "cp9")
	region := strings.TrimSpace(session.Target.Region)
	output, err := starterStatusCommandRunner(
		ctx,
		nil,
		"aws",
		"eks", "describe-cluster",
		"--profile", profile,
		"--region", region,
		"--name", remote.ClusterName,
		"--query", "cluster.status",
		"--output", "text",
	)
	if err != nil {
		message := trimmedCommandError("aws eks describe-cluster", output, err)
		if strings.Contains(message, "ResourceNotFoundException") {
			return nil
		}
		return cp9errors.Newf(
			cp9errors.CategoryResource,
			"verify tainted Starter EKS %q before resume; refusing a replacement plan: %s",
			remote.ClusterName,
			message,
		)
	}
	status := strings.TrimSpace(string(output))
	if status != "ACTIVE" && status != "CREATING" {
		return nil
	}
	address := "aws_eks_cluster.starter_remote[0]"
	output, err = starterStatusCommandRunner(
		ctx,
		nil,
		"terraform",
		"-chdir="+filepath.Join(filepath.Dir(statePath), "source"),
		"untaint",
		"-state="+statePath,
		address,
	)
	if err != nil {
		return cp9errors.Newf(
			cp9errors.CategoryResource,
			"recover live tainted Starter EKS %q: %s",
			remote.ClusterName,
			trimmedCommandError("terraform untaint", output, err),
		)
	}
	verbose.Logf("starter activation resume recovered live tainted EKS cluster=%s status=%s", remote.ClusterName, status)
	return nil
}

type starterRemoteWorkloadActivationPlan struct {
	Env                   []string
	Images                []starterRemoteWorkloadImage
	KernelImages          []starterKernelImagePin
	Commands              []starterRemoteWorkloadCommand
	Remote                *starterRemoteWorkloadExecution
	Runtime               starterRemoteWorkloadRuntime
	WaitTargets           []starterActivationWaitTarget
	SelectedImageSnapshot *starterRemoteSelectedImageSnapshot
	StateDir              string
}

// starterKernelImagePin is a TiDB kernel or kernel-adjacent image reference that
// cp9 ships as a Starter activation input. These are not sourcebacked contracts,
// so they are outside plan.Images and would otherwise reach the cluster without any
// existence check: a mistyped tag would surface only as ImagePullBackOff.
type starterKernelImagePin struct {
	// Name identifies the input for the preflight check name and message.
	Name string
	// Source is the repo-relative file the reference was read from.
	Source string
	// Image is the full reference.
	Image string
}

type starterRemoteWorkloadCommand struct {
	Name   string
	Remote bool
	Script string
}

type starterRemoteWorkloadExecution struct {
	Selected            *host.Session
	TiltStatus          *cptilt.Status
	LocalStateDir       string
	LocalKubeconfigPath string
	KubeconfigPath      string
	ContextPath         string
	EnvironmentPath     string
	EvidenceDir         string
}

type starterRemoteWorkloadImage struct {
	Service    string `json:"service"`
	Image      string `json:"image"`
	Source     string `json:"source"`
	Provenance string `json:"provenance"`
	ValidUntil string `json:"valid_until,omitempty"`
}

type starterImageRegistryCheckFunc func(context.Context, []string, string, ecrImageReference) (bool, error)

var starterImagePreflightNow = time.Now

type starterReadinessError struct {
	Checks []starterStatusCheck
}

func (e *starterReadinessError) Error() string {
	var failures []string
	for i := range e.Checks {
		check := &e.Checks[i]
		if check.Status != "passed" {
			failures = append(failures, check.Name+": "+check.Message)
		}
	}
	return "starter readiness preflight failed: " + strings.Join(failures, "; ")
}

type starterRemoteSelectedImage struct {
	Service    string `json:"service"`
	Image      string `json:"image"`
	Kind       string `json:"kind,omitempty"`
	Namespace  string `json:"namespace"`
	Deployment string `json:"deployment"`
	Container  string `json:"container"`
}

func (i *starterRemoteSelectedImage) resourceKind() string {
	switch i.Kind {
	case starterRemoteSelectedImageStatefulSetKind:
		return "statefulset"
	case starterRemoteSelectedImageDaemonSetKind:
		return "daemonset"
	default:
		return "deployment"
	}
}

type starterRemoteSelectedImageSnapshot struct {
	APIVersion           string                       `json:"apiVersion"`
	Kind                 string                       `json:"kind"`
	SourceSummaryPath    string                       `json:"source_summary_path,omitempty"`
	SourceSummaryLoaded  bool                         `json:"source_summary_loaded"`
	SourceBuildServices  []string                     `json:"source_build_services,omitempty"`
	ActivationImages     []starterRemoteWorkloadImage `json:"activation_images,omitempty"`
	SelectedSourceImages []starterRemoteSelectedImage `json:"selected_source_images,omitempty"`
}

type starterRemoteWorkloadRuntime struct {
	ImagePullProfileOverride string
	AWSAccountID             string
	AWSProfile               string
	AWSRegion                string
	ComponentPCAARN          string
	ComponentPCARegion       string
	SourceECRAccountID       string
	SourceECRProfile         string
	SourceECRRegion          string
	StorageBucket            string
	StoragePrefix            string
	StorageRegion            string
	UseDefaultAWSCredentials bool
}

func runStarterWorkloadActivation(ctx context.Context, stateDir string, baseline *host.Session, session *config.SessionConfig) error {
	if session == nil {
		return cp9errors.New(cp9errors.CategoryConfig, "starter activation session is required")
	}
	if baseline == nil {
		return cp9errors.New(cp9errors.CategoryConfig, "Dev full baseline host is required")
	}
	remote, err := loadStarterRemoteEKSArtifact(stateDir)
	if err != nil {
		return err
	}
	state, err := cluster.LoadState(stateDir)
	if err != nil {
		return fmt.Errorf("load lifecycle state: %w", err)
	}
	if strings.TrimSpace(state.Owner) == "" {
		if awsState, aerr := cpaws.LoadSessionState(stateDir); aerr == nil {
			state.Owner = strings.TrimSpace(awsState.Owner)
		}
	}
	if err := (&lifecycle.SessionContextWriteStep{StateDir: stateDir}).RunWithState(ctx, state); err != nil {
		return err
	}
	if err := cluster.SaveState(stateDir, state); err != nil {
		return fmt.Errorf("persist lifecycle state: %w", err)
	}
	plan, err := buildStarterRemoteWorkloadActivationPlan(stateDir, baseline, session, state, remote)
	if err != nil {
		return err
	}
	return runStarterRemoteWorkloadActivationPlan(ctx, plan)
}

func runStarterRemoteWorkloadActivationPlan(ctx context.Context, plan *starterRemoteWorkloadActivationPlan) (runErr error) {
	checks := starterImagePreflightChecks(ctx, plan)
	for i := range checks {
		check := &checks[i]
		if check.Status != "passed" {
			return &starterReadinessError{Checks: checks}
		}
	}
	remotePrepared := false
	defer func() {
		if !remotePrepared {
			return
		}
		cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
		defer cancel()
		if err := syncStarterRemoteWorkloadEvidence(cleanupCtx, plan); err != nil {
			runErr = errors.Join(runErr, err)
		}
		if err := cleanupStarterRemoteWorkloadExecution(cleanupCtx, plan.Remote); err != nil {
			runErr = errors.Join(runErr, err)
		}
	}()
	for _, command := range plan.Commands {
		var output []byte
		var err error
		if command.Remote {
			if !remotePrepared {
				if err := prepareStarterRemoteWorkloadExecution(ctx, plan.Remote); err != nil {
					return err
				}
				remotePrepared = true
			}
			output, err = runStarterRemoteWorkloadCommand(ctx, plan.Remote, command.Script)
		} else {
			output, err = starterStatusCommandRunner(ctx, plan.Env, "sh", "-c", command.Script)
		}
		if err != nil {
			return fmt.Errorf("%s: %s", command.Name, trimmedCommandError("run command", output, err))
		}
		if command.Name == "starter-remote-core-workloads" {
			if err := runStarterActivationWaits(ctx, plan); err != nil {
				return err
			}
			if plan.Remote != nil {
				if err := syncStarterWaitOutputsToRemote(ctx, plan); err != nil {
					return err
				}
			}
		}
	}
	return nil
}

func prepareStarterRemoteWorkloadExecution(ctx context.Context, remote *starterRemoteWorkloadExecution) (resultErr error) {
	if remote == nil || remote.Selected == nil || !remote.Selected.IsRemote() {
		return cp9errors.New(cp9errors.CategoryConfig, "Starter remote workload execution requires a persisted remote EC2 host")
	}
	if _, err := cptilt.RefreshRemoteControllerInputs(
		ctx,
		remote.Selected,
		remote.LocalStateDir,
		remote.TiltStatus,
	); err != nil {
		return fmt.Errorf("refresh remote controller inputs for Starter activation: %w", err)
	}
	kubeconfig, err := os.ReadFile(remote.LocalKubeconfigPath)
	if err != nil {
		return fmt.Errorf("read Starter remote EKS kubeconfig: %w", err)
	}
	if err := cptilt.WriteRemoteFile(ctx, remote.Selected, remote.KubeconfigPath, kubeconfig, 0o600); err != nil {
		return fmt.Errorf("materialize Starter remote EKS kubeconfig on EC2: %w", err)
	}
	if err := cptilt.WriteRemoteFile(
		ctx,
		remote.Selected,
		remote.EnvironmentPath,
		starterRemoteEnvironment(remote),
		0o600,
	); err != nil {
		return fmt.Errorf("materialize Starter remote execution environment on EC2: %w", err)
	}
	defer func() {
		if resultErr == nil {
			return
		}
		cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
		defer cancel()
		resultErr = errors.Join(resultErr, cleanupStarterRemoteWorkloadExecution(cleanupCtx, remote))
	}()
	if _, err := host.RunSSH(
		ctx,
		remote.Selected,
		nil,
		"sh",
		"-lc",
		"set -eu; test -f "+host.ShellEscape(remote.ContextPath)+
			"; test -f "+host.ShellEscape(remote.KubeconfigPath)+
			"; mkdir -p "+host.ShellEscape(remote.EvidenceDir),
	); err != nil {
		return fmt.Errorf("prepare Starter remote workload execution on EC2: %w", err)
	}
	return syncStarterWaitOutputsToRemote(ctx, &starterRemoteWorkloadActivationPlan{
		Remote:   remote,
		StateDir: remote.LocalStateDir,
	})
}

func starterRemoteEnvironment(remote *starterRemoteWorkloadExecution) []byte {
	values := map[string]string{
		"CP9_AWS_RUNTIME_IDENTITY":       "ec2-instance-profile",
		"CP9_PRESERVE_IMAGE_PULL_POLICY": "1",
		"CP9_SESSION_CONTEXT":            remote.ContextPath,
	}
	for _, name := range []string{
		"CP9_STARTER_INFERENCE_SERVICE_API_KEY",
		"CP9_STARTER_AI_GATEWAY_USERNAME",
		"CP9_STARTER_AI_GATEWAY_PASSWORD",
		"CP9_STARTER_FTS_JWT_HOST",
		"CP9_STARTER_FTS_JWT_AUTH_TOKEN",
		"CP9_STARTER_FTS_JWT_ORG_ID",
		"CP9_STARTER_FTS_JWT_CLUSTER_ID",
	} {
		if value := os.Getenv(name); value != "" {
			values[name] = value
		}
	}
	var names []string
	for name := range values {
		names = append(names, name)
	}
	slices.Sort(names)
	var lines []string
	lines = append(lines, "unset AWS_PROFILE AWS_DEFAULT_PROFILE AWS_SHARED_CREDENTIALS_FILE AWS_CONFIG_FILE")
	for _, name := range names {
		lines = append(lines, "export "+name+"="+shellQuote(values[name]))
	}
	return []byte(strings.Join(lines, "\n") + "\n")
}

func runStarterRemoteWorkloadCommand(
	ctx context.Context,
	remote *starterRemoteWorkloadExecution,
	script string,
) ([]byte, error) {
	if remote == nil {
		return nil, cp9errors.New(cp9errors.CategoryConfig, "Starter remote workload execution is missing")
	}
	script = "set -eu\n. " + host.ShellEscape(remote.EnvironmentPath) + "\n" + script + "\n"
	return host.RunSSH(
		ctx,
		remote.Selected,
		[]byte(script),
		"sh",
		"-s",
	)
}

func syncStarterWaitOutputsToRemote(ctx context.Context, plan *starterRemoteWorkloadActivationPlan) error {
	if plan == nil || plan.Remote == nil {
		return nil
	}
	for _, name := range []string{
		starterDBAASProxyHostFileName,
		starterGatewayHostFileName,
		starterRegionalServiceURLFileName,
		starterOpsServiceHostFileName,
		starterAIGatewayHostFileName,
		starterAIGatewayDNSReceiptFileName,
	} {
		localPath := filepath.Join(plan.StateDir, "reports", "evidence", name)
		data, err := os.ReadFile(localPath)
		if errors.Is(err, os.ErrNotExist) {
			continue
		}
		if err != nil {
			return fmt.Errorf("read Starter activation output %s: %w", name, err)
		}
		remotePath := pathpkg.Join(plan.Remote.EvidenceDir, name)
		if err := cptilt.WriteRemoteFile(ctx, plan.Remote.Selected, remotePath, data, 0o600); err != nil {
			return fmt.Errorf("sync Starter activation output %s to EC2: %w", name, err)
		}
	}
	return nil
}

func syncStarterRemoteWorkloadEvidence(ctx context.Context, plan *starterRemoteWorkloadActivationPlan) error {
	if plan == nil || plan.Remote == nil {
		return nil
	}
	err := cptilt.SyncRemoteArtifacts(
		ctx,
		plan.Remote.Selected,
		plan.Remote.EvidenceDir,
		filepath.Join(plan.StateDir, "reports", "evidence"),
		[]string{
			starterParityContractFileName,
			starterPreApplyPlanFileName,
			starterSelectedImageSnapshotFileName,
			starterActivationWaitEvidenceFileName,
		},
	)
	if err != nil {
		return fmt.Errorf("sync Starter remote workload evidence from EC2: %w", err)
	}
	return nil
}

func cleanupStarterRemoteWorkloadExecution(ctx context.Context, remote *starterRemoteWorkloadExecution) error {
	if remote == nil {
		return nil
	}
	if _, err := host.RunSSH(ctx, remote.Selected, nil, "rm", "-f", remote.EnvironmentPath); err != nil {
		return fmt.Errorf("remove Starter remote execution environment: %w", err)
	}
	return nil
}

// starterKernelImagePinsFromAppDelivery collects the Starter kernel image inputs
// that cp9 owns and that have a single source of truth: the pooled serving TiDB
// image and the starter manager image flags.
//
// Coverage is deliberately bounded to these. The TidbCluster CR images and the
// standalone serverless workload images are spread across many files with a
// different reference shape (baseImage plus version) and are not verified here;
// see skills/get-started/references/change-kernel-images.md for the full map.
func starterKernelImagePinsFromAppDelivery(appDelivery string) ([]starterKernelImagePin, error) {
	poolPatch := filepath.Join(
		appDelivery, "tidb-pools", "components", "pool-tidb-image", "tidb-image-patch.yaml",
	)
	poolImage, err := starterYAMLMappingValue(poolPatch, "tidb-image")
	if err != nil {
		return nil, err
	}
	pins := []starterKernelImagePin{{
		Name:   "pool-tidb-image",
		Source: starterRepoRelativePath(poolPatch),
		Image:  poolImage,
	}}

	managerPatch := filepath.Join(appDelivery, "manager", "runtime-patch.yaml")
	for _, flag := range []string{
		"activate-tidb-image",
		"serverless-lightning-image",
		"dumpling-image",
	} {
		image, flagErr := starterManagerFlagValue(managerPatch, flag)
		if flagErr != nil {
			return nil, flagErr
		}
		pins = append(pins, starterKernelImagePin{
			Name:   flag,
			Source: starterRepoRelativePath(managerPatch),
			Image:  image,
		})
	}
	return pins, nil
}

// starterRepoRelativePath trims everything above personas/ so preflight messages
// carry the same repo-relative shape as the rest of cp9's recorded evidence paths.
func starterRepoRelativePath(path string) string {
	slashed := filepath.ToSlash(path)
	if idx := strings.Index(slashed, "personas/"); idx >= 0 {
		return slashed[idx:]
	}
	return slashed
}

// starterYAMLMappingValue reads a single scalar mapping value such as
// "  tidb-image: <ref>" from a small cp9-owned manifest.
func starterYAMLMappingValue(path, key string) (string, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return "", cp9errors.Newf(
			cp9errors.CategoryConfig, "read %s for Starter kernel image %s: %v", path, key, err,
		)
	}
	prefix := key + ":"
	for line := range strings.SplitSeq(string(data), "\n") {
		trimmed := strings.TrimSpace(strings.TrimSuffix(line, "\r"))
		if value, ok := strings.CutPrefix(trimmed, prefix); ok {
			value = strings.Trim(starterFlagValueWithoutComment(value), `"'`)
			if value == "" {
				return "", cp9errors.Newf(
					cp9errors.CategoryConfig, "Starter kernel image %s is empty in %s", key, path,
				)
			}
			return value, nil
		}
	}
	return "", cp9errors.Newf(
		cp9errors.CategoryConfig, "Starter kernel image %s not found in %s", key, path,
	)
}

// starterManagerFlagValue reads a "--<flag>=<value>" argument from the starter
// manager runtime patch.
func starterManagerFlagValue(path, flag string) (string, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return "", cp9errors.Newf(
			cp9errors.CategoryConfig, "read %s for Starter manager %s: %v", path, flag, err,
		)
	}
	prefix := "--" + flag + "="
	for line := range strings.SplitSeq(string(data), "\n") {
		trimmed := strings.TrimSpace(strings.TrimSuffix(line, "\r"))
		trimmed = strings.TrimPrefix(trimmed, "- ")
		trimmed = strings.Trim(strings.TrimSpace(trimmed), `"'`)
		if value, ok := strings.CutPrefix(trimmed, prefix); ok {
			value = starterFlagValueWithoutComment(value)
			if value == "" {
				return "", cp9errors.Newf(
					cp9errors.CategoryConfig, "Starter manager %s is empty in %s", flag, path,
				)
			}
			return value, nil
		}
	}
	return "", cp9errors.Newf(
		cp9errors.CategoryConfig, "Starter manager %s not found in %s", flag, path,
	)
}

// starterFlagValueWithoutComment trims a trailing YAML comment from a value read
// out of a cp9-owned manifest. These files carry comment headers, so an inline
// comment on the value line is a realistic edit that must not corrupt the value.
func starterFlagValueWithoutComment(value string) string {
	if idx := strings.Index(value, " #"); idx >= 0 {
		value = value[:idx]
	}
	return strings.TrimSpace(value)
}

// starterKernelImagePreflightChecks verifies that each cp9-owned kernel image pin
// exists in its registry before activation mutates anything.
//
// A reference that is not an ECR reference cannot be checked here, and pointing a
// kernel image at a non-ECR registry is a legitimate configuration, so those pass
// with an explicit "not verified" message instead of blocking activation.
// starterImageLooksLikeECR reports whether a reference is trying to be an ECR
// reference, so a near-miss registry host fails instead of silently skipping.
func starterImageLooksLikeECR(image string) bool {
	registry, _, found := strings.Cut(image, "/")
	if !found {
		return false
	}
	registry = strings.ToLower(registry)
	return strings.Contains(registry, "amazonaws") || strings.Contains(registry, ".ecr.") ||
		strings.Contains(registry, "dkr.")
}

func starterKernelImagePreflightChecks(
	ctx context.Context,
	plan *starterRemoteWorkloadActivationPlan,
) []starterStatusCheck {
	checks := make([]starterStatusCheck, 0, len(plan.KernelImages))
	for _, pin := range plan.KernelImages {
		check := starterStatusCheck{
			Name:    "starter-kernel-image-" + pin.Name,
			Status:  "not_ready",
			Service: pin.Name,
			Pin:     pin.Image,
		}
		ref, ok := ecrImageReferenceFromImage(pin.Image)
		if !ok {
			// A malformed ECR host is the most likely typo, so it must not pass as
			// "some other registry". Only a reference that is clearly not ECR at all
			// is allowed through unverified.
			if starterImageLooksLikeECR(pin.Image) {
				check.ReasonCode = lifecycle.StarterReasonInvalidImagePin
				check.Message = "Starter kernel image looks like an ECR reference but is malformed: " + pin.Source
				checks = append(checks, check)
				continue
			}
			check.Status = "passed"
			check.Message = "Starter kernel image is not an ECR reference; existence not verified"
			checks = append(checks, check)
			continue
		}
		profile := starterImagePullProfileForRegistry(&plan.Runtime, ref.Registry)
		exists, err := starterImageRegistryChecker(ctx, plan.Env, profile, ref)
		if err != nil {
			check.ReasonCode = lifecycle.StarterReasonImagePinUnreadable
			check.Message = "Starter kernel image could not be verified: " + err.Error()
			checks = append(checks, check)
			continue
		}
		if !exists {
			check.ReasonCode = lifecycle.StarterReasonMissingImagePin
			check.Message = "Starter kernel image does not exist: " + pin.Source
			checks = append(checks, check)
			continue
		}
		check.Status = "passed"
		check.Message = "Starter kernel image exists"
		checks = append(checks, check)
	}
	return checks
}

func starterImagePreflightChecks(ctx context.Context, plan *starterRemoteWorkloadActivationPlan) []starterStatusCheck {
	if plan == nil {
		return []starterStatusCheck{{
			Name:       "starter-image-pins",
			Status:     "not_ready",
			ReasonCode: lifecycle.StarterReasonInvalidImagePin,
			Message:    "Starter image activation plan is missing",
		}}
	}
	checks := make([]starterStatusCheck, 0, len(plan.Images))
	for _, image := range plan.Images {
		check := starterStatusCheck{
			Name:       "starter-image-pin-" + image.Service,
			Status:     "not_ready",
			Service:    image.Service,
			Pin:        image.Image,
			Provenance: image.Provenance,
			ValidUntil: image.ValidUntil,
		}
		if strings.TrimSpace(image.Image) == "" {
			check.ReasonCode = lifecycle.StarterReasonMissingImagePin
			check.Message = "selected Starter image pin is missing"
			checks = append(checks, check)
			continue
		}
		switch image.Provenance {
		case sourcebacked.ImageProvenanceDurableRelease,
			sourcebacked.ImageProvenanceLifecycleManagedEphemeral,
			starterImageSourceOwnerSnapshot,
			starterImageProvenanceReviewedOverride,
			starterImageProvenanceSourceBuild:
		default:
			check.ReasonCode = lifecycle.StarterReasonInvalidImagePin
			check.Message = "selected Starter image pin has unsupported provenance"
			checks = append(checks, check)
			continue
		}
		if image.Provenance == sourcebacked.ImageProvenanceLifecycleManagedEphemeral {
			validUntil, err := time.Parse(time.RFC3339, image.ValidUntil)
			if err != nil || !starterImagePreflightNow().Before(validUntil) {
				check.ReasonCode = lifecycle.StarterReasonExpiredImagePin
				check.Message = "selected lifecycle-managed Starter image pin is expired or missing a valid time bound"
				checks = append(checks, check)
				continue
			}
		}
		ref, ok := ecrImageReferenceFromImage(image.Image)
		if !ok {
			check.ReasonCode = lifecycle.StarterReasonInvalidImagePin
			check.Message = "selected Starter image pin uses an unsupported registry format"
			checks = append(checks, check)
			continue
		}
		profile := starterImagePullProfileForRegistry(&plan.Runtime, ref.Registry)
		exists, err := starterImageRegistryChecker(ctx, plan.Env, profile, ref)
		if err != nil {
			check.ReasonCode = lifecycle.StarterReasonImagePinUnreadable
			check.Message = "selected Starter image pin could not be verified: " + err.Error()
			checks = append(checks, check)
			continue
		}
		if !exists {
			check.ReasonCode = lifecycle.StarterReasonMissingImagePin
			check.Message = "selected Starter image pin does not exist"
			checks = append(checks, check)
			continue
		}
		check.Status = "passed"
		check.Message = "selected Starter image pin exists"
		checks = append(checks, check)
	}
	return append(checks, starterKernelImagePreflightChecks(ctx, plan)...)
}

func checkStarterImageRegistry(ctx context.Context, env []string, profile string, ref ecrImageReference) (bool, error) {
	account, region, ok := ecrRegistryParts(ref.Registry)
	if !ok {
		return false, cp9errors.New(cp9errors.CategoryConfig, "unsupported ECR registry")
	}
	args := []string{}
	if strings.TrimSpace(profile) != "" {
		args = append(args, "--profile", profile)
	}
	args = append(args,
		"ecr", "batch-get-image",
		"--region", region,
		"--registry-id", account,
		"--repository-name", ref.Repository,
		"--image-ids", ref.ImageID(),
		"--query", "length(images)",
		"--output", "text",
	)
	out, err := starterStatusCommandRunner(ctx, env, "aws", args...)
	if err != nil {
		return false, fmt.Errorf("read ECR pin: %s", trimmedCommandError("batch-get-image", out, err))
	}
	return strings.TrimSpace(string(out)) == "1", nil
}

func buildStarterRemoteWorkloadActivationPlan(
	stateDir string,
	baseline *host.Session,
	session *config.SessionConfig,
	state *cluster.State,
	remote *cpterraform.RemoteEKSOutput,
) (*starterRemoteWorkloadActivationPlan, error) {
	if session == nil {
		return nil, cp9errors.New(cp9errors.CategoryConfig, "starter activation session is required")
	}
	if len(strings.TrimSpace(session.Name)) > 30 {
		return nil, cp9errors.New(
			cp9errors.CategoryConfig,
			"Starter regional parity requires a session name of at most 30 characters so every AWS load balancer remains uniquely session-prefixed within the 32-character AWS limit",
		)
	}
	if baseline == nil {
		return nil, cp9errors.New(cp9errors.CategoryConfig, "Dev full baseline host is required")
	}
	if remote == nil {
		return nil, cp9errors.New(cp9errors.CategoryConfig, "starter remote EKS artifact is required")
	}
	repoRoot, err := sessionRepoRoot(session)
	if err != nil {
		return nil, err
	}
	outputsPath := starterTerraformOutputsPath(stateDir, session, state)
	outputs, err := loadStarterTerraformOutputs(outputsPath)
	if err != nil {
		return nil, err
	}
	runtime := starterRemoteWorkloadRuntimeFromOutputs(outputs, session)
	sessionTags, err := cpaws.DeriveSessionTags(state.Owner, session.Name)
	if err != nil {
		return nil, fmt.Errorf("derive Starter Account signing-key tags: %w", err)
	}
	selectedSnapshot, err := loadStarterRemoteSelectedImageSnapshot(stateDir, state)
	if err != nil {
		return nil, err
	}
	var ownerSnapshot *artifactsync.Snapshot
	if environment := strings.TrimSpace(session.Artifacts.Environment); environment != "" {
		ownerSnapshot, err = artifactsync.LoadSnapshot(artifactsync.OutputDir(stateDir))
		if err != nil {
			return nil, fmt.Errorf("load Starter owner artifact snapshot: %w", err)
		}
		if ownerSnapshot.Environment != environment {
			return nil, cp9errors.Newf(
				cp9errors.CategoryConfig,
				"Starter owner artifact snapshot environment %s does not match session %s",
				ownerSnapshot.Environment,
				environment,
			)
		}
	}
	images, err := starterRemoteWorkloadImagesForSnapshot(selectedSnapshot, ownerSnapshot)
	if err != nil {
		return nil, err
	}
	gatewayImage, err := starterGatewayImage(images)
	if err != nil {
		return nil, err
	}
	selectedSnapshot.ActivationImages = append([]starterRemoteWorkloadImage(nil), images...)
	kubeconfig := filepath.Join(stateDir, "remote-eks", "kubeconfig.yaml")
	kubeContext := strings.TrimSpace(remote.KubeconfigContext)
	if kubeContext == "" {
		kubeContext = strings.TrimSpace(remote.ClusterName)
	}
	contextPath, err := filepath.Abs(filepath.Join(cluster.ContextDir(stateDir), starterContextIndexFileName))
	if err != nil {
		contextPath = filepath.Join(cluster.ContextDir(stateDir), starterContextIndexFileName)
	}
	env := starterAWSExecEnv(session.Target.AWS.IAM.Profile, firstNonEmpty(remote.Region, session.Target.Region))
	env = append(env,
		"CP9_SESSION_CONTEXT="+contextPath,
		"CP9_PRESERVE_IMAGE_PULL_POLICY=1",
	)

	builder := starterRemoteWorkloadScriptBuilder{
		RepoRoot:              repoRoot,
		StateDir:              stateDir,
		Kubeconfig:            kubeconfig,
		KubeContext:           kubeContext,
		RemoteClusterName:     remote.ClusterName,
		Baseline:              baseline,
		Session:               session,
		Runtime:               runtime,
		Images:                images,
		GatewayImage:          gatewayImage,
		SelectedImageSnapshot: selectedSnapshot,
		OwnerTag:              sessionTags[cpaws.TagOwner],
	}
	mutationBuilder := &builder
	var remoteExecution *starterRemoteWorkloadExecution
	if baseline.IsRemote() {
		tiltStatus, err := cptilt.LoadStatusForState(stateDir, state)
		if err != nil {
			return nil, fmt.Errorf("load remote Tilt controller status for Starter activation: %w", err)
		}
		remoteRepoRoot, err := cptilt.RemoteControllerRepositoryRoot(tiltStatus)
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(tiltStatus.RemoteContextIndexPath) == "" {
			return nil, cp9errors.New(cp9errors.CategoryConfig, "remote Tilt controller context is missing")
		}
		if strings.TrimSpace(tiltStatus.RemoteKubeconfigPath) == "" {
			return nil, cp9errors.New(cp9errors.CategoryConfig, "remote Tilt controller kubeconfig is missing")
		}
		remoteStateDir := pathpkg.Join(tiltStatus.RemoteRoot, "starter")
		remoteExecution = &starterRemoteWorkloadExecution{
			Selected:            baseline,
			TiltStatus:          tiltStatus,
			LocalStateDir:       stateDir,
			LocalKubeconfigPath: kubeconfig,
			KubeconfigPath:      pathpkg.Join(remoteStateDir, "kubeconfig.yaml"),
			ContextPath:         tiltStatus.RemoteContextIndexPath,
			EnvironmentPath:     pathpkg.Join(remoteStateDir, "activation.env"),
			EvidenceDir:         pathpkg.Join(remoteStateDir, "reports", "evidence"),
		}
		remoteBaseline := *baseline
		remoteBaseline.KubeconfigPath = tiltStatus.RemoteKubeconfigPath
		remoteRuntime := runtime
		remoteRuntime.ImagePullProfileOverride = "-"
		remoteRuntime.UseDefaultAWSCredentials = true
		remoteBuilder := builder
		remoteBuilder.RepoRoot = remoteRepoRoot
		remoteBuilder.StateDir = remoteStateDir
		remoteBuilder.Kubeconfig = remoteExecution.KubeconfigPath
		remoteBuilder.Baseline = &remoteBaseline
		remoteBuilder.Runtime = remoteRuntime
		mutationBuilder = &remoteBuilder
	}
	runRemotely := remoteExecution != nil
	commands := []starterRemoteWorkloadCommand{
		{Name: "starter-pca-preflight", Script: builder.pcaPreflightScript()},
		{Name: "starter-selected-image-snapshot", Script: builder.selectedImageSnapshotScript()},
		{Name: "starter-remote-namespaces", Remote: runRemotely, Script: mutationBuilder.namespacesScript()},
	}
	if !config.StarterAIFTSEnabled(session) {
		commands = append(
			commands,
			starterRemoteWorkloadCommand{
				Name:   "starter-ai-fts-disabled-cleanup",
				Remote: runRemotely,
				Script: mutationBuilder.aiFTSDisabledCleanupScript(),
			},
		)
	}
	commands = append(
		commands,
		starterRemoteWorkloadCommand{Name: "starter-regional-platform", Remote: runRemotely, Script: mutationBuilder.regionalPlatformScript()},
		starterRemoteWorkloadCommand{Name: "starter-remote-runtime-prereqs", Remote: runRemotely, Script: mutationBuilder.runtimePrereqsScript()},
		starterRemoteWorkloadCommand{Name: "starter-remote-core-workloads", Remote: runRemotely, Script: mutationBuilder.coreWorkloadsScript()},
		starterRemoteWorkloadCommand{Name: "starter-account-auth-jwks", Remote: runRemotely, Script: mutationBuilder.accountAuthJWKSScript()},
		starterRemoteWorkloadCommand{
			Name:   "starter-ops-service-local-bridge",
			Remote: runRemotely,
			Script: mutationBuilder.opsServiceLocalBridgeScript(),
		},
		starterRemoteWorkloadCommand{
			Name:   "starter-dbaas-proxy-local-bridge",
			Remote: runRemotely,
			Script: mutationBuilder.dbaasProxyLocalBridgeScript(),
		},
		starterRemoteWorkloadCommand{Name: "starter-global-ops-endpoint", Remote: runRemotely, Script: mutationBuilder.globalOpsEndpointScript()},
		starterRemoteWorkloadCommand{
			Name:   "starter-sql-gateway-dns-annotation",
			Remote: runRemotely,
			Script: mutationBuilder.sqlGatewayDNSAnnotationScript(),
		},
		starterRemoteWorkloadCommand{
			Name:   "starter-privatelink-service",
			Remote: runRemotely,
			Script: mutationBuilder.privateLinkServiceScript(),
		},
	)
	if config.StarterAIFTSEnabled(session) {
		commands = append(commands, starterRemoteWorkloadCommand{
			Name:   "starter-ai-gateway-dns",
			Remote: runRemotely,
			Script: mutationBuilder.aiGatewayDNSScript(),
		})
	}
	commands = append(commands, starterRemoteWorkloadCommand{
		Name:   "starter-serverless-global-catalog",
		Remote: runRemotely,
		Script: mutationBuilder.globalCatalogScript(),
	})

	waitTargets := builder.activationWaitTargets()
	if remoteExecution != nil {
		for i := range waitTargets {
			waitTargets[i].Remote = remoteExecution.Selected
			if waitTargets[i].Kubeconfig == baseline.KubeconfigPath {
				waitTargets[i].Kubeconfig = remoteExecution.TiltStatus.RemoteKubeconfigPath
			} else {
				waitTargets[i].Kubeconfig = remoteExecution.KubeconfigPath
			}
		}
	}
	kernelImages, err := starterKernelImagePinsFromAppDelivery(
		filepath.Join(repoRoot, "personas", "app-delivery", "envs", "dev-full", "starter-regional"),
	)
	if err != nil {
		return nil, err
	}
	return &starterRemoteWorkloadActivationPlan{
		Env:                   env,
		Images:                append([]starterRemoteWorkloadImage(nil), images...),
		KernelImages:          kernelImages,
		Remote:                remoteExecution,
		Runtime:               runtime,
		SelectedImageSnapshot: selectedSnapshot,
		WaitTargets:           waitTargets,
		StateDir:              stateDir,
		Commands:              commands,
	}, nil
}

type starterActivationWaitTargetKind string

const (
	starterActivationWaitRollout     starterActivationWaitTargetKind = "rollout"
	starterActivationWaitServiceHost starterActivationWaitTargetKind = "service-host"
	starterActivationWaitServingPods starterActivationWaitTargetKind = "serving-pods"
	starterActivationWaitKeyspaces   starterActivationWaitTargetKind = "keyspaces"
	starterActivationWaitCronJob     starterActivationWaitTargetKind = "cronjob"
)

type starterActivationWaitTarget struct {
	Name        string
	Kind        starterActivationWaitTargetKind
	Remote      *host.Session
	Kubeconfig  string
	Context     string
	Namespace   string
	Resource    string
	Expected    int
	OutputPath  string
	ValuePrefix string
}

type starterActivationWaitResult struct {
	Name           string `json:"name"`
	Kind           string `json:"kind"`
	Namespace      string `json:"namespace,omitempty"`
	Resource       string `json:"resource"`
	Status         string `json:"status"`
	Message        string `json:"message,omitempty"`
	DurationMillis int64  `json:"duration_millis"`
	OutputPath     string `json:"output_path,omitempty"`
}

type starterActivationWaitEvidence struct {
	APIVersion string                        `json:"apiVersion"`
	Kind       string                        `json:"kind"`
	Deadline   string                        `json:"deadline"`
	Results    []starterActivationWaitResult `json:"results"`
}

type starterCronJob struct {
	Spec starterCronJobSpec `json:"spec"`
}

type starterCronJobSpec struct {
	Schedule string `json:"schedule"`
	Suspend  *bool  `json:"suspend"`
}

func runStarterActivationWaits(ctx context.Context, plan *starterRemoteWorkloadActivationPlan) error {
	if plan == nil || len(plan.WaitTargets) == 0 {
		return nil
	}
	deadline := starterActivationWaitDeadline
	if deadline <= 0 {
		deadline = 20 * time.Minute
	}
	waitCtx, cancel := context.WithTimeout(ctx, deadline)
	defer cancel()

	results := make([]starterActivationWaitResult, len(plan.WaitTargets))
	var wg sync.WaitGroup
	for i := range plan.WaitTargets {
		target := &plan.WaitTargets[i]
		wg.Go(func() {
			results[i] = target.wait(waitCtx, plan.Env)
		})
	}
	wg.Wait()

	if err := writeStarterActivationWaitEvidence(plan.StateDir, deadline, results); err != nil {
		return err
	}
	var pending []string
	for _, result := range results {
		if result.Status != "passed" {
			pending = append(pending, result.Name+": "+result.Message)
		}
	}
	if len(pending) > 0 {
		failure := cp9errors.Newf(
			cp9errors.CategoryResource,
			"starter activation waits not ready before %s: %s",
			deadline,
			strings.Join(pending, "; "),
		)
		if waitCtx.Err() != nil {
			return errors.Join(failure, waitCtx.Err())
		}
		return failure
	}
	return nil
}

func (t *starterActivationWaitTarget) wait(ctx context.Context, env []string) starterActivationWaitResult {
	started := time.Now()
	result := starterActivationWaitResult{
		Name:       t.Name,
		Kind:       string(t.Kind),
		Namespace:  t.Namespace,
		Resource:   t.Resource,
		Status:     "not_ready",
		OutputPath: t.OutputPath,
	}
	pollInterval := starterActivationWaitPollInterval
	if pollInterval <= 0 {
		pollInterval = 5 * time.Second
	}
	var lastMessage string
	for {
		value, passed, message := t.probe(ctx, env)
		if passed {
			if t.OutputPath != "" {
				if err := writeStarterActivationWaitValue(t.OutputPath, t.ValuePrefix+value); err != nil {
					result.Message = err.Error()
					break
				}
			}
			result.Status = "passed"
			result.Message = message
			result.DurationMillis = time.Since(started).Milliseconds()
			return result
		}
		if message != "" {
			lastMessage = message
		}
		select {
		case <-ctx.Done():
			if lastMessage == "" {
				lastMessage = ctx.Err().Error()
			}
			result.Message = lastMessage
			result.DurationMillis = time.Since(started).Milliseconds()
			return result
		case <-time.After(pollInterval):
		}
	}
	result.DurationMillis = time.Since(started).Milliseconds()
	return result
}

func (t *starterActivationWaitTarget) probe(ctx context.Context, env []string) (string, bool, string) {
	args := []string{
		"--kubeconfig", t.Kubeconfig,
		"--context", t.Context,
	}
	access := starterKubeAccess{ExecEnv: env, Remote: t.Remote}
	if strings.TrimSpace(t.Namespace) != "" {
		args = append(args, "--namespace", t.Namespace)
	}
	switch t.Kind {
	case starterActivationWaitRollout:
		args = append(args, "rollout", "status", t.Resource, "--timeout=5s")
		out, err := access.run(ctx, args...)
		if err != nil {
			return "", false, trimmedCommandError("check "+t.Resource+" rollout", out, err)
		}
		return "", true, t.Resource + " rollout is available"
	case starterActivationWaitServiceHost:
		args = append(args, "get", t.Resource, "-o", "jsonpath={.status.loadBalancer.ingress[0].hostname}{.status.loadBalancer.ingress[0].ip}")
		out, err := access.run(ctx, args...)
		value := strings.TrimSpace(string(out))
		if err != nil {
			return "", false, trimmedCommandError("read "+t.Resource+" LoadBalancer host/IP", out, err)
		}
		if value == "" {
			return "", false, t.Resource + " has no LoadBalancer host/IP"
		}
		return value, true, t.Resource + " LoadBalancer host/IP is " + value
	case starterActivationWaitServingPods:
		args = append(args,
			"get", "pods",
			"-l", "app.kubernetes.io/component=serverless-tidb",
			"-o", "json",
		)
		out, err := access.run(ctx, args...)
		if err != nil {
			return "", false, trimmedCommandError("read selected serving-lane Pods", out, err)
		}
		var pods starterPodList
		if err := json.Unmarshal(out, &pods); err != nil {
			return "", false, "parse selected serving-lane Pods: " + err.Error()
		}
		ready := 0
		for _, pod := range pods.Items {
			for _, condition := range pod.Status.Conditions {
				if condition.Type == "Ready" && condition.Status == "True" {
					ready++
					break
				}
			}
		}
		message := fmt.Sprintf("selected serving-lane Pods live=%d ready=%d expected=%d", len(pods.Items), ready, t.Expected)
		if len(pods.Items) < t.Expected || ready < t.Expected {
			return "", false, message
		}
		return "", true, message
	case starterActivationWaitKeyspaces:
		args = append(args, "get", t.Resource, "-o", "json")
		out, err := access.run(ctx, args...)
		if err != nil {
			return "", false, trimmedCommandError("read "+t.Namespace+" Keyspaces", out, err)
		}
		var keyspaces starterKeyspaceList
		if err := json.Unmarshal(out, &keyspaces); err != nil {
			return "", false, "parse " + t.Namespace + " Keyspaces: " + err.Error()
		}
		available := 0
		for _, keyspace := range keyspaces.Items {
			if keyspace.Status.Phase == "Available" {
				available++
			}
		}
		message := fmt.Sprintf("Keyspaces available=%d total=%d expected=%d", available, len(keyspaces.Items), t.Expected)
		if available < t.Expected {
			return "", false, message
		}
		return "", true, message
	case starterActivationWaitCronJob:
		args = append(args, "get", t.Resource, "-o", "json")
		out, err := access.run(ctx, args...)
		if err != nil {
			return "", false, trimmedCommandError("read "+t.Resource, out, err)
		}
		var cronJob starterCronJob
		if err := json.Unmarshal(out, &cronJob); err != nil {
			return "", false, "parse " + t.Resource + ": " + err.Error()
		}
		if strings.TrimSpace(cronJob.Spec.Schedule) == "" {
			return "", false, t.Resource + " has no schedule"
		}
		if cronJob.Spec.Suspend != nil && *cronJob.Spec.Suspend {
			return "", false, t.Resource + " is suspended"
		}
		return "", true, t.Resource + " is scheduled and enabled"
	default:
		return "", false, "unknown wait target kind " + string(t.Kind)
	}
}

func writeStarterActivationWaitValue(path, value string) error {
	if strings.TrimSpace(path) == "" {
		return nil
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return err
	}
	return os.WriteFile(path, []byte(strings.TrimSpace(value)+"\n"), 0o600)
}

func writeStarterActivationWaitEvidence(stateDir string, deadline time.Duration, results []starterActivationWaitResult) error {
	if strings.TrimSpace(stateDir) == "" {
		return nil
	}
	path := starterActivationWaitEvidencePath(stateDir)
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return err
	}
	data, err := json.MarshalIndent(starterActivationWaitEvidence{
		APIVersion: starterActivationWaitAPIVersion,
		Kind:       starterActivationWaitKind,
		Deadline:   deadline.String(),
		Results:    results,
	}, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(path, data, 0o600)
}

func starterTerraformOutputsPath(stateDir string, session *config.SessionConfig, state *cluster.State) string {
	paths := cpterraform.WorkspacePathsForVarFile(stateDir, session.Terraform.VarFile)
	if state != nil && state.Paths != nil {
		if strings.TrimSpace(state.Paths.TerraformOutputs) != "" {
			return state.Paths.TerraformOutputs
		}
	}
	return paths.OutputsPath
}

func loadStarterTerraformOutputs(outputsPath string) (map[string]cpterraform.OutputValue, error) {
	data, err := os.ReadFile(outputsPath)
	if err != nil {
		return nil, fmt.Errorf("read terraform outputs: %w", err)
	}
	outputs, _, err := cpterraform.ParseOutputs(data)
	if err != nil {
		return nil, err
	}
	return outputs, nil
}

func starterRemoteWorkloadRuntimeFromOutputs(outputs map[string]cpterraform.OutputValue, session *config.SessionConfig) starterRemoteWorkloadRuntime {
	targetProfile := ""
	targetRegion := starterImagePullDefaultRegion
	if session != nil {
		targetProfile = strings.TrimSpace(session.Target.AWS.IAM.Profile)
		targetRegion = firstNonEmpty(session.Target.Region, targetRegion)
	}
	runtime := starterRemoteWorkloadRuntime{
		AWSProfile:         firstNonEmpty(targetProfile, "cp9"),
		AWSRegion:          targetRegion,
		SourceECRAccountID: starterImagePullDefaultAccountID,
		SourceECRProfile:   starterImagePullDefaultProfile,
		SourceECRRegion:    starterImagePullDefaultRegion,
		StorageRegion:      targetRegion,
	}
	if runtimePatch := outputObject(outputs, "runtime_patch_inputs"); runtimePatch != nil {
		if aws := objectField(runtimePatch, "aws"); aws != nil {
			runtime.AWSAccountID = stringField(aws, "account_id")
			runtime.AWSProfile = firstNonEmpty(stringField(aws, "profile"), runtime.AWSProfile)
			runtime.AWSRegion = firstNonEmpty(stringField(aws, "region"), runtime.AWSRegion)
			runtime.SourceECRAccountID = firstNonEmpty(stringField(aws, "source_ecr_account_id"), runtime.SourceECRAccountID)
			runtime.SourceECRProfile = firstNonEmpty(stringField(aws, "source_ecr_profile"), runtime.SourceECRProfile)
			runtime.SourceECRRegion = firstNonEmpty(stringField(aws, "source_ecr_region"), runtime.SourceECRRegion)
		}
	}
	if storage := outputObject(outputs, "starter_serverless_storage"); storage != nil {
		runtime.StorageBucket = stringField(storage, "bucket")
		runtime.StoragePrefix = stringField(storage, "prefix")
		runtime.StorageRegion = firstNonEmpty(stringField(storage, "region"), runtime.StorageRegion)
	}
	if pca := outputObject(outputs, "session_product_pca"); pca != nil {
		runtime.ComponentPCAARN = stringField(pca, "arn")
		runtime.ComponentPCARegion = stringField(pca, "region")
	}
	if pcaRegion, _, ok := acmPCAARNParts(runtime.ComponentPCAARN); ok {
		runtime.ComponentPCARegion = firstNonEmpty(runtime.ComponentPCARegion, pcaRegion)
	}
	if runtime.StorageBucket == "" {
		runtime.StorageBucket = "cp9-" + session.Name + "-starter-cse"
	}
	if runtime.StoragePrefix == "" {
		runtime.StoragePrefix = "starter/" + session.Name + "/tidb-serverless"
	}
	runtime.ImagePullProfileOverride = strings.TrimSpace(os.Getenv("CP9_STARTER_IMAGE_PULL_PROFILE"))
	return runtime
}

func starterRemoteWorkloadImagesForSnapshot(
	snapshot *starterRemoteSelectedImageSnapshot,
	ownerSnapshot *artifactsync.Snapshot,
) ([]starterRemoteWorkloadImage, error) {
	contracts, err := starterRemoteSourceBackedContracts()
	if err != nil {
		return nil, err
	}
	sourceImages := map[string]string{}
	if snapshot != nil {
		for _, selected := range snapshot.SelectedSourceImages {
			sourceImages[selected.Service] = selected.Image
		}
	}
	ownerImages := map[string]string{}
	if ownerSnapshot != nil {
		for i := range ownerSnapshot.Artifacts {
			artifact := &ownerSnapshot.Artifacts[i]
			if artifact.Kind == "service" {
				ownerImages[artifact.Name] = strings.TrimSpace(artifact.Image)
			}
		}
	}
	var images []starterRemoteWorkloadImage
	for i := range contracts {
		contract := &contracts[i]
		envKey := starterRemoteImageEnvKey(contract.Name)
		if image := strings.TrimSpace(os.Getenv(envKey)); image != "" {
			images = append(
				images,
				starterRemoteWorkloadImage{
					Service: contract.Name, Image: image, Source: starterRemoteWorkloadImageSourceEnvOverride,
					Provenance: starterImageProvenanceReviewedOverride,
				},
			)
			continue
		}
		if image := strings.TrimSpace(sourceImages[contract.Name]); image != "" {
			images = append(images, starterRemoteWorkloadImage{
				Service: contract.Name, Image: image, Source: starterRemoteWorkloadImageSourceSummary,
				Provenance: starterImageProvenanceSourceBuild,
			})
			continue
		}
		if ownerSnapshot != nil {
			image := ownerImages[contract.Name]
			if image == "" {
				return nil, cp9errors.Newf(
					cp9errors.CategoryConfig,
					"Starter owner artifact snapshot has no image for %s",
					contract.Name,
				)
			}
			images = append(images, starterRemoteWorkloadImage{
				Service: contract.Name, Image: image, Source: starterImageSourceOwnerSnapshot,
				Provenance: starterImageSourceOwnerSnapshot,
			})
			continue
		}
		if strings.TrimSpace(contract.StagingImage) != "" {
			images = append(
				images,
				starterRemoteWorkloadImage{
					Service:    contract.Name,
					Image:      strings.TrimSpace(contract.StagingImage),
					Source:     starterRemoteWorkloadImageSourceStaging,
					Provenance: contract.StagingImageProvenance,
					ValidUntil: contract.StagingImageValidUntil,
				},
			)
			continue
		}
		return nil, cp9errors.Newf(
			cp9errors.CategoryConfig,
			"starter remote workload image %s is required; set %s to a staging/pinned image",
			contract.Name,
			envKey,
		)
	}
	if ownerSnapshot != nil {
		image := ownerImages[starterO11yAgentControlServerArtifact]
		if image == "" {
			return nil, cp9errors.Newf(
				cp9errors.CategoryConfig,
				"Starter owner artifact snapshot has no image for %s",
				starterO11yAgentControlServerArtifact,
			)
		}
		images = append(images, starterRemoteWorkloadImage{
			Service: starterO11yAgentControlServerArtifact, Image: image, Source: starterImageSourceOwnerSnapshot,
			Provenance: starterImageSourceOwnerSnapshot,
		})
	}
	return images, nil
}

func loadStarterRemoteSelectedImageSnapshot(stateDir string, state *cluster.State) (*starterRemoteSelectedImageSnapshot, error) {
	path := proof.ImageSourceSummaryPath(stateDir, state)
	snapshot := &starterRemoteSelectedImageSnapshot{
		APIVersion:        starterSelectedImageSnapshotAPIVersion,
		Kind:              starterSelectedImageSnapshotKind,
		SourceSummaryPath: path,
	}
	imageSource, err := proof.LoadImageSourceSummary(path)
	if err != nil {
		if errors.Is(err, proof.ErrImageSourceSummaryNotFound) {
			return snapshot, nil
		}
		return nil, fmt.Errorf("load Starter remote image source summary: %w", err)
	}
	if imageSource == nil {
		return snapshot, nil
	}
	snapshot.SourceSummaryLoaded = true
	sourceBuilt := map[string]bool{}
	for _, service := range imageSource.SourceBuildServices {
		service = strings.TrimSpace(service)
		if service == "" {
			continue
		}
		sourceBuilt[service] = true
		snapshot.SourceBuildServices = append(snapshot.SourceBuildServices, service)
	}
	contracts, err := starterRemoteSourceBackedContracts()
	if err != nil {
		return nil, err
	}
	for i := range contracts {
		contract := &contracts[i]
		service := contract.Name
		if !sourceBuilt[service] {
			continue
		}
		if err := validateStarterRemoteSelectedImageContract(contract); err != nil {
			return nil, err
		}
		image := strings.TrimSpace(imageSource.Images[service])
		if image == "" {
			if classification, ok := imageSource.ServiceClassifications[service]; ok {
				image = strings.TrimSpace(classification.SelectedImage)
			}
		}
		if image == "" {
			return nil, cp9errors.Newf(
				cp9errors.CategoryConfig,
				"image source summary marks Starter remote service %s as source-built but has no image",
				service,
			)
		}
		snapshot.SelectedSourceImages = append(snapshot.SelectedSourceImages, starterRemoteSelectedImage{
			Service:    service,
			Image:      image,
			Kind:       strings.TrimSpace(contract.RolloutTarget.Kind),
			Namespace:  strings.TrimSpace(contract.RolloutTarget.Namespace),
			Deployment: strings.TrimSpace(contract.RolloutTarget.Name),
			Container:  strings.TrimSpace(contract.RolloutTarget.Container),
		})
	}
	return snapshot, nil
}

func validateStarterRemoteSelectedImageContract(contract *sourcebacked.Contract) error {
	if contract == nil {
		return cp9errors.New(cp9errors.CategoryConfig, "selected Starter remote service lacks workload replacement data: contract is nil")
	}
	kind := strings.TrimSpace(contract.RolloutTarget.Kind)
	if kind != starterRemoteSelectedImageDeploymentKind &&
		kind != starterRemoteSelectedImageStatefulSetKind &&
		kind != starterRemoteSelectedImageDaemonSetKind {
		return cp9errors.Newf(
			cp9errors.CategoryConfig,
			"selected Starter remote service %s lacks workload replacement data: rollout target kind %q is unsupported",
			contract.Name,
			contract.RolloutTarget.Kind,
		)
	}
	if strings.TrimSpace(contract.RolloutTarget.Namespace) == "" ||
		strings.TrimSpace(contract.RolloutTarget.Name) == "" ||
		strings.TrimSpace(contract.RolloutTarget.Container) == "" {
		return cp9errors.Newf(
			cp9errors.CategoryConfig,
			"selected Starter remote service %s lacks workload replacement data: namespace, deployment, and container are required",
			contract.Name,
		)
	}
	return nil
}

func starterRemoteSourceBackedContracts() ([]sourcebacked.Contract, error) {
	registry, err := sourcebacked.DefaultRegistry()
	if err != nil {
		return nil, fmt.Errorf("load Starter remote source-backed contracts: %w", err)
	}
	contracts := registry.RealEKSDataPlaneContracts()
	if len(contracts) == 0 {
		return nil, cp9errors.New(cp9errors.CategoryConfig, "source-backed registry has no Starter remote-EKS contracts")
	}
	for i := range contracts {
		contract := &contracts[i]
		if !contract.SourceBuildSupported {
			return nil, cp9errors.Newf(
				cp9errors.CategoryConfig,
				"Starter remote-EKS source-backed contract %s is not supported: %s",
				contract.Name,
				contract.SupportStatus,
			)
		}
		if strings.TrimSpace(contract.StagingImage) == "" {
			return nil, cp9errors.Newf(cp9errors.CategoryConfig, "Starter remote-EKS contract %s has no staging image", contract.Name)
		}
	}
	return contracts, nil
}

func starterRemoteImageEnvKey(service string) string {
	switch service {
	case "manager":
		return "CP9_STARTER_MANAGER_IMAGE"
	case "serverless-regional-service":
		return "CP9_STARTER_SERVERLESS_REGIONAL_SERVICE_IMAGE"
	case "tidb-gateway-inner":
		return "CP9_STARTER_TIDB_GATEWAY_INNER_IMAGE"
	case "tidb-worker-scaler":
		return "CP9_STARTER_TIDB_WORKER_SCALER_IMAGE"
	default:
		return "CP9_STARTER_" + strings.ToUpper(strings.NewReplacer("-", "_").Replace(service)) + "_IMAGE"
	}
}

type starterRemoteWorkloadScriptBuilder struct {
	RepoRoot              string
	StateDir              string
	Kubeconfig            string
	KubeContext           string
	RemoteClusterName     string
	Baseline              *host.Session
	Session               *config.SessionConfig
	Runtime               starterRemoteWorkloadRuntime
	Images                []starterRemoteWorkloadImage
	GatewayImage          ecrImageReference
	SelectedImageSnapshot *starterRemoteSelectedImageSnapshot
	OwnerTag              string
}

type starterPCACommandArgs struct {
	issue []string
	get   []string
	getCA []string
}

func (b *starterRemoteWorkloadScriptBuilder) pcaCommandArgs() (starterPCACommandArgs, bool) {
	pcaARN := strings.TrimSpace(b.Runtime.ComponentPCAARN)
	pcaRegion, _, ok := acmPCAARNParts(pcaARN)
	if !ok {
		return starterPCACommandArgs{}, false
	}
	pcaRegion = firstNonEmpty(b.Runtime.ComponentPCARegion, pcaRegion, b.Runtime.AWSRegion)
	awsArgs := []string{}
	if profile := b.pcaProfileForARN(pcaARN); profile != "" {
		awsArgs = append(awsArgs, "--profile", profile)
	}
	prefix := append(slices.Clone(awsArgs), "acm-pca")
	return starterPCACommandArgs{
		issue: append(slices.Clone(prefix),
			"issue-certificate",
			"--region", pcaRegion,
			"--certificate-authority-arn", pcaARN,
		),
		get: append(slices.Clone(prefix),
			"get-certificate",
			"--region", pcaRegion,
			"--certificate-authority-arn", pcaARN,
		),
		getCA: append(slices.Clone(prefix),
			"get-certificate-authority-certificate",
			"--region", pcaRegion,
			"--certificate-authority-arn", pcaARN,
			"--query", "Certificate", "--output", "text",
		),
	}, true
}

func starterSelectedImageSnapshotPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterSelectedImageSnapshotFileName)
}

func starterStatusEvidencePath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterStatusEvidenceFileName)
}

func starterActivationWaitEvidencePath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterActivationWaitEvidenceFileName)
}

func starterDBAASProxyHostPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterDBAASProxyHostFileName)
}

func starterGatewayHostPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterGatewayHostFileName)
}

func starterRegionalServiceURLPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterRegionalServiceURLFileName)
}

func starterOpsServiceHostPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterOpsServiceHostFileName)
}

func starterAIGatewayHostPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterAIGatewayHostFileName)
}

func starterAIGatewayDNSReceiptPath(stateDir string) string {
	return filepath.Join(stateDir, "reports", "evidence", starterAIGatewayDNSReceiptFileName)
}

func (b *starterRemoteWorkloadScriptBuilder) activationWaitTargets() []starterActivationWaitTarget {
	var targets []starterActivationWaitTarget
	if b.Baseline != nil {
		targets = append(targets,
			starterActivationWaitTarget{
				Name:       "starter-premium-metadb-rollout",
				Kind:       starterActivationWaitRollout,
				Kubeconfig: b.Baseline.KubeconfigPath,
				Context:    b.Baseline.Context,
				Namespace:  "cp9-system",
				Resource:   "deployment/premium-metadb",
			},
			starterActivationWaitTarget{
				Name:       "starter-global-service-rollout",
				Kind:       starterActivationWaitRollout,
				Kubeconfig: b.Baseline.KubeconfigPath,
				Context:    b.Baseline.Context,
				Namespace:  "cp9-global",
				Resource:   "deployment/serverless-global-service",
			},
		)
	}
	for _, rollout := range starterRemoteWorkloadRollouts(b.Session) {
		targets = append(targets, starterActivationWaitTarget{
			Name:       rollout.checkName,
			Kind:       starterActivationWaitRollout,
			Kubeconfig: b.Kubeconfig,
			Context:    b.KubeContext,
			Namespace:  rollout.namespace,
			Resource:   rollout.resource,
		})
	}
	targets = append(
		targets,
		starterActivationWaitTarget{
			Name:       "starter-selected-serving-lane-pods",
			Kind:       starterActivationWaitServingPods,
			Kubeconfig: b.Kubeconfig,
			Context:    b.KubeContext,
			Namespace:  starterSelectedServingLane,
			Resource:   "pods/app.kubernetes.io/component=serverless-tidb",
			Expected:   config.ResolvedStarterTiDBPoolPods(b.Session),
		},
		starterActivationWaitTarget{
			Name:       "starter-keyspace-pool-available",
			Kind:       starterActivationWaitKeyspaces,
			Kubeconfig: b.Kubeconfig,
			Context:    b.KubeContext,
			Namespace:  "keyspace-pool",
			Resource:   "keyspaces.serverless.pingcap.com",
			Expected:   1,
		},
	)
	if starterObservabilityEnabled(b.Session) {
		for _, target := range []struct {
			name     string
			resource string
		}{
			{name: "starter-observability-prometheus", resource: "deployment/prometheus"},
			{name: "starter-observability-grafana", resource: "deployment/grafana"},
			{name: "starter-observability-kube-state-metrics", resource: "deployment/kube-state-metrics"},
			{name: "starter-observability-node-exporter", resource: "daemonset/node-exporter"},
		} {
			targets = append(targets, starterActivationWaitTarget{
				Name:       target.name,
				Kind:       starterActivationWaitRollout,
				Kubeconfig: b.Kubeconfig,
				Context:    b.KubeContext,
				Namespace:  "monitoring",
				Resource:   target.resource,
			})
		}
	}
	targets = append(targets,
		starterActivationWaitTarget{
			Name:        "starter-dbaas-proxy-host",
			Kind:        starterActivationWaitServiceHost,
			Kubeconfig:  b.Kubeconfig,
			Context:     b.KubeContext,
			Namespace:   starterRemoteRegionalNamespace,
			Resource:    "service/dbaas-proxy",
			OutputPath:  starterDBAASProxyHostPath(b.StateDir),
			ValuePrefix: "",
		},
		starterActivationWaitTarget{
			Name:        "starter-tidb-gateway-host",
			Kind:        starterActivationWaitServiceHost,
			Kubeconfig:  b.Kubeconfig,
			Context:     b.KubeContext,
			Namespace:   starterRemoteRegionalNamespace,
			Resource:    "service/tidb-gateway-new",
			OutputPath:  starterGatewayHostPath(b.StateDir),
			ValuePrefix: "",
		},
		starterActivationWaitTarget{
			Name:        "starter-regional-service-endpoint",
			Kind:        starterActivationWaitServiceHost,
			Kubeconfig:  b.Kubeconfig,
			Context:     b.KubeContext,
			Namespace:   starterRemoteRegionalNamespace,
			Resource:    "service/serverless-regional-service",
			OutputPath:  starterRegionalServiceURLPath(b.StateDir),
			ValuePrefix: "http://",
		},
		starterActivationWaitTarget{
			Name:        "starter-ops-service-host",
			Kind:        starterActivationWaitServiceHost,
			Kubeconfig:  b.Kubeconfig,
			Context:     b.KubeContext,
			Namespace:   starterRemoteRegionalNamespace,
			Resource:    "service/ops-service-svc",
			OutputPath:  starterOpsServiceHostPath(b.StateDir),
			ValuePrefix: "",
		},
	)
	if config.StarterAIFTSEnabled(b.Session) {
		targets = append(targets,
			starterAIFTSCronJobTarget(b.Kubeconfig, b.KubeContext),
			starterActivationWaitTarget{
				Name:        "starter-ai-gateway-host",
				Kind:        starterActivationWaitServiceHost,
				Kubeconfig:  b.Kubeconfig,
				Context:     b.KubeContext,
				Namespace:   starterRemoteRegionalNamespace,
				Resource:    "service/ai-gateway",
				OutputPath:  starterAIGatewayHostPath(b.StateDir),
				ValuePrefix: "",
			},
		)
	}
	return targets
}

func writeStarterStatusEvidence(result *starterStatusResult) error {
	if result == nil || strings.TrimSpace(result.StateDir) == "" {
		return nil
	}
	path := starterStatusEvidencePath(result.StateDir)
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return fmt.Errorf("create Starter status evidence dir: %w", err)
	}
	data, err := json.MarshalIndent(starterStatusEvidence{
		APIVersion:     starterStatusEvidenceAPIVersion,
		Kind:           starterStatusEvidenceKind,
		ArtifactSHA256: starterEvidenceArtifactDigests(result.StateDir),
		Result:         result,
	}, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal Starter status evidence: %w", err)
	}
	data = append(data, '\n')
	if err := os.WriteFile(path, data, 0o600); err != nil {
		return fmt.Errorf("write Starter status evidence: %w", err)
	}
	return nil
}

func starterEvidenceArtifactDigests(stateDir string) map[string]string {
	paths := map[string]string{
		starterParityContractFileName:         filepath.Join(stateDir, "reports", "evidence", starterParityContractFileName),
		starterPreApplyPlanFileName:           filepath.Join(stateDir, "reports", "evidence", starterPreApplyPlanFileName),
		starterSelectedImageSnapshotFileName:  starterSelectedImageSnapshotPath(stateDir),
		starterActivationWaitEvidenceFileName: filepath.Join(stateDir, "reports", "evidence", starterActivationWaitEvidenceFileName),
	}
	digests := map[string]string{}
	for name, path := range paths {
		if digest, err := sha256File(path); err == nil {
			digests[name] = digest
		}
	}
	return digests
}

func sha256File(path string) (string, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return "", err
	}
	digest := sha256.Sum256(data)
	return fmt.Sprintf("%x", digest), nil
}

func (b *starterRemoteWorkloadScriptBuilder) selectedImageSnapshotScript() string {
	snapshot := starterRemoteSelectedImageSnapshot{
		APIVersion:       starterSelectedImageSnapshotAPIVersion,
		Kind:             starterSelectedImageSnapshotKind,
		ActivationImages: append([]starterRemoteWorkloadImage(nil), b.Images...),
	}
	if b.SelectedImageSnapshot != nil {
		snapshot.SourceSummaryPath = b.SelectedImageSnapshot.SourceSummaryPath
		snapshot.SourceSummaryLoaded = b.SelectedImageSnapshot.SourceSummaryLoaded
		snapshot.SourceBuildServices = append([]string(nil), b.SelectedImageSnapshot.SourceBuildServices...)
		snapshot.SelectedSourceImages = append([]starterRemoteSelectedImage(nil), b.SelectedImageSnapshot.SelectedSourceImages...)
	}
	data, _ := json.MarshalIndent(snapshot, "", "  ")
	path := starterSelectedImageSnapshotPath(b.StateDir)
	return "mkdir -p " + shellQuote(
		filepath.Dir(path),
	) + " && cat > " + shellQuote(
		path,
	) + " <<'CP9_SELECTED_IMAGES'\n" + string(
		data,
	) + "\nCP9_SELECTED_IMAGES"
}

func (b *starterRemoteWorkloadScriptBuilder) namespacesScript() string {
	kubectl := b.remoteKubectl("")
	var commands []string
	for _, namespace := range starterRemoteWorkloadNamespaces(b.Session) {
		commands = append(commands, kubectl+" create namespace "+shellQuote(namespace)+" --dry-run=client -o yaml | "+kubectl+" apply -f -")
	}
	return shellJoin(commands)
}

func validateStarterAIFTSInputs(session *config.SessionConfig) error {
	if !config.StarterAIFTSEnabled(session) {
		return nil
	}
	var missing []string
	if strings.TrimSpace(session.Host.Exposure.ZoneName) == "" {
		missing = append(missing, "host.exposure.zone_name")
	}
	if strings.TrimSpace(session.Host.Exposure.ZoneID) == "" {
		missing = append(missing, "host.exposure.zone_id")
	}
	if strings.TrimSpace(session.Host.Exposure.CertificateARN) == "" {
		missing = append(missing, "host.exposure.certificate_arn")
	}
	for _, name := range []string{
		"CP9_STARTER_INFERENCE_SERVICE_API_KEY",
		"CP9_STARTER_AI_GATEWAY_USERNAME",
		"CP9_STARTER_AI_GATEWAY_PASSWORD",
		"CP9_STARTER_FTS_JWT_HOST",
		"CP9_STARTER_FTS_JWT_AUTH_TOKEN",
		"CP9_STARTER_FTS_JWT_ORG_ID",
		"CP9_STARTER_FTS_JWT_CLUSTER_ID",
	} {
		if strings.TrimSpace(os.Getenv(name)) == "" {
			missing = append(missing, name)
		}
	}
	if len(missing) > 0 {
		return cp9errors.Newf(
			cp9errors.CategoryConfig,
			"starter.ai_fts.enabled requires session-safe AI/FTS inputs before cloud mutation: %s",
			strings.Join(missing, ", "),
		)
	}
	return nil
}

func (b *starterRemoteWorkloadScriptBuilder) aiGatewayPublicHost() string {
	if b == nil || b.Session == nil {
		return ""
	}
	zoneName := strings.TrimSuffix(strings.TrimSpace(b.Session.Host.Exposure.ZoneName), ".")
	if zoneName == "" || strings.TrimSpace(b.Session.Name) == "" {
		return ""
	}
	return strings.TrimSpace(b.Session.Name) + "-qdrant." + zoneName
}

func (b *starterRemoteWorkloadScriptBuilder) regionalPlatformScript() string {
	script := filepath.Join(b.RepoRoot, "personas", "dev-full", "starter", "install-regional-platform.sh")
	return shellCommand(script, b.Kubeconfig, b.KubeContext, b.RemoteClusterName, b.Runtime.AWSRegion)
}

func (b *starterRemoteWorkloadScriptBuilder) pcaPreflightScript() string {
	sourceNS := "cp9-global"
	sourceKubectl := b.baselineKubectl(sourceNS)
	sourceProfile := firstNonEmpty(b.Runtime.SourceECRProfile, starterImagePullDefaultProfile)
	sourceAccountID := firstNonEmpty(b.Runtime.SourceECRAccountID, starterImagePullDefaultAccountID)
	targetProfile := firstNonEmpty(b.Runtime.AWSProfile, "cp9")
	targetAccountID := strings.TrimSpace(b.Runtime.AWSAccountID)
	var commands []string
	checks := []struct {
		service    string
		configMaps []string
	}{
		{service: "serverless-service", configMaps: []string{"serverless-service"}},
		{service: "serverless-global-service", configMaps: []string{"serverless-global-service-config", "serverless-global-service"}},
	}
	for _, check := range checks {
		quotedConfigMaps := make([]string, 0, len(check.configMaps))
		for _, configMap := range check.configMaps {
			quotedConfigMaps = append(quotedConfigMaps, shellQuote(configMap))
		}
		placeholderMessage := fmt.Sprintf(
			"starter PCA preflight failed: %s ConfigMap has missing or placeholder pcaArn; use a reachable ACM PCA ARN or a session-owned PCA before running Starter OpenAPI",
			check.service,
		)
		accessMessage := fmt.Sprintf(
			"starter PCA preflight failed: cannot describe %s pcaArn with selected AWS profile; grant ACM PCA read/issue permissions or use a session-owned PCA before running Starter OpenAPI",
			check.service,
		)
		getConfig := "config_yaml=''; for configmap in " + strings.Join(quotedConfigMaps, " ") + "; do " +
			"if config_yaml=$(" + sourceKubectl + " get configmap \"$configmap\" -o " + shellQuote(`jsonpath={.data.config\.yaml}`) + " 2>/dev/null) && [ -n \"$config_yaml\" ]; then break; fi; " +
			"done; [ -n \"$config_yaml\" ] || { echo " + shellQuote(placeholderMessage) + " >&2; exit 1; }"
		extractPCAARN := "printf '%s' \"$config_yaml\"" +
			" | sed -n 's/^[[:space:]]*pcaArn:[[:space:]]*//p'" +
			" | head -n1" +
			" | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//' -e 's/^\"//' -e 's/\"$//' -e \"s/^'//\" -e \"s/'$//\""
		command := getConfig + " && " +
			"pca_arn=$(" + extractPCAARN + ") && " +
			"case \"$pca_arn\" in arn:aws:acm-pca:*:*:certificate-authority/*) ;; *) echo " + shellQuote(placeholderMessage) + " >&2; exit 1;; esac && " +
			"case \"$pca_arn\" in */cp9-*|*/cp9-local) echo " + shellQuote(placeholderMessage) + " >&2; exit 1;; esac && " +
			"pca_region=$(printf '%s' \"$pca_arn\" | awk -F: '{print $4}') && " +
			"pca_account=$(printf '%s' \"$pca_arn\" | awk -F: '{print $5}') && " +
			"pca_profile=" + shellQuote(sourceProfile) + "; " +
			"case \"$pca_account\" in " + shellQuote(targetAccountID) + ") pca_profile=" + shellQuote(targetProfile) + ";; " + shellQuote(sourceAccountID) + ") pca_profile=" + shellQuote(sourceProfile) + ";; esac; " +
			"pca_status=$(aws --profile \"$pca_profile\" acm-pca describe-certificate-authority --region \"$pca_region\" --certificate-authority-arn \"$pca_arn\" --query 'CertificateAuthority.Status' --output text) && " +
			"[ \"$pca_status\" = 'ACTIVE' ] || { echo " + shellQuote(accessMessage) + " >&2; exit 1; }"
		commands = append(commands, command)
	}
	return shellJoin(commands)
}

func (b *starterRemoteWorkloadScriptBuilder) runtimePrereqsScript() string {
	pullSecretScript := filepath.Join(b.RepoRoot, "personas", "dev-full", "runtime", "scripts", "apply-staging-image-pull-secret.sh")
	var pullSecretCommands []string
	for _, registry := range b.ecrRegistries() {
		imagePullProfile := firstNonEmpty(b.imagePullProfileForRegistry(registry), "-")
		for _, namespace := range starterRemoteWorkloadNamespaces(b.Session) {
			pullSecretCommands = append(
				pullSecretCommands,
				shellCommand(
					pullSecretScript,
					namespace,
					b.Kubeconfig,
					b.KubeContext,
					imagePullProfile,
					b.imagePullSecretForRegistry(registry),
					registry,
				),
			)
		}
	}
	regionalInfraKubectl := b.remoteKubectl("regional-infra")
	remoteRegionalKubectl := b.remoteKubectl(starterRemoteRegionalNamespace)
	regionalRedisCert := "tmpdir=$(mktemp -d); trap 'rm -rf \"$tmpdir\"' EXIT; " +
		"if " + regionalInfraKubectl + " get secret regional-redis-tls >/dev/null 2>&1; then " +
		regionalInfraKubectl + " get secret regional-redis-tls -o jsonpath='{.data.tls\\.crt}' | base64 -d > \"$tmpdir/tls.crt\"; " +
		regionalInfraKubectl + " get secret regional-redis-tls -o jsonpath='{.data.tls\\.key}' | base64 -d > \"$tmpdir/tls.key\"; " +
		"ca=$(" + regionalInfraKubectl + " get secret regional-redis-tls -o jsonpath='{.data.ca\\.crt}' 2>/dev/null || true); " +
		"if [ -n \"$ca\" ]; then printf '%s' \"$ca\" | base64 -d > \"$tmpdir/ca.crt\"; else cp \"$tmpdir/tls.crt\" \"$tmpdir/ca.crt\"; fi; " +
		"else " +
		"openssl req -x509 -newkey rsa:2048 -sha256 -nodes -days 7 -subj /CN=regional-redis -addext basicConstraints=critical,CA:TRUE -addext keyUsage=critical,keyCertSign,digitalSignature,keyEncipherment -addext subjectAltName=DNS:regional-redis,DNS:regional-redis.regional-infra,DNS:regional-redis.regional-infra.svc -keyout \"$tmpdir/tls.key\" -out \"$tmpdir/tls.crt\" >/dev/null 2>&1; " +
		"cp \"$tmpdir/tls.crt\" \"$tmpdir/ca.crt\"; " +
		regionalInfraKubectl + " create secret generic regional-redis-tls --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | " +
		regionalInfraKubectl + " apply -f -; " +
		"fi; " +
		"test -s \"$tmpdir/ca.crt\" && test -s \"$tmpdir/tls.crt\" && test -s \"$tmpdir/tls.key\"; " +
		remoteRegionalKubectl + " create secret generic regional-redis-tls --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | " +
		remoteRegionalKubectl + " apply -f -"
	clusterProxyCert := b.clusterProxyCertScript()
	commands := append([]string{}, pullSecretCommands...)
	commands = append(commands,
		"("+regionalRedisCert+")",
		"("+clusterProxyCert+")",
		"("+b.serverlessClusterCertsScript()+")",
		"("+b.o11yAuthSecretScript()+")",
	)
	return shellJoin(commands)
}

func (b *starterRemoteWorkloadScriptBuilder) o11yAuthSecretScript() string {
	source := b.baselineKubectl("observability")
	commands := []string{
		"tmpdir=$(mktemp -d); trap 'rm -rf \"$tmpdir\"' EXIT; umask 077",
		source + " get secret/cp9-o11y-runtime -o " +
			shellQuote(`jsonpath={.data.auth0-client-id}`) +
			" | base64 -d >\"$tmpdir/client-id\"",
		source + " get secret/cp9-o11y-runtime -o " +
			shellQuote(`jsonpath={.data.auth0-client-secret}`) +
			" | base64 -d >\"$tmpdir/client-secret\"",
		"test -s \"$tmpdir/client-id\"",
		"test -s \"$tmpdir/client-secret\"",
	}
	for _, namespace := range []string{"observability", starterRemoteRegionalNamespace} {
		target := b.remoteKubectl(namespace)
		commands = append(commands, target+" create secret generic o11y-auth0-client "+
			"--from-file=Auth0ClientID=\"$tmpdir/client-id\" "+
			"--from-file=Auth0ClientSecret=\"$tmpdir/client-secret\" "+
			"--dry-run=client -o yaml | "+
			target+" apply --server-side --force-conflicts --field-manager=cp9-o11y-runtime -f -")
	}
	return shellJoin(commands)
}

func (b *starterRemoteWorkloadScriptBuilder) clusterProxyCertScript() string {
	remoteRegionalKubectl := b.remoteKubectl(starterRemoteRegionalNamespace)
	if strings.TrimSpace(b.Runtime.ComponentPCAARN) == "" {
		return "tmpdir=$(mktemp -d); trap 'rm -rf \"$tmpdir\"' EXIT; " +
			"openssl req -x509 -newkey rsa:2048 -sha256 -nodes -days 7 -subj /CN=dbaas-proxy.tidb-admin.svc -addext subjectAltName=DNS:dbaas-proxy,DNS:dbaas-proxy.tidb-admin,DNS:dbaas-proxy.tidb-admin.svc -keyout \"$tmpdir/tls.key\" -out \"$tmpdir/tls.crt\" >/dev/null 2>&1; " +
			remoteRegionalKubectl + " create secret tls proxy-certificate-secret --cert=\"$tmpdir/tls.crt\" --key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | " +
			remoteRegionalKubectl + " apply -f -"
	}
	pcaArgs, ok := b.pcaCommandArgs()
	if !ok {
		return "echo " + shellQuote(
			"starter dbaas-proxy certificate failed: terraform output session_product_pca.arn must be an ACM PCA ARN",
		) + " >&2; exit 1"
	}
	issueArgsSuffix := []string{
		"--signing-algorithm",
		"SHA256WITHRSA",
		"--validity",
		fmt.Sprintf("Value=%d,Type=DAYS", starterDbaasProxyPCACertificateDays),
		"--template-arn",
		"arn:aws:acm-pca:::template/EndEntityCertificate/V1",
		"--query",
		"CertificateArn",
		"--output",
		"text",
	}
	getCertArgs := append(slices.Clone(pcaArgs.get), "--certificate-arn")
	getCertSuffix := []string{
		"--query",
		"Certificate",
		"--output",
		"text",
	}
	return "tmpdir=$(mktemp -d); trap 'rm -rf \"$tmpdir\"' EXIT; " +
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=dbaas-proxy.tidb-admin.svc -addext subjectAltName=DNS:dbaas-proxy,DNS:dbaas-proxy.tidb-admin,DNS:dbaas-proxy.tidb-admin.svc -keyout \"$tmpdir/tls.key\" -out \"$tmpdir/dbaas-proxy.csr\" >/dev/null 2>&1; " +
		"cert_arn=$(" + shellCommand("aws", pcaArgs.issue...) + " --csr \"fileb://$tmpdir/dbaas-proxy.csr\" " + shellArgs(issueArgsSuffix...) + ") && " +
		"test -n \"$cert_arn\" && " +
		"for i in $(seq 1 30); do " + shellCommand("aws", getCertArgs...) + " \"$cert_arn\" " + shellArgs(getCertSuffix...) + " > \"$tmpdir/tls.crt\" 2>/dev/null && break; sleep 2; done; " +
		"grep -q 'BEGIN CERTIFICATE' \"$tmpdir/tls.crt\" || { echo " + shellQuote("starter dbaas-proxy certificate failed: ACM PCA certificate was not issued before timeout") + " >&2; exit 1; }; " +
		shellCommand("aws", pcaArgs.getCA...) + " > \"$tmpdir/ca.crt\" && " +
		"grep -q 'BEGIN CERTIFICATE' \"$tmpdir/ca.crt\" && " +
		remoteRegionalKubectl + " create secret generic proxy-certificate-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | " +
		remoteRegionalKubectl + " apply -f -"
}

func (b *starterRemoteWorkloadScriptBuilder) pcaProfileForARN(arn string) string {
	if b.Runtime.UseDefaultAWSCredentials {
		return ""
	}
	_, account, ok := acmPCAARNParts(arn)
	if ok && strings.TrimSpace(b.Runtime.AWSAccountID) != "" && account == strings.TrimSpace(b.Runtime.AWSAccountID) {
		return firstNonEmpty(b.Runtime.AWSProfile, "cp9")
	}
	if ok && strings.TrimSpace(b.Runtime.SourceECRAccountID) != "" && account == strings.TrimSpace(b.Runtime.SourceECRAccountID) {
		return firstNonEmpty(b.Runtime.SourceECRProfile, starterImagePullDefaultProfile)
	}
	return firstNonEmpty(b.Runtime.AWSProfile, "cp9")
}

func (b *starterRemoteWorkloadScriptBuilder) imagePullProfileForRegistry(registry string) string {
	return starterImagePullProfileForRegistry(&b.Runtime, registry)
}

func starterImagePullProfileForRegistry(runtime *starterRemoteWorkloadRuntime, registry string) string {
	if runtime == nil {
		return starterImagePullDefaultProfile
	}
	if runtime.UseDefaultAWSCredentials {
		return ""
	}
	if strings.TrimSpace(runtime.ImagePullProfileOverride) != "" {
		return strings.TrimSpace(runtime.ImagePullProfileOverride)
	}
	account, _, ok := ecrRegistryParts(registry)
	if ok && strings.TrimSpace(runtime.AWSAccountID) != "" && account == strings.TrimSpace(runtime.AWSAccountID) {
		return firstNonEmpty(runtime.AWSProfile, "cp9")
	}
	return firstNonEmpty(runtime.SourceECRProfile, starterImagePullDefaultProfile)
}

func (b *starterRemoteWorkloadScriptBuilder) serverlessClusterCertsScript() string {
	pcaArgs, ok := b.pcaCommandArgs()
	if !ok {
		return "echo " + shellQuote(
			"starter serverless certificate failed: terraform output session_product_pca.arn must be an ACM PCA ARN",
		) + " >&2; exit 1"
	}
	sans := []string{
		"DNS:*.tidb-serverless",
		"DNS:*.tidb-serverless.svc",
		"DNS:cop-worker-service",
		"DNS:keyspace-etcd-group-0",
		"DNS:keyspace-etcd-group-0-peer",
		"DNS:keyspace-etcd-group-1",
		"DNS:keyspace-etcd-group-1-peer",
		"DNS:keyspace-etcd-group-2",
		"DNS:keyspace-etcd-group-2-peer",
		"DNS:load-data-worker",
		"DNS:resource-manager-service",
		"DNS:sche-service",
		"DNS:serverless-cluster-dm-master",
		"DNS:serverless-cluster-dm-worker",
		"DNS:serverless-cluster-pd",
		"DNS:serverless-cluster-pd-local",
		"DNS:serverless-cluster-standby-pd",
		"DNS:serverless-cluster-standby-pd-local",
		"DNS:serverless-cluster-standby-pd-peer",
		"DNS:serverless-cluster-witness-pd",
		"DNS:serverless-cluster-witness-pd-local",
		"DNS:serverless-cluster-witness-pd-peer",
		"DNS:serverless-cluster-tidb",
		"DNS:serverless-cluster-tikv-peer",
		"DNS:tikv-worker-service",
		"DNS:tso-service",
	}
	peerServices := []string{
		"keyspace-etcd-group-0-peer",
		"keyspace-etcd-group-1-peer",
		"keyspace-etcd-group-2-peer",
		"load-data-worker",
		"resource-manager-service",
		"sche-service",
		"serverless-cluster-dm-master-peer",
		"serverless-cluster-pd-peer",
		"serverless-cluster-standby-pd-peer",
		"serverless-cluster-tidb-peer",
		"serverless-cluster-tikv-peer",
		"serverless-cluster-witness-pd-peer",
		"tikv-r8g2xl-34v260403-p3-tikv-peer",
		"tikv-r8gxl-34v260403-p0-tikv-peer",
		"tikv-r8gxl-34v260403-p1-tikv-peer",
		"tikv-r8gxl-34v260403-p2-tikv-peer",
		"tikv-r8gxl-34v260403-standby-tikv-peer",
		"tikv-r8gxl-34v260403-witness-tikv-peer",
		"tso-service",
	}
	for _, service := range peerServices {
		sans = append(sans,
			"DNS:*."+service+".tidb-serverless",
			"DNS:*."+service+".tidb-serverless.svc",
		)
	}
	pdmsEnabled := config.ResolvedStarterPDTopology(b.Session).Mode == config.StarterPDModePDMS
	if !pdmsEnabled {
		sans = slices.DeleteFunc(sans, func(value string) bool {
			return strings.Contains(value, "tso-service") || strings.Contains(value, "resource-manager-service")
		})
	}
	for _, san := range slices.Clone(sans) {
		if strings.HasSuffix(san, ".svc") {
			sans = append(sans, san+".cluster.local")
		}
	}
	secretNames := []string{
		"serverless-cluster-cluster-client-secret",
		"serverless-cluster-pd-cluster-secret",
		"serverless-cluster-standby-cluster-client-secret",
		"serverless-cluster-standby-pd-cluster-secret",
		"serverless-cluster-witness-cluster-client-secret",
		"serverless-cluster-witness-pd-cluster-secret",
		"pd-tso-server-cluster-secret",
		"pd-resource-manager-server-cluster-secret",
		"pd-sche-server-cluster-secret",
		"serverless-cluster-tidb-cluster-secret",
		"serverless-cluster-tikv-cluster-secret",
		"tikv-worker-cluster-secret",
		"cop-worker-cluster-secret",
		"cse-backup-cluster-secret",
		"datacleaner-cluster-secret",
		"load-data-worker-cluster-secret",
		"tikv-r8gxl-34v260403-p0-cluster-client-secret",
		"tikv-r8gxl-34v260403-p0-tikv-cluster-secret",
		"tikv-r8gxl-34v260403-p1-cluster-client-secret",
		"tikv-r8gxl-34v260403-p1-tikv-cluster-secret",
		"tikv-r8gxl-34v260403-p2-cluster-client-secret",
		"tikv-r8gxl-34v260403-p2-tikv-cluster-secret",
		"tikv-r8gxl-34v260403-standby-cluster-client-secret",
		"tikv-r8gxl-34v260403-standby-tikv-cluster-secret",
		"tikv-r8gxl-34v260403-witness-cluster-client-secret",
		"tikv-r8gxl-34v260403-witness-tikv-cluster-secret",
		"tikv-r8g2xl-34v260403-p3-cluster-client-secret",
		"tikv-r8g2xl-34v260403-p3-tikv-cluster-secret",
		"keyspace-etcd-group-0-cluster-secret",
		"keyspace-etcd-group-1-cluster-secret",
		"keyspace-etcd-group-2-cluster-secret",
		"serverless-cluster-dm-client-secret",
		"serverless-cluster-dm-master-cluster-secret",
		"serverless-cluster-dm-worker-cluster-secret",
		"schema-manager-cluster-secret",
		"replication-worker-cluster-secret",
		"tiflash-columnar-readnode-cluster-secret",
		"cse-archive-cluster-secret",
		"cse-dfsgc-cluster-secret",
	}
	if !pdmsEnabled {
		secretNames = slices.DeleteFunc(secretNames, func(value string) bool {
			return value == "pd-tso-server-cluster-secret" || value == "pd-resource-manager-server-cluster-secret" ||
				value == "pd-sche-server-cluster-secret"
		})
	}
	poolNamespaces := append([]string{"keyspace-pool", "encrypt-keyspace-pool"}, starterServingLanes...)
	poolNamespaces = append(poolNamespaces, "tidb-worker")
	if starterObservabilityEnabled(b.Session) {
		poolNamespaces = append(poolNamespaces, "monitoring")
	}
	var commands []string
	commands = append(
		commands,
		"tmpdir=$(mktemp -d)",
		"trap 'rm -rf \"$tmpdir\"' EXIT",
		shellCommand("aws", pcaArgs.getCA...)+" > \"$tmpdir/session-root.crt\"",
		"grep -q 'BEGIN CERTIFICATE' \"$tmpdir/session-root.crt\"",
		"reuse_starter_cert=false",
		"if "+b.remoteKubectl("tidb-serverless")+" get secret serverless-cluster-pd-cluster-secret >/dev/null 2>&1; then "+
			b.remoteKubectl(
				"tidb-serverless",
			)+" get secret serverless-cluster-pd-cluster-secret -o jsonpath='{.data.tls\\.crt}' | base64 -d > \"$tmpdir/tls.crt\"; "+
			b.remoteKubectl(
				"tidb-serverless",
			)+" get secret serverless-cluster-pd-cluster-secret -o jsonpath='{.data.tls\\.key}' | base64 -d > \"$tmpdir/tls.key\"; "+
			"ca=$("+b.remoteKubectl("tidb-serverless")+" get secret serverless-cluster-pd-cluster-secret -o jsonpath='{.data.ca\\.crt}' 2>/dev/null || true); "+
			"ca_key=$("+b.remoteKubectl("tidb-serverless")+" get secret serverless-cluster-pd-cluster-secret -o jsonpath='{.data.ca\\.key}' 2>/dev/null || true); "+
			"if [ -n \"$ca\" ]; then printf '%s' \"$ca\" | base64 -d > \"$tmpdir/ca.crt\"; else cp \"$tmpdir/tls.crt\" \"$tmpdir/ca.crt\"; fi; "+
			"if [ -n \"$ca_key\" ]; then printf '%s' \"$ca_key\" | base64 -d > \"$tmpdir/ca.key\"; fi; "+
			"openssl x509 -in \"$tmpdir/ca.crt\" -out \"$tmpdir/intermediate.crt\" >/dev/null 2>&1 || true; "+
			"if test -s \"$tmpdir/ca.key\" && "+
			"openssl verify -CAfile \"$tmpdir/session-root.crt\" \"$tmpdir/intermediate.crt\" >/dev/null 2>&1 && "+
			"openssl verify -CAfile \"$tmpdir/ca.crt\" \"$tmpdir/tls.crt\" >/dev/null 2>&1 && "+
			"[ \"$(openssl x509 -in \"$tmpdir/ca.crt\" -noout -modulus)\" = \"$(openssl rsa -in \"$tmpdir/ca.key\" -noout -modulus 2>/dev/null)\" ] && "+
			"[ \"$(openssl x509 -in \"$tmpdir/tls.crt\" -noout -modulus)\" = \"$(openssl rsa -in \"$tmpdir/tls.key\" -noout -modulus 2>/dev/null)\" ] && "+
			"[ \"$(openssl x509 -in \"$tmpdir/ca.crt\" -noout -modulus)\" != \"$(openssl x509 -in \"$tmpdir/tls.crt\" -noout -modulus)\" ] && "+
			"openssl x509 -in \"$tmpdir/ca.crt\" -text -noout | grep -Fq 'CA:TRUE' && "+
			"openssl x509 -in \"$tmpdir/tls.crt\" -text -noout | grep -Fq 'CA:FALSE' && "+
			"openssl x509 -in \"$tmpdir/tls.crt\" -text -noout | grep -Fq 'DNS:*.tidb-serverless.svc.cluster.local'; then reuse_starter_cert=true; fi; fi",
		"if [ \"$reuse_starter_cert\" != true ]; then "+
			"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=starter-serverless-ca "+
			"-addext basicConstraints=critical,CA:TRUE,pathlen:0 -addext keyUsage=critical,keyCertSign,cRLSign "+
			"-keyout \"$tmpdir/ca.key\" -out \"$tmpdir/ca.csr\" >/dev/null 2>&1; "+
			"cert_arn=$("+shellCommand("aws", pcaArgs.issue...)+" --csr \"fileb://$tmpdir/ca.csr\" "+shellArgs(
			"--signing-algorithm", "SHA256WITHRSA",
			"--validity", fmt.Sprintf("Value=%d,Type=DAYS", starterDbaasProxyPCACertificateDays),
			"--template-arn", "arn:aws:acm-pca:::template/SubordinateCACertificate_PathLen0/V1",
			"--query", "CertificateArn", "--output", "text",
		)+"); "+
			"test -n \"$cert_arn\"; "+
			"for i in $(seq 1 30); do "+shellCommand("aws", pcaArgs.get...)+" --certificate-arn \"$cert_arn\" --query Certificate --output text > \"$tmpdir/intermediate.crt\" 2>/dev/null && break; sleep 2; done; "+
			"grep -q 'BEGIN CERTIFICATE' \"$tmpdir/intermediate.crt\"; "+
			"openssl verify -CAfile \"$tmpdir/session-root.crt\" \"$tmpdir/intermediate.crt\" >/dev/null; "+
			"cat \"$tmpdir/intermediate.crt\" \"$tmpdir/session-root.crt\" > \"$tmpdir/ca.crt\"; "+
			"openssl req -new -newkey rsa:2048 -sha256 -nodes -keyout \"$tmpdir/tls.key\" -subj /CN=starter-serverless -out \"$tmpdir/tls.csr\" >/dev/null 2>&1; "+
			"printf '%s\\n' '[v3_req]' 'subjectAltName="+strings.Join(sans, ",")+"' "+
			"'extendedKeyUsage=serverAuth,clientAuth' 'keyUsage=digitalSignature,keyEncipherment' "+
			"'basicConstraints=critical,CA:FALSE' > \"$tmpdir/tls.ext\"; "+
			"openssl x509 -req -in \"$tmpdir/tls.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" "+
			"-CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/tls.ext\" -extensions v3_req -out \"$tmpdir/tls.crt\" >/dev/null 2>&1; fi",
		"test -s \"$tmpdir/ca.crt\" && test -s \"$tmpdir/ca.key\" && test -s \"$tmpdir/tls.crt\" && test -s \"$tmpdir/tls.key\"",
	)
	for _, name := range secretNames {
		caKeyFile := ""
		if name == "serverless-cluster-pd-cluster-secret" {
			caKeyFile = " --from-file=ca.key=\"$tmpdir/ca.key\""
		}
		commands = append(
			commands,
			b.remoteKubectl(
				"tidb-serverless",
			)+" create secret generic "+shellQuote(
				name,
			)+" --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\""+caKeyFile+" --dry-run=client -o yaml | "+b.remoteKubectl(
				"tidb-serverless",
			)+" apply -f -",
		)
	}
	commands = append(
		commands,
		b.remoteKubectl(
			"import-jobs",
		)+" create secret generic import-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"import-jobs",
		)+" apply -f -",
		b.remoteKubectl(
			"export-job",
		)+" create secret generic export-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"export-job",
		)+" apply -f -",
		b.remoteKubectl(
			"tidb-worker",
		)+" create secret generic tidb-worker-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"tidb-worker",
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:manager-server,DNS:manager-server.tidb-admin,DNS:manager-server.tidb-admin.svc,DNS:manager-server.tidb-admin.svc.cluster.local' 'extendedKeyUsage=serverAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/manager-server.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=manager-server.tidb-admin.svc -keyout \"$tmpdir/manager-server.key\" -out \"$tmpdir/manager-server.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/manager-server.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/manager-server.ext\" -extensions v3_req -out \"$tmpdir/manager-server.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/manager-server.crt\" && test -s \"$tmpdir/manager-server.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic manager-server-cert --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/manager-server.crt\" --from-file=tls.key=\"$tmpdir/manager-server.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:ops-service,DNS:ops-service.cp9-global,DNS:ops-service.cp9-global.svc,DNS:ops-service.cp9-global.svc.cluster.local,DNS:ops-service.tidb-admin,DNS:ops-service.tidb-admin.svc,DNS:ops-service.tidb-admin.svc.cluster.local' 'extendedKeyUsage=serverAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/ops-service.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=ops-service.tidb-admin.svc -keyout \"$tmpdir/ops-service.key\" -out \"$tmpdir/ops-service.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/ops-service.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/ops-service.ext\" -extensions v3_req -out \"$tmpdir/ops-service.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/ops-service.crt\" && test -s \"$tmpdir/ops-service.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic ops-service-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/ops-service.crt\" --from-file=tls.key=\"$tmpdir/ops-service.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:tikv-api-service,DNS:tikv-api-service.tidb-serverless,DNS:tikv-api-service.tidb-serverless.svc,DNS:tikv-api-service.tidb-serverless.svc.cluster.local' 'extendedKeyUsage=serverAuth,clientAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/tikv-api.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=tikv-api-service.tidb-serverless.svc -keyout \"$tmpdir/tikv-api.key\" -out \"$tmpdir/tikv-api.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/tikv-api.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/tikv-api.ext\" -extensions v3_req -out \"$tmpdir/tikv-api.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/tikv-api.crt\" && test -s \"$tmpdir/tikv-api.key\"",
		b.remoteKubectl(
			"tidb-serverless",
		)+" create secret generic tikv-api-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tikv-api.crt\" --from-file=tls.key=\"$tmpdir/tikv-api.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"tidb-serverless",
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'extendedKeyUsage=clientAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/clusterproxy-client.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=clusterproxy-client -keyout \"$tmpdir/clusterproxy-client.key\" -out \"$tmpdir/clusterproxy-client.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/clusterproxy-client.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/clusterproxy-client.ext\" -extensions v3_req -out \"$tmpdir/clusterproxy-client.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/clusterproxy-client.crt\" && test -s \"$tmpdir/clusterproxy-client.key\"",
		b.remoteKubectl(
			"tidb-serverless",
		)+" create secret generic clusterproxy-client-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/clusterproxy-client.crt\" --from-file=tls.key=\"$tmpdir/clusterproxy-client.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"tidb-serverless",
		)+" apply -f -",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic manager-cluster-client-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic clusterproxy-client-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/clusterproxy-client.crt\" --from-file=tls.key=\"$tmpdir/clusterproxy-client.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:tidb-http-gateway,DNS:tidb-http-gateway.tidb-admin,DNS:tidb-http-gateway.tidb-admin.svc' 'extendedKeyUsage=serverAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/http-gateway.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=tidb-http-gateway.tidb-admin.svc -keyout \"$tmpdir/http-gateway.key\" -out \"$tmpdir/http-gateway.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/http-gateway.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/http-gateway.ext\" -extensions v3_req -out \"$tmpdir/http-gateway.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/http-gateway.crt\" && test -s \"$tmpdir/http-gateway.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic gateway-letsencrypt-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/http-gateway.crt\" --from-file=tls.key=\"$tmpdir/http-gateway.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:tidb-gateway-inner,DNS:tidb-gateway-inner.tidb-admin,DNS:tidb-gateway-inner.tidb-admin.svc' 'extendedKeyUsage=serverAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/gateway-inner.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=tidb-gateway-inner.tidb-admin.svc -keyout \"$tmpdir/gateway-inner.key\" -out \"$tmpdir/gateway-inner.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/gateway-inner.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/gateway-inner.ext\" -extensions v3_req -out \"$tmpdir/gateway-inner.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/gateway-inner.crt\" && test -s \"$tmpdir/gateway-inner.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic gateway-cert-inner-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/gateway-inner.crt\" --from-file=tls.key=\"$tmpdir/gateway-inner.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic tidb-gateway-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic tidb-gateway-infra-redis-client-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
		b.remoteKubectl(
			"observability",
		)+" create secret generic vmagent-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"observability",
		)+" apply -f -",
		b.remoteKubectl(
			"observability",
		)+" create secret generic vector-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			"observability",
		)+" apply -f -",
		"printf '%s\\n' '[v3_req]' 'extendedKeyUsage=clientAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/meta-store.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=meta-store -keyout \"$tmpdir/meta-store.key\" -out \"$tmpdir/meta-store.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/meta-store.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/meta-store.ext\" -extensions v3_req -out \"$tmpdir/meta-store.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/meta-store.crt\" && test -s \"$tmpdir/meta-store.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" create secret generic meta-store-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/meta-store.crt\" --from-file=tls.key=\"$tmpdir/meta-store.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
			starterRemoteRegionalNamespace,
		)+" apply -f -",
	)
	for _, namespace := range poolNamespaces {
		commands = append(
			commands,
			"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=starter-"+namespace+" -keyout \"$tmpdir/pool-tls.key\" -out \"$tmpdir/pool-tls.csr\" >/dev/null 2>&1",
			"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:*."+namespace+",DNS:*."+namespace+".svc,DNS:*."+namespace+".pod.cluster.local,IP:127.0.0.1,IP:127.0.0.6,IP:::1' 'extendedKeyUsage=serverAuth,clientAuth' 'keyUsage=digitalSignature,keyEncipherment' 'basicConstraints=critical,CA:FALSE' > \"$tmpdir/pool-tls.ext\"",
			"openssl x509 -req -in \"$tmpdir/pool-tls.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/pool-tls.ext\" -extensions v3_req -out \"$tmpdir/pool-tls.crt\" >/dev/null 2>&1",
			"test -s \"$tmpdir/pool-tls.crt\" && test -s \"$tmpdir/pool-tls.key\"",
			b.remoteKubectl(
				namespace,
			)+" create secret generic tidb-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/pool-tls.crt\" --from-file=tls.key=\"$tmpdir/pool-tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
				namespace,
			)+" apply -f -",
			b.remoteKubectl(
				namespace,
			)+" create secret generic sql-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
				namespace,
			)+" apply -f -",
			b.remoteKubectl(
				namespace,
			)+" create secret generic tidb-session-certificate-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/tls.crt\" --from-file=tls.key=\"$tmpdir/tls.key\" --dry-run=client -o yaml | "+b.remoteKubectl(
				namespace,
			)+" apply -f -",
		)
	}
	if config.StarterAIFTSEnabled(b.Session) {
		commands = append(commands, b.aiFTSCertificateCommands()...)
	}
	return shellJoin(commands)
}

func (b *starterRemoteWorkloadScriptBuilder) aiFTSCertificateCommands() []string {
	return []string{
		"printf '%s\\n' '[v3_req]' 'subjectAltName=DNS:fts-server,DNS:fts-server.tidb-admin,DNS:fts-server.tidb-admin.svc,DNS:fts-server.tidb-admin.svc.cluster.local' 'extendedKeyUsage=serverAuth,clientAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/fts-server.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=fts-server.tidb-admin.svc -keyout \"$tmpdir/fts-server.key\" -out \"$tmpdir/fts-server.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/fts-server.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/fts-server.ext\" -extensions v3_req -out \"$tmpdir/fts-server.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/fts-server.crt\" && test -s \"$tmpdir/fts-server.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		) + " create secret generic fts-server-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/fts-server.crt\" --from-file=tls.key=\"$tmpdir/fts-server.key\" --dry-run=client -o yaml | " + b.remoteKubectl(
			starterRemoteRegionalNamespace,
		) + " apply -f -",
		"printf '%s\\n' '[v3_req]' " + shellQuote(
			"subjectAltName=DNS:ai-gateway,DNS:ai-gateway.tidb-admin,DNS:ai-gateway.tidb-admin.svc,DNS:ai-gateway.tidb-admin.svc.cluster.local,DNS:"+b.aiGatewayPublicHost(),
		) + " 'extendedKeyUsage=serverAuth' 'keyUsage=digitalSignature,keyEncipherment' > \"$tmpdir/ai-gateway.ext\"",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=ai-gateway.tidb-admin.svc -keyout \"$tmpdir/ai-gateway.key\" -out \"$tmpdir/ai-gateway.csr\" >/dev/null 2>&1",
		"openssl x509 -req -in \"$tmpdir/ai-gateway.csr\" -CA \"$tmpdir/ca.crt\" -CAkey \"$tmpdir/ca.key\" -CAcreateserial -days 7 -sha256 -extfile \"$tmpdir/ai-gateway.ext\" -extensions v3_req -out \"$tmpdir/ai-gateway.crt\" >/dev/null 2>&1",
		"test -s \"$tmpdir/ai-gateway.crt\" && test -s \"$tmpdir/ai-gateway.key\"",
		b.remoteKubectl(
			starterRemoteRegionalNamespace,
		) + " create secret generic ai-gateway-cluster-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/ai-gateway.crt\" --from-file=tls.key=\"$tmpdir/ai-gateway.key\" --dry-run=client -o yaml | " + b.remoteKubectl(
			starterRemoteRegionalNamespace,
		) + " apply -f -",
	}
}

func (b *starterRemoteWorkloadScriptBuilder) coreWorkloadsScript() string {
	appDelivery := filepath.Join(b.RepoRoot, "personas", "app-delivery", "envs", "dev-full", "starter-regional")
	applyScript := filepath.Join(b.RepoRoot, "personas", "dev-full", "starter", "apply-remote-kustomize-images.sh")
	o11y := filepath.Join(appDelivery, "o11y-agent-control-server")
	pd := config.ResolvedStarterPDTopology(b.Session)
	tidbServerlessOverlay := "tidb-serverless"
	if pd.Mode == config.StarterPDModeMonolithic {
		tidbServerlessOverlay = "tidb-serverless-pdms-disabled"
	}
	var commands []string
	commands = append(
		commands,
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "premium-metadb"), starterRemoteRegionalNamespace),
		b.remoteKubectl(starterRemoteRegionalNamespace)+" rollout status deployment/premium-metadb --timeout=300s",
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "regional-redis"), "regional-infra"),
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "contract-stubs"), starterRemoteRegionalNamespace),
		b.remoteKubectl(starterRemoteRegionalNamespace)+" rollout status deployment/starter-contract-stub --timeout=300s",
		b.applyKustomizeForServiceImage(applyScript, filepath.Join(appDelivery, "cluster-proxy"), starterRemoteRegionalNamespace, "cluster-proxy"),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "tidb-http-gateway"),
			starterRemoteRegionalNamespace,
			"tidb-http-gateway",
		),
		b.applyKustomizeCRDsOnly(applyScript, o11y),
		b.applyKustomizeCRDsOnly(applyScript, filepath.Join(appDelivery, "manager")),
		b.waitForCRDsScript(),
		b.applyTidbServerless(applyScript, filepath.Join(appDelivery, tidbServerlessOverlay), "tidb-serverless"),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "serverless-meta-store"),
			starterRemoteRegionalNamespace,
			"serverless-meta-store",
		),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "serverless-meta-server"),
			starterRemoteRegionalNamespace,
			"serverless-meta-service",
		),
		b.applyKustomizeForServiceImage(applyScript, filepath.Join(appDelivery, "vpa-agent"), "tidb-vpa", "vpa-agent"),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "gateway-controller"),
			starterRemoteRegionalNamespace,
			"gateway-controller",
			starterGatewayImageRepositoryPlaceholder+"="+b.GatewayImage.Registry+"/"+b.GatewayImage.Repository,
			starterGatewayImagePullSecretPlaceholder+"="+b.imagePullSecretForRegistry(b.GatewayImage.Registry),
		),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "manager"),
			starterRemoteRegionalNamespace,
			"manager",
			starterGatewayVersionPlaceholder+"="+b.GatewayImage.Tag,
		),
		b.applyKustomizeForServiceImage(applyScript, filepath.Join(appDelivery, "ops-service"), starterRemoteRegionalNamespace, "ops-service"),
		b.applyAISearch(applyScript, filepath.Join(appDelivery, "ai-search")),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "tidb-gateway-inner"),
			starterRemoteRegionalNamespace,
			"tidb-gateway-inner",
		),
		b.applyKustomizeForServiceImage(
			applyScript,
			filepath.Join(appDelivery, "serverless-regional-service"),
			starterRemoteRegionalNamespace,
			"serverless-regional-service",
		),
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "keyspace-pool"), "keyspace-pool"),
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "encrypt-keyspace-pool"), "encrypt-keyspace-pool"),
		b.applyKustomizeForServiceImage(applyScript, filepath.Join(appDelivery, "tidb-worker-scaler"), "tidb-worker", "tidb-worker-scaler"),
	)
	for _, lane := range starterServingLanes {
		poolOverlay := strings.TrimSuffix(lane, "-tidb-pool")
		if lane == "tidb-pool" {
			poolOverlay = "default"
		}
		commands = append(commands, b.applyKustomize(
			applyScript,
			filepath.Join(appDelivery, "tidb-pools", poolOverlay),
			lane,
		))
	}
	commands = append(
		commands,
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "import-export", "import"), "import-jobs"),
		b.applyKustomize(applyScript, filepath.Join(appDelivery, "import-export", "export"), "export-job"),
		b.o11yStorageClassPreflightScript(),
		b.applyKustomizeForServiceImage(
			applyScript,
			o11y,
			"observability",
			starterO11yAgentControlServerArtifact,
		),
		b.o11yReadinessScript(),
	)
	if starterObservabilityEnabled(b.Session) {
		retention := 24 * time.Hour
		if b.Session.Starter.Observability.PrometheusRetention != nil {
			retention = b.Session.Starter.Observability.PrometheusRetention.Duration
		}
		pdmsMode := starterPDMSModeEnabled
		if pd.Mode == config.StarterPDModeMonolithic {
			pdmsMode = starterPDMSModeDisabled
		}
		commands = append(commands, "CP9_STARTER_PDMS_MODE="+shellQuote(pdmsMode)+" "+b.applyKustomize(
			applyScript,
			filepath.Join(appDelivery, "observability"),
			"monitoring",
			"__CP9_STARTER_PROMETHEUS_RETENTION__="+retention.String(),
		))
	}
	return shellJoin(commands)
}

func (b *starterRemoteWorkloadScriptBuilder) o11yStorageClassPreflightScript() string {
	check := `import json, sys
ctx = json.load(open(sys.argv[1]))
storage_class = json.load(sys.stdin)
tags = ctx.get("session_tags") or {}
expected = {
    "provisioner": "ebs.csi.aws.com",
    "type": "gp3",
    "tagSpecification_1": "cp9-owner=" + str(tags.get("cp9-owner") or ctx.get("owner") or ""),
    "tagSpecification_2": "cp9-session=" + str(tags.get("cp9-session") or ctx.get("session_name") or ""),
    "tagSpecification_3": "cp9-managed=true",
    "tagSpecification_4": "cp9-ttl=" + str(ctx.get("ttl_expires_at") or ""),
}
actual = {"provisioner": storage_class.get("provisioner")}
actual.update(storage_class.get("parameters") or {})
wrong = [key for key, value in expected.items() if not value or actual.get(key) != value]
if wrong:
    raise SystemExit("Starter O11Y requires immutable tagged ebs-sc; mismatched: " + ", ".join(wrong))`
	return b.remoteKubectl("") + " get storageclass/ebs-sc -o json | python3 -c " +
		shellQuote(check) + " \"$CP9_SESSION_CONTEXT\""
}

func (b *starterRemoteWorkloadScriptBuilder) o11yReadinessScript() string {
	kubectl := b.remoteKubectl("observability")
	managerKubectl := b.remoteKubectl(starterRemoteRegionalNamespace)
	session := shellQuote(strings.TrimSpace(b.Session.Name))
	resources := []string{
		"deployment/o11y-agent-control-server",
		"deployment/victoria-metrics-operator",
		"deployment/vector-operator",
		"daemonset/o11y-vector",
		"statefulset/o11y-vector-sharedpool",
		"statefulset/vmagent-o11y-vmagent-0",
		"statefulset/vmagent-o11y-vmagent-1",
		"statefulset/vmagent-o11y-vmagent-sharedpool-0",
		"statefulset/vmagent-o11y-vmagent-sharedpool-1",
	}
	quoted := make([]string, 0, len(resources))
	for _, resourceRef := range resources {
		quoted = append(quoted, shellQuote(resourceRef))
	}
	return shellJoin([]string{
		kubectl + " rollout status deployment/o11y-agent-control-server --timeout=300s",
		kubectl + " wait --for=jsonpath='{.status.phase}'=Normal kubernetesagent/" + session + " --timeout=900s",
		kubectl + " annotate sharedpoolagent/" + session + " o11y.cloud.pingcap.com/pause-reconciliation=false --overwrite",
		kubectl + " wait --for=jsonpath='{.status.phase}'=Normal sharedpoolagent/" + session + " --timeout=900s",
		"parent_id=$(" + kubectl + " get kubernetesagent/" + session + " -o jsonpath='{.metadata.annotations.o11y\\.cloud\\.pingcap\\.com/id}')",
		"test -n \"$parent_id\"",
		"manager_json=$(" + managerKubectl + " get deployment/manager -o json)",
		"disabled=$(printf '%s' \"$manager_json\" | jq '[.spec.template.spec.containers[] | select(.name == \"manager\") | .args[]? | select(. == \"--o11y-enable=false\")] | length')",
		"enabled=$(printf '%s' \"$manager_json\" | jq '[.spec.template.spec.containers[] | select(.name == \"manager\") | .args[]? | select(. == \"--o11y-enable=true\")] | length')",
		"configured_parent=$(printf '%s' \"$manager_json\" | jq -r '[.spec.template.spec.containers[] | select(.name == \"manager\") | .env[]? | select(.name == \"O11Y_GLOBAL_CLUSTER_ID\") | .value] | if length == 1 then .[0] else \"\" end')",
		"if test \"$disabled\" = 1 && test \"$enabled\" = 0; then " +
			"test -z \"$configured_parent\" || test \"$configured_parent\" = \"$parent_id\"; " +
			"patch=$(printf '%s' \"$manager_json\" | jq -c --arg id \"$parent_id\" '(.spec.template.spec.containers[] | select(.name == \"manager\")) as $manager | {spec:{template:{spec:{containers:[{name:\"manager\",args:($manager.args | map(if . == \"--o11y-enable=false\" then \"--o11y-enable=true\" else . end)),env:[{name:\"O11Y_GLOBAL_CLUSTER_ID\",value:$id}]}]}}}}'); " +
			managerKubectl + " patch deployment/manager --type=strategic --patch \"$patch\"; " +
			"elif test \"$disabled\" = 0 && test \"$enabled\" = 1 && test \"$configured_parent\" = \"$parent_id\"; then :; " +
			"else echo 'Starter manager O11Y enable state is inconsistent' >&2; exit 1; fi",
		managerKubectl + " rollout status deployment/manager --timeout=300s",
		managerKubectl + " get deployment/manager -o json | jq -e --arg id \"$parent_id\" '([.spec.template.spec.containers[] | select(.name == \"manager\") | .args[]? | select(. == \"--o11y-enable=true\")] | length) == 1 and ([.spec.template.spec.containers[] | select(.name == \"manager\") | .args[]? | select(. == \"--o11y-enable=false\")] | length) == 0 and ([.spec.template.spec.containers[] | select(.name == \"manager\") | .env[]? | select(.name == \"O11Y_GLOBAL_CLUSTER_ID\" and .value == $id)] | length) == 1' >/dev/null",
		"for resource in " + strings.Join(quoted, " ") + "; do " +
			"for i in $(seq 1 120); do " + kubectl + " get \"$resource\" >/dev/null 2>&1 && break; sleep 5; done; " +
			"if test \"$(" + kubectl + " get \"$resource\" -o jsonpath='{.spec.updateStrategy.type}')\" = OnDelete; then " +
			"replicas=$(" + kubectl + " get \"$resource\" -o jsonpath='{.spec.replicas}'); " +
			kubectl + " wait --for=jsonpath='{.status.readyReplicas}'=\"$replicas\" \"$resource\" --timeout=600s; " +
			"else " + kubectl + " rollout status \"$resource\" --timeout=600s; fi; done",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) accountAuthJWKSScript() string {
	sourceKubectl := b.baselineKubectl("cp9-global")
	databaseKubectl := b.baselineKubectl("cp9-system")
	targetKubectl := b.remoteKubectl("")
	helper := filepath.Join(b.RepoRoot, "personas", "dev-full", "starter", "account-jwks")
	namespaces := append([]string{"tidb-serverless"}, starterServingLanes...)
	return strings.Join([]string{
		"set -euo pipefail",
		"jwks_file=$(mktemp)",
		"pf_log=$(mktemp)",
		"pf_pid=",
		"trap 'test -z \"${pf_pid:-}\" || kill \"$pf_pid\" 2>/dev/null || true; rm -f \"${jwks_file:-}\" \"${pf_log:-}\"' EXIT",
		sourceKubectl + " rollout status deployment/account-provider --timeout=300s",
		sourceKubectl + " port-forward service/account-svc 18000:50051 >\"$pf_log\" 2>&1 & pf_pid=$!",
		"for _ in $(seq 1 30); do timeout 1 bash -c '</dev/tcp/127.0.0.1/18000' 2>/dev/null && break; kill -0 \"$pf_pid\" 2>/dev/null || { cat \"$pf_log\" >&2; exit 1; }; sleep 1; done",
		"timeout 1 bash -c '</dev/tcp/127.0.0.1/18000' 2>/dev/null",
		"cd " + shellQuote(b.RepoRoot) + " && go run " + shellQuote(helper) + " 127.0.0.1:18000 " +
			strconv.FormatUint(starterSeedOrgID, 10) + " " +
			strconv.FormatUint(starterAccountJWKSEnsureClusterID, 10) + " >\"$jwks_file\"",
		"jq -e '.keys | length > 0' \"$jwks_file\" >/dev/null",
		"key_id=$(" + databaseKubectl + " exec deployment/premium-metadb -- mysql -uroot -pcp9-local -N -B -e " + shellQuote(
			"SELECT key_id FROM account.org_keys WHERE org_id=0 AND status='active' ORDER BY id DESC LIMIT 1",
		) + " 2>/dev/null)",
		"test -n \"$key_id\"",
		"aws kms tag-resource --region " + shellQuote(b.Runtime.AWSRegion) + " --key-id \"$key_id\" --tags " +
			shellQuote("TagKey="+cpaws.TagOwner+",TagValue="+b.OwnerTag) + " " +
			shellQuote("TagKey="+cpaws.TagSession+",TagValue="+b.Session.Name),
		"for namespace in " + shellArgs(namespaces...) + "; do " +
			targetKubectl + " --namespace \"$namespace\" create secret generic db-tidb-auth-token-jwks-secret " +
			"--from-file=tidb_auth_token_jwks.json=\"$jwks_file\" --dry-run=client -o yaml | " +
			targetKubectl + " --namespace \"$namespace\" apply -f - >/dev/null; done",
	}, "\n")
}

func (b *starterRemoteWorkloadScriptBuilder) sqlGatewayDNSAnnotationScript() string {
	remoteKubectl := b.remoteKubectl(starterRemoteRegionalNamespace)
	serverlessKubectl := b.remoteKubectl("tidb-serverless")
	pcaArgs, ok := b.pcaCommandArgs()
	if !ok {
		return "echo " + shellQuote(
			"starter SQL gateway certificate failed: terraform output session_product_pca.arn must be an ACM PCA ARN",
		) + " >&2; exit 1"
	}
	return shellJoin([]string{
		"gateway_host=$(cat " + shellQuote(starterGatewayHostPath(b.StateDir)) + " 2>/dev/null || true)",
		"test -n \"$gateway_host\"",
		"tmpdir=$(mktemp -d)",
		"trap 'rm -rf \"$tmpdir\"' EXIT",
		"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=tidb-gateway-inner " +
			"-addext \"subjectAltName=DNS:tidb-gateway-inner,DNS:tidb-gateway-inner.tidb-admin,DNS:tidb-gateway-inner.tidb-admin.svc,DNS:$gateway_host\" " +
			"-keyout \"$tmpdir/gateway-inner.key\" -out \"$tmpdir/gateway-inner.csr\" >/dev/null 2>&1",
		"cert_arn=$(" + shellCommand("aws", pcaArgs.issue...) + " --csr \"fileb://$tmpdir/gateway-inner.csr\" " + shellArgs(
			"--signing-algorithm", "SHA256WITHRSA",
			"--validity", fmt.Sprintf("Value=%d,Type=DAYS", starterDbaasProxyPCACertificateDays),
			"--template-arn", "arn:aws:acm-pca:::template/EndEntityCertificate/V1",
			"--query", "CertificateArn", "--output", "text",
		) + ")",
		"test -n \"$cert_arn\"",
		"for i in $(seq 1 30); do " + shellCommand(
			"aws",
			pcaArgs.get...) + " --certificate-arn \"$cert_arn\" --query Certificate --output text > \"$tmpdir/gateway-inner.crt\" 2>/dev/null && break; sleep 2; done",
		"grep -q 'BEGIN CERTIFICATE' \"$tmpdir/gateway-inner.crt\" || { echo " + shellQuote(
			"starter SQL gateway certificate failed: ACM PCA certificate was not issued before timeout",
		) + " >&2; exit 1; }",
		serverlessKubectl + " get secret serverless-cluster-pd-cluster-secret -o jsonpath='{.data.ca\\.crt}' | base64 -d > \"$tmpdir/ca.crt\"",
		"openssl verify -CAfile \"$tmpdir/ca.crt\" \"$tmpdir/gateway-inner.crt\"",
		remoteKubectl + " create secret generic gateway-cert-inner-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/gateway-inner.crt\" --from-file=tls.key=\"$tmpdir/gateway-inner.key\" --dry-run=client -o yaml | " + remoteKubectl + " apply -f -",
		remoteKubectl + " rollout restart deployment/tidb-gateway-inner",
		remoteKubectl + " rollout status deployment/tidb-gateway-inner --timeout=300s",
		remoteKubectl + " annotate service/tidb-gateway-new " + shellQuote(starterGatewayDNSAnnotation) + "=\"$gateway_host\" --overwrite",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) privateLinkServiceScript() string {
	remoteKubectl := b.remoteKubectl(starterRemoteRegionalNamespace)
	pcaArgs, ok := b.pcaCommandArgs()
	if !ok {
		return "echo " + shellQuote(
			"starter PrivateLink certificate failed: terraform output session_product_pca.arn must be an ACM PCA ARN",
		) + " >&2; exit 1"
	}
	privateService := b.Session.Name + "-privatelink"
	privateLBName := b.Session.Name + "-private"
	tagSpec := fmt.Sprintf(
		"ResourceType=vpc-endpoint-service,Tags=[{Key=%s,Value=%s},{Key=%s,Value=%s},{Key=%s,Value=true},{Key=Name,Value=%s}]",
		cpaws.TagOwner,
		b.OwnerTag,
		cpaws.TagSession,
		b.Session.Name,
		cpaws.TagManaged,
		privateService,
	)
	return shellJoin([]string{
		"private_service=" + shellQuote(privateService),
		"private_lb_name=" + shellQuote(privateLBName),
		"lb_tags=$(" + remoteKubectl + " get service/tidb-gateway-new -o jsonpath='{.metadata.annotations.service\\.beta\\.kubernetes\\.io/aws-load-balancer-additional-resource-tags}')",
		"test -n \"$lb_tags\"",
		remoteKubectl + " get service/privatelink -o json | jq --arg name \"$private_service\" --arg lb \"$private_lb_name\" --arg tags \"$lb_tags\" '" +
			".metadata={name:$name,namespace:\"" + starterRemoteRegionalNamespace + "\",annotations:{" +
			"\"service.beta.kubernetes.io/aws-load-balancer-type\":\"external\"," +
			"\"service.beta.kubernetes.io/aws-load-balancer-nlb-target-type\":\"ip\"," +
			"\"service.beta.kubernetes.io/aws-load-balancer-scheme\":\"internal\"," +
			"\"service.beta.kubernetes.io/aws-load-balancer-name\":$lb," +
			"\"service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags\":$tags," +
			"\"service.beta.kubernetes.io/aws-load-balancer-attributes\":\"load_balancing.cross_zone.enabled=true\"," +
			"\"service.beta.kubernetes.io/aws-load-balancer-target-group-attributes\":\"proxy_protocol_v2.enabled=true,proxy_protocol_v2.client_to_server.header_placement=on_first_ack,preserve_client_ip.enabled=false\"," +
			"\"service.beta.kubernetes.io/aws-load-balancer-healthcheck-protocol\":\"TCP\"}} | " +
			".spec.type=\"LoadBalancer\" | del(.spec.clusterIP,.spec.clusterIPs,.spec.ipFamilies,.spec.ipFamilyPolicy,.spec.healthCheckNodePort,.status)' | " +
			remoteKubectl + " apply -f -",
		"lb_host=; for i in $(seq 1 90); do lb_host=$(" + remoteKubectl + " get service/\"$private_service\" -o jsonpath='{.status.loadBalancer.ingress[0].hostname}' 2>/dev/null || true); test -n \"$lb_host\" && break; sleep 5; done",
		"test -n \"$lb_host\" || { echo 'starter PrivateLink NLB hostname was not ready before timeout' >&2; exit 1; }",
		"lb_json=; for i in $(seq 1 30); do lb_json=$(aws elbv2 describe-load-balancers --region " + shellQuote(
			b.Runtime.AWSRegion,
		) + " --names \"$private_lb_name\" --output json 2>/dev/null || true); test \"$(jq -r '.LoadBalancers[0].State.Code // empty' <<<\"$lb_json\")\" = active && break; sleep 5; done",
		"lb_arn=$(jq -r '.LoadBalancers[0].LoadBalancerArn // empty' <<<\"$lb_json\")",
		"availability_zones=$(jq -r '[.LoadBalancers[0].AvailabilityZones[].ZoneName] | join(\",\")' <<<\"$lb_json\")",
		"test -n \"$lb_arn\" && test -n \"$availability_zones\"",
		"endpoint_services=$(aws ec2 describe-vpc-endpoint-service-configurations --region " + shellQuote(b.Runtime.AWSRegion) + " --output json)",
		"service_id=$(jq -r --arg arn \"$lb_arn\" 'first(.ServiceConfigurations[]? | select(.NetworkLoadBalancerArns | index($arn)) | .ServiceId) // empty' <<<\"$endpoint_services\")",
		"if test -z \"$service_id\"; then service_id=$(aws ec2 create-vpc-endpoint-service-configuration --region " + shellQuote(
			b.Runtime.AWSRegion,
		) + " --network-load-balancer-arns \"$lb_arn\" --no-acceptance-required --tag-specifications " + shellQuote(
			tagSpec,
		) + " --query 'ServiceConfiguration.ServiceId' --output text); fi",
		"test -n \"$service_id\"",
		"aws ec2 create-tags --region " + shellQuote(b.Runtime.AWSRegion) + " --resources \"$service_id\" --tags " + shellArgs(
			"Key="+cpaws.TagOwner+",Value="+b.OwnerTag,
			"Key="+cpaws.TagSession+",Value="+b.Session.Name,
			"Key="+cpaws.TagManaged+",Value=true",
		),
		"aws ec2 modify-vpc-endpoint-service-configuration --region " + shellQuote(
			b.Runtime.AWSRegion,
		) + " --service-id \"$service_id\" --no-acceptance-required",
		"aws ec2 modify-vpc-endpoint-service-permissions --region " + shellQuote(
			b.Runtime.AWSRegion,
		) + " --service-id \"$service_id\" --add-allowed-principals '*'",
		"service_name=$(aws ec2 describe-vpc-endpoint-service-configurations --region " + shellQuote(
			b.Runtime.AWSRegion,
		) + " --service-ids \"$service_id\" --query 'ServiceConfigurations[0].ServiceName' --output text)",
		"test -n \"$service_name\"",
		"tmpdir=$(mktemp -d)",
		"trap 'rm -rf \"$tmpdir\"' EXIT",
		remoteKubectl + " get secret/gateway-letsencrypt-secret -o jsonpath='{.data.tls\\.crt}' | base64 -d > \"$tmpdir/current.crt\" 2>/dev/null || true",
		"if ! openssl x509 -in \"$tmpdir/current.crt\" -noout -checkhost \"$lb_host\" >/dev/null 2>&1; then " +
			"openssl req -new -newkey rsa:2048 -sha256 -nodes -subj /CN=tidb-gateway-privatelink " +
			"-addext \"subjectAltName=DNS:tidb-http-gateway,DNS:tidb-http-gateway.tidb-admin,DNS:tidb-http-gateway.tidb-admin.svc,DNS:$lb_host\" " +
			"-keyout \"$tmpdir/privatelink.key\" -out \"$tmpdir/privatelink.csr\" >/dev/null 2>&1; " +
			"cert_arn=$(" + shellCommand("aws", pcaArgs.issue...) + " --csr \"fileb://$tmpdir/privatelink.csr\" " + shellArgs(
			"--signing-algorithm", "SHA256WITHRSA",
			"--validity", fmt.Sprintf("Value=%d,Type=DAYS", starterDbaasProxyPCACertificateDays),
			"--template-arn", "arn:aws:acm-pca:::template/EndEntityCertificate/V1",
			"--query", "CertificateArn", "--output", "text",
		) + "); " +
			"test -n \"$cert_arn\"; " +
			"for i in $(seq 1 30); do " + shellCommand("aws", pcaArgs.get...) + " --certificate-arn \"$cert_arn\" --query Certificate --output text > \"$tmpdir/privatelink.crt\" 2>/dev/null && break; sleep 2; done; " +
			"grep -q 'BEGIN CERTIFICATE' \"$tmpdir/privatelink.crt\"; " +
			shellCommand("aws", pcaArgs.getCA...) + " > \"$tmpdir/ca.crt\"; " +
			"openssl verify -CAfile \"$tmpdir/ca.crt\" \"$tmpdir/privatelink.crt\"; " +
			remoteKubectl + " create secret generic gateway-letsencrypt-secret --type=kubernetes.io/tls --from-file=ca.crt=\"$tmpdir/ca.crt\" --from-file=tls.crt=\"$tmpdir/privatelink.crt\" --from-file=tls.key=\"$tmpdir/privatelink.key\" --dry-run=client -o yaml | " + remoteKubectl + " apply -f -; " +
			"gateway_deployment=$(" + remoteKubectl + " get deployment -l \"version=$(" + remoteKubectl + " get service/privatelink -o jsonpath='{.spec.selector.version}')\" -o jsonpath='{.items[0].metadata.name}'); " +
			"test -n \"$gateway_deployment\"; " + remoteKubectl + " rollout restart deployment/\"$gateway_deployment\"; " + remoteKubectl + " rollout status deployment/\"$gateway_deployment\" --timeout=300s; fi",
		remoteKubectl + " create configmap privatelink-service-config --from-literal=availability-zones=\"$availability_zones\" --from-literal=service-name=\"$service_name\" --from-literal=private-dns=\"$lb_host\" --dry-run=client -o yaml | " + remoteKubectl + " apply -f -",
		remoteKubectl + " rollout restart deployment/manager",
		remoteKubectl + " rollout status deployment/manager --timeout=300s",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) aiGatewayDNSScript() string {
	recordName := b.aiGatewayPublicHost()
	zoneID := strings.TrimSpace(b.Session.Host.Exposure.ZoneID)
	receiptPath := starterAIGatewayDNSReceiptPath(b.StateDir)
	return shellJoin([]string{
		"host=$(cat " + shellQuote(starterAIGatewayHostPath(b.StateDir)) + " 2>/dev/null || true)",
		"test -n \"$host\" || { echo " + shellQuote(
			"starter AI gateway DNS failed: remote ai-gateway LoadBalancer has no hostname/IP",
		) + " >&2; exit 1; }",
		"change_batch=$(printf " + shellQuote(
			`{"Comment":"cp9 Starter AI gateway","Changes":[{"Action":"UPSERT","ResourceRecordSet":{"Name":"%s","Type":"CNAME","TTL":30,"ResourceRecords":[{"Value":"%s"}]}}]}`,
		) + " " + shellQuote(recordName) + " \"$host\")",
		"aws route53 change-resource-record-sets --hosted-zone-id " + shellQuote(zoneID) + " --change-batch \"$change_batch\" >/dev/null",
		"mkdir -p " + shellQuote(filepath.Dir(receiptPath)),
		"printf " + shellQuote(`{"zone_id":"%s","name":"%s","type":"CNAME","target":"%s"}\n`) +
			" " + shellQuote(zoneID) + " " + shellQuote(recordName) + " \"$host\" > " + shellQuote(receiptPath),
	})
}

func (b *starterRemoteWorkloadScriptBuilder) aiFTSDisabledCleanupScript() string {
	receiptPath := starterAIGatewayDNSReceiptPath(b.StateDir)
	aiSearchPath := filepath.Join(
		b.RepoRoot,
		"personas", "app-delivery", "envs", "dev-full", "starter-regional", "ai-search",
	)
	readReceiptField := "python3 -c " + shellQuote(
		`import json,sys; print(str(json.load(open(sys.argv[1])).get(sys.argv[2], "")).strip())`,
	) + " " + shellQuote(receiptPath)
	deleteBatch := shellQuote(
		`import json,sys; name=sys.argv[1].rstrip("."); record=json.loads(sys.argv[2]); matched=isinstance(record,dict) and record.get("Type")=="CNAME" and str(record.get("Name","")).rstrip(".")==name; print(json.dumps({"Comment":"cp9 Starter AI gateway cleanup","Changes":[{"Action":"DELETE","ResourceRecordSet":record}]}) if matched else "")`,
	)
	dnsCleanup := "if [ -s " + shellQuote(receiptPath) + " ]; then " +
		"zone_id=$(" + readReceiptField + " zone_id); record_name=$(" + readReceiptField + " name); " +
		"if [ -n \"$zone_id\" ] && [ -n \"$record_name\" ]; then " +
		"record=$(aws route53 list-resource-record-sets --hosted-zone-id \"$zone_id\" --start-record-name \"$record_name\" --start-record-type CNAME --max-items 1 --query 'ResourceRecordSets[0]' --output json); " +
		"change_batch=$(python3 -c " + deleteBatch + " \"$record_name\" \"$record\"); " +
		"if [ -n \"$change_batch\" ]; then aws route53 change-resource-record-sets --hosted-zone-id \"$zone_id\" --change-batch \"$change_batch\" >/dev/null; fi; " +
		"fi; fi"
	return shellJoin([]string{
		dnsCleanup,
		b.remoteKubectl("") + " delete -k " + shellQuote(aiSearchPath) + " --ignore-not-found --wait=true",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) dbaasProxyLocalBridgeScript() string {
	localRootKubectl := b.baselineKubectl("")
	localAdminKubectl := b.baselineKubectl(starterRemoteRegionalNamespace)
	return shellJoin([]string{
		"host=$(cat " + shellQuote(starterDBAASProxyHostPath(b.StateDir)) + " 2>/dev/null || true)",
		"test -n \"$host\" || { echo " + shellQuote(
			"starter dbaas-proxy bridge failed: remote dbaas-proxy LoadBalancer has no hostname/IP",
		) + " >&2; exit 1; }",
		localRootKubectl + " create namespace " + shellQuote(
			starterRemoteRegionalNamespace,
		) + " --dry-run=client -o yaml | " + localRootKubectl + " apply -f -",
		localAdminKubectl + " create service externalname dbaas-proxy --external-name=\"$host\" --tcp=443:443 --dry-run=client -o yaml | " + localAdminKubectl + " apply -f -",
		localAdminKubectl + " annotate service/dbaas-proxy cp9.tidbcloud.com/remote-dbaas-proxy-host=\"$host\" --overwrite",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) opsServiceLocalBridgeScript() string {
	remoteKubectl := b.remoteKubectl(starterRemoteRegionalNamespace)
	localKubectl := b.baselineKubectl("cp9-global")
	hostPath := starterOpsServiceHostPath(b.StateDir)
	deploymentPatch := `{"spec":{"template":{"spec":{"containers":[{"name":"serverless-service","env":[{"name":"SSL_CERT_DIR","value":"/etc/cp9-starter-ca"}],"volumeMounts":[{"name":"starter-ops-service-ca","mountPath":"/etc/cp9-starter-ca","readOnly":true}]}],"volumes":[{"name":"starter-ops-service-ca","secret":{"secretName":"starter-ops-service-ca"}}]}}}}`
	return shellJoin([]string{
		"host=$(cat " + shellQuote(hostPath) + " 2>/dev/null || true)",
		"test -n \"$host\" || { echo " + shellQuote(
			"starter ops-service bridge failed: remote ops-service LoadBalancer has no hostname/IP",
		) + " >&2; exit 1; }",
		localKubectl + " create service externalname ops-service --external-name=\"$host\" --tcp=443:443 --dry-run=client -o yaml | " + localKubectl + " apply -f -",
		"tmpdir=$(mktemp -d) && trap 'rm -rf \"$tmpdir\"' EXIT",
		remoteKubectl + " get secret/ops-service-cluster-secret -o " + shellQuote(`jsonpath={.data.ca\.crt}`) + " | base64 -d >\"$tmpdir/ca.crt\"",
		"test -s \"$tmpdir/ca.crt\"",
		localKubectl + " create secret generic starter-ops-service-ca --from-file=ca.crt=\"$tmpdir/ca.crt\" --dry-run=client -o yaml | " + localKubectl + " apply -f -",
		"patch=" + shellQuote(deploymentPatch),
		"for deployment in serverless-service serverless-svc-grpc-server serverless-service-worker serverless-svc-br-worker; do " + localKubectl + " patch deployment/\"$deployment\" --type=strategic -p \"$patch\"; done",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) globalOpsEndpointScript() string {
	globalKubectl := b.baselineKubectl("cp9-global")
	patchScript := fmt.Sprintf(`import json, re, sys
url = "https://ops-service.cp9-global.svc:443"
cluster = %q
body = json.load(sys.stdin)
config = (body.get("data") or {}).get("config.yaml") or ""
if (body.get("metadata") or {}).get("name") == "serverless-service-brworker":
    config = re.sub(
        r"(?m)^(?P<indent>\s*)k8sConfig:\s*\[\]\s*$",
        lambda match: (
            f"{match.group('indent')}k8sConfig:\n"
            f"{match.group('indent')}  - kubernetes: {cluster}\n"
            f"{match.group('indent')}    o11yUrl: o11y-stub:9090\n"
            f"{match.group('indent')}    clusterProxyUrl: dbaas-proxy.tidb-admin.svc/proxy/namespace/tidb-serverless/cluster/serverless-cluster\n"
            f"{match.group('indent')}    k8sIngressUrl: {url}"
        ),
        config,
        count=1,
    )
config, count = re.subn(r"(?m)^(?P<indent>\s*)k8sIngressUrl:\s*.*$", lambda match: f"{match.group('indent')}k8sIngressUrl: {url}", config, count=1)
if count != 1:
    raise SystemExit("Starter serverless ConfigMap missing k8sIngressUrl")
json.dump({"data": {"config.yaml": config}}, sys.stdout)`, b.RemoteClusterName)
	digestScript := `import hashlib, json, sys
body = json.load(sys.stdin)
configs = [
    {
        "name": (item.get("metadata") or {}).get("name"),
        "data": item.get("data") or {},
        "binaryData": item.get("binaryData") or {},
    }
    for item in body.get("items") or []
]
payload = json.dumps(sorted(configs, key=lambda item: str(item["name"])), sort_keys=True, separators=(",", ":")).encode()
print(hashlib.sha256(payload).hexdigest())`
	return shellJoin([]string{
		"tmp=$(mktemp) && trap 'rm -f \"$tmp\"' EXIT",
		"for configmap in serverless-service serverless-service-brworker; do " + globalKubectl +
			" get configmap/\"$configmap\" -o json | python3 -c " + shellQuote(patchScript) + " >\"$tmp\" && " + globalKubectl +
			" patch configmap/\"$configmap\" --type=merge --patch-file=\"$tmp\"; done",
		"stamp=$(" + globalKubectl + " get configmap serverless-service serverless-service-brworker -o json | python3 -c " + shellQuote(
			digestScript,
		) + ")",
		`patch=$(printf '{"spec":{"template":{"metadata":{"annotations":{"cp9.tidbcloud.com/runtime-config-sha256":"%s"}}}}}' "$stamp")`,
		"for deployment in serverless-service serverless-svc-grpc-server serverless-service-worker serverless-svc-br-worker; do " + globalKubectl + " patch deployment/\"$deployment\" --type=merge -p \"$patch\"; done",
		"for deployment in serverless-service serverless-svc-grpc-server serverless-service-worker serverless-svc-br-worker; do " + globalKubectl + " rollout status deployment/\"$deployment\" --timeout=300s; done",
	})
}

func (b *starterRemoteWorkloadScriptBuilder) waitForCRDsScript() string {
	crds := []string{
		"backups.pingcap.com",
		"backupschedules.pingcap.com",
		"branches.cloud.pingcap.com",
		"clusters.cloud.pingcap.com",
		"exports.cloud.pingcap.com",
		"gateways.gateway.serverless.pingcap.com",
		"imports.cloud.pingcap.com",
		"keyspaces.serverless.pingcap.com",
		"nodescores.serverless.pingcap.com",
		"restores.pingcap.com",
		"tidbautoscalings.serverless.pingcap.com",
		"tidbclusterautoscalers.pingcap.com",
		"tidbclusters.pingcap.com",
		"tidbdashboards.pingcap.com",
		"tidbinitializers.pingcap.com",
		"tidbmonitors.pingcap.com",
		"tidbngmonitorings.pingcap.com",
		"dmclusters.pingcap.com",
		"kubernetesagents.agent.observability.pingcap.com",
		"sharedpoolagents.agent.observability.pingcap.com",
	}
	var commands []string
	for _, crd := range crds {
		commands = append(
			commands,
			b.remoteKubectl("")+" wait --for=create crd/"+shellQuote(crd)+" --timeout=180s",
			b.remoteKubectl("")+" wait --for=condition=Established crd/"+shellQuote(crd)+" --timeout=180s",
		)
	}
	for _, group := range []string{"cloud.pingcap.com", "gateway.serverless.pingcap.com", "serverless.pingcap.com", "pingcap.com"} {
		commands = append(commands, b.remoteKubectl("")+" api-resources --api-group="+shellQuote(group)+" >/dev/null")
	}
	return shellJoin(commands)
}

func (b *starterRemoteWorkloadScriptBuilder) globalCatalogScript() string {
	evidenceDir := filepath.Join(b.StateDir, "reports", "evidence")
	seedPath := filepath.Join(
		b.RepoRoot,
		"personas",
		"app-delivery",
		"envs",
		"dev-full",
		"starter-regional",
		"serverless-global-service",
		"catalog-seed.sql",
	)
	baselineSystemKubectl := shellCommand(
		"kubectl",
		"--kubeconfig",
		b.Baseline.KubeconfigPath,
		"--context",
		b.Baseline.Context,
		"--namespace",
		"cp9-system",
	)
	regionalURLPath := starterRegionalServiceURLPath(b.StateDir)
	return shellJoin([]string{
		"mkdir -p " + shellQuote(evidenceDir),
		"for i in $(seq 1 60); do tables=$(" + baselineSystemKubectl + " exec deploy/premium-metadb -- mysql -uroot -pcp9-local -Nse " + shellQuote(
			"SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='serverless_global_control_dev' AND table_name IN ('providers','regions');",
		) + " 2>/dev/null || true); if [ \"$tables\" = \"2\" ]; then break; fi; sleep 2; done; test \"${tables:-}\" = \"2\"",
		"regional_url=$(cat " + shellQuote(
			regionalURLPath,
		) + ") && sed \"s|CP9_STARTER_REGIONAL_SERVICE_URL|$regional_url|g\" " + shellQuote(
			seedPath,
		) + " | " + baselineSystemKubectl + " exec -i deploy/premium-metadb -- mysql -uroot -pcp9-local serverless_global_control_dev",
		baselineSystemKubectl + " exec deploy/premium-metadb -- mysql -uroot -pcp9-local -Nse " + shellQuote(
			"SELECT p.name, r.name, r.service_plan, r.status, r.is_public, r.service_address FROM serverless_global_control_dev.providers p JOIN serverless_global_control_dev.regions r ON r.provider_id = p.id ORDER BY p.name, r.name, r.service_plan;",
		) + " > " + shellQuote(
			filepath.Join(evidenceDir, "starter-serverless-global-catalog.txt"),
		),
	})
}

func (b *starterRemoteWorkloadScriptBuilder) applyKustomize(applyScript, path, namespace string, replacements ...string) string {
	args := []string{applyScript, path, namespace, b.Kubeconfig, b.KubeContext}
	args = append(args, replacements...)
	command := shellCommand(args[0], args[1:]...)
	aiFTSEnabled := "0"
	if config.StarterAIFTSEnabled(b.Session) {
		aiFTSEnabled = "1"
	}
	command = "CP9_STARTER_AI_FTS_ENABLED=" + aiFTSEnabled + " " + command
	if secret := b.imagePullSecretForReplacements(replacements); secret != "" {
		command = "CP9_IMAGE_PULL_SECRET=" + shellQuote(secret) + " " + command
	}
	return command
}

func (b *starterRemoteWorkloadScriptBuilder) applyKustomizeForServiceImage(
	applyScript, path, namespace, service string,
	replacements ...string,
) string {
	image, ok := b.workloadImageForService(service)
	if !ok {
		return b.applyKustomize(applyScript, path, namespace, append(
			[]string{starterRemoteWorkloadPlaceholderImage + "="}, replacements...,
		)...)
	}
	if target, found := b.selectedSourceImageForService(service); image.Source == starterRemoteWorkloadImageSourceSummary && found {
		return b.applyKustomizePreservingExistingImage(applyScript, path, namespace, &image, target, replacements...)
	}
	return b.applyKustomize(applyScript, path, namespace, append(
		[]string{starterRemoteWorkloadPlaceholderImage + "=" + image.Image}, replacements...,
	)...)
}

func (b *starterRemoteWorkloadScriptBuilder) applyAISearch(applyScript, path string) string {
	if !config.StarterAIFTSEnabled(b.Session) {
		return "true"
	}
	command := b.applyKustomize(
		applyScript,
		path,
		starterRemoteRegionalNamespace,
		"__CP9_AI_GATEWAY_CERTIFICATE_ARN__="+strings.TrimSpace(b.Session.Host.Exposure.CertificateARN),
	)
	return "CP9_IMAGE_PULL_SECRET=" + shellQuote(
		b.imagePullSecretForRegistry(starterFixedDataplaneStagingRegistry),
	) + " " + command
}

func (b *starterRemoteWorkloadScriptBuilder) applyKustomizePreservingExistingImage(
	applyScript, path, namespace string,
	image *starterRemoteWorkloadImage,
	target *starterRemoteSelectedImage,
	replacements ...string,
) string {
	varName := "cp9_" + strings.ReplaceAll(image.Service, "-", "_") + "_image"
	jsonPath := "jsonpath={.spec.template.spec.containers[?(@.name=='" + target.Container + "')].image}"
	readCurrent := b.remoteKubectl(
		target.Namespace,
	) + " get " + shellQuote(
		target.resourceKind()+"/"+target.Deployment,
	) + " -o " + shellQuote(
		jsonPath,
	) + " 2>/dev/null || true"
	args := []string{applyScript, path, namespace, b.Kubeconfig, b.KubeContext}
	command := shellCommand(args[0], args[1:]...) + " " + shellQuote(starterRemoteWorkloadPlaceholderImage+"=") + "\"$" + varName + "\""
	for _, replacement := range replacements {
		command += " " + shellQuote(replacement)
	}
	if secret := b.imagePullSecretForReplacements([]string{starterRemoteWorkloadPlaceholderImage + "=" + image.Image}); secret != "" {
		command = "CP9_IMAGE_PULL_SECRET=" + shellQuote(secret) + " " + command
	}
	return varName + "=" + shellQuote(image.Image) +
		"; current_image=$(" + readCurrent + ")" +
		"; if [ -n \"$current_image\" ]; then " + varName + "=\"$current_image\"; fi" +
		"; " + command
}

func starterGatewayImage(images []starterRemoteWorkloadImage) (ecrImageReference, error) {
	for _, image := range images {
		if image.Service != "tidb-gateway-inner" {
			continue
		}
		ref, ok := ecrImageReferenceFromImage(image.Image)
		if !ok || ref.Digest != "" || ref.Tag == "" || ref.Tag == "latest" {
			return ecrImageReference{}, cp9errors.New(
				cp9errors.CategoryConfig,
				"selected Starter tidb-gateway image must use an explicit ECR tag",
			)
		}
		return ref, nil
	}
	return ecrImageReference{}, cp9errors.New(cp9errors.CategoryConfig, "selected Starter tidb-gateway image is missing")
}

func (b *starterRemoteWorkloadScriptBuilder) workloadImageForService(service string) (starterRemoteWorkloadImage, bool) {
	for _, item := range b.Images {
		if item.Service == service {
			return item, true
		}
	}
	return starterRemoteWorkloadImage{}, false
}

func (b *starterRemoteWorkloadScriptBuilder) selectedSourceImageForService(service string) (*starterRemoteSelectedImage, bool) {
	if b.SelectedImageSnapshot == nil {
		return nil, false
	}
	for i := range b.SelectedImageSnapshot.SelectedSourceImages {
		if b.SelectedImageSnapshot.SelectedSourceImages[i].Service == service {
			return &b.SelectedImageSnapshot.SelectedSourceImages[i], true
		}
	}
	return nil, false
}

func (b *starterRemoteWorkloadScriptBuilder) applyKustomizeCRDsOnly(applyScript, path string) string {
	return "CP9_APPLY_CRDS_ONLY=1 " + b.applyKustomize(applyScript, path, "-")
}

func (b *starterRemoteWorkloadScriptBuilder) applyTidbServerless(applyScript, path, namespace string) string {
	return b.applyKustomize(
		applyScript,
		path,
		namespace,
		"CP9_STARTER_SERVERLESS_BUCKET="+b.Runtime.StorageBucket,
		"CP9_STARTER_SERVERLESS_PREFIX="+b.Runtime.StoragePrefix,
		"CP9_STARTER_SERVERLESS_REGION="+b.Runtime.StorageRegion,
	)
}

func (b *starterRemoteWorkloadScriptBuilder) baselineKubectl(namespace string) string {
	args := []string{"kubectl", "--kubeconfig", b.Baseline.KubeconfigPath, "--context", b.Baseline.Context}
	if strings.TrimSpace(namespace) != "" {
		args = append(args, "--namespace", namespace)
	}
	return shellCommand(args[0], args[1:]...)
}

func (b *starterRemoteWorkloadScriptBuilder) remoteKubectl(namespace string) string {
	args := []string{"kubectl", "--kubeconfig", b.Kubeconfig, "--context", b.KubeContext}
	if strings.TrimSpace(namespace) != "" {
		args = append(args, "--namespace", namespace)
	}
	return shellCommand(args[0], args[1:]...)
}

func (b *starterRemoteWorkloadScriptBuilder) ecrRegistries() []string {
	seen := map[string]bool{}
	var registries []string
	add := func(registry string) {
		registry = strings.TrimSpace(registry)
		if registry == "" || seen[registry] {
			return
		}
		seen[registry] = true
		registries = append(registries, registry)
	}
	for _, image := range b.Images {
		if registry, ok := ecrRegistryFromImage(image.Image); ok {
			add(registry)
		}
	}
	add(starterFixedDataplaneStagingRegistry)
	add(starterFixedRegionalStagingRegistry)
	return registries
}

func (b *starterRemoteWorkloadScriptBuilder) imagePullSecretForReplacements(replacements []string) string {
	for _, replacement := range replacements {
		_, image, ok := strings.Cut(replacement, "=")
		if !ok {
			continue
		}
		if registry, ok := ecrRegistryFromImage(image); ok {
			return b.imagePullSecretForRegistry(registry)
		}
	}
	return ""
}

func (b *starterRemoteWorkloadScriptBuilder) imagePullSecretForRegistry(registry string) string {
	account, region, ok := ecrRegistryParts(registry)
	if !ok {
		return ""
	}
	if account == starterImagePullDefaultAccountID && region == starterImagePullDefaultRegion {
		return "cp9-ecr-pull-" + account
	}
	return "cp9-ecr-pull-" + account + "-" + region
}

func sessionRepoRoot(session *config.SessionConfig) (string, error) {
	if session != nil && strings.TrimSpace(session.Tilt.File) != "" {
		tiltfile := strings.TrimSpace(session.Tilt.File)
		if filepath.Base(tiltfile) != "" {
			root := filepath.Clean(filepath.Join(filepath.Dir(tiltfile), "..", ".."))
			if _, err := os.Stat(filepath.Join(root, "personas")); err == nil {
				return filepath.Abs(root)
			}
		}
	}
	return os.Getwd()
}

func outputObject(outputs map[string]cpterraform.OutputValue, key string) map[string]any {
	if outputs == nil {
		return nil
	}
	value, ok := outputs[key]
	if !ok || value.Value == nil {
		return nil
	}
	obj, ok := value.Value.(map[string]any)
	if !ok {
		return nil
	}
	return obj
}

func objectField(parent map[string]any, key string) map[string]any {
	if parent == nil {
		return nil
	}
	obj, ok := parent[key].(map[string]any)
	if !ok {
		return nil
	}
	return obj
}

func stringField(parent map[string]any, key string) string {
	if parent == nil {
		return ""
	}
	value, ok := parent[key].(string)
	if !ok {
		return ""
	}
	return strings.TrimSpace(value)
}

func ecrRegistryFromImage(image string) (string, bool) {
	image = strings.TrimSpace(image)
	before, _, ok0 := strings.Cut(image, "/")
	if !ok0 {
		return "", false
	}
	registry := before
	_, _, ok := ecrRegistryParts(registry)
	return registry, ok
}

type ecrImageReference struct {
	Registry   string
	Repository string
	Tag        string
	Digest     string
}

func (r ecrImageReference) ImageID() string {
	if r.Digest != "" {
		return "imageDigest=" + r.Digest
	}
	return "imageTag=" + firstNonEmpty(r.Tag, "latest")
}

func ecrImageReferenceFromImage(image string) (ecrImageReference, bool) {
	image = strings.TrimSpace(image)
	registry, rest, ok := strings.Cut(image, "/")
	if !ok {
		return ecrImageReference{}, false
	}
	if _, _, ok := ecrRegistryParts(registry); !ok {
		return ecrImageReference{}, false
	}
	ref := ecrImageReference{Registry: registry}
	if repository, digest, ok := strings.Cut(rest, "@"); ok {
		ref.Repository = strings.TrimSpace(repository)
		ref.Digest = strings.TrimSpace(digest)
		return ref, ref.Repository != "" && ref.Digest != ""
	}
	ref.Repository = strings.TrimSpace(rest)
	ref.Tag = "latest"
	if tagIndex := strings.LastIndex(rest, ":"); tagIndex >= 0 {
		ref.Repository = strings.TrimSpace(rest[:tagIndex])
		ref.Tag = strings.TrimSpace(rest[tagIndex+1:])
	}
	return ref, ref.Repository != "" && ref.Tag != ""
}

func ecrRegistryParts(registry string) (account, region string, ok bool) {
	parts := strings.Split(strings.TrimSpace(registry), ".")
	if len(parts) < 6 ||
		parts[1] != "dkr" ||
		parts[2] != "ecr" ||
		parts[4] != "amazonaws" ||
		parts[5] != "com" {
		return "", "", false
	}
	account = parts[0]
	region = parts[3]
	if len(account) != 12 || region == "" {
		return "", "", false
	}
	for _, ch := range account {
		if ch < '0' || ch > '9' {
			return "", "", false
		}
	}
	return account, region, true
}

func acmPCAARNParts(arn string) (region, account string, ok bool) {
	parts := strings.SplitN(strings.TrimSpace(arn), ":", 6)
	if len(parts) != 6 ||
		parts[0] != "arn" ||
		parts[2] != "acm-pca" ||
		parts[3] == "" ||
		len(parts[4]) != 12 ||
		!strings.HasPrefix(parts[5], "certificate-authority/") {
		return "", "", false
	}
	for _, ch := range parts[4] {
		if ch < '0' || ch > '9' {
			return "", "", false
		}
	}
	return parts[3], parts[4], true
}

func shellCommand(name string, args ...string) string {
	parts := []string{shellQuote(name)}
	for _, arg := range args {
		parts = append(parts, shellQuote(arg))
	}
	return strings.Join(parts, " ")
}

func shellArgs(args ...string) string {
	var parts []string
	for _, arg := range args {
		parts = append(parts, shellQuote(arg))
	}
	return strings.Join(parts, " ")
}

func shellJoin(commands []string) string {
	var trimmed []string
	for _, command := range commands {
		command = strings.TrimSpace(command)
		if command != "" {
			trimmed = append(trimmed, command)
		}
	}
	return strings.Join(trimmed, " && ")
}

func shellQuote(value string) string {
	return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
}

func starterRemoteWorkloadNamespaces(session *config.SessionConfig) []string {
	namespaces := []string{
		starterRemoteRegionalNamespace,
		"regional-infra",
		"kube-system",
		"local-path-storage",
		"tidb-serverless",
		"keyspace-pool",
		"encrypt-keyspace-pool",
		"tidb-pool",
		"vip-tidb-pool",
		"canary-tidb-pool",
		"canary-vip-tidb-pool",
		"export-tidb-pool",
		"super-vip-tidb-pool",
		"export-super-vip-tidb-pool",
		"export-vip-tidb-pool",
		"canary-super-vip-tidb-pool",
		"tidb-worker",
		"import-jobs",
		"export-job",
		"tidb-vpa",
		"observability",
	}
	if starterObservabilityEnabled(session) {
		namespaces = append(namespaces, "monitoring")
	}
	return namespaces
}

func starterObservabilityEnabled(session *config.SessionConfig) bool {
	return session != nil && session.Starter.Observability.Enabled
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}