package ec2k3d

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"hash/fnv"
	"net/netip"
	"os"
	"os/exec"
	"slices"
	"strconv"
	"strings"
	"time"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	awsarn "github.com/aws/aws-sdk-go-v2/aws/arn"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	elbv2 "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
	elbv2types "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/types"
	"github.com/aws/aws-sdk-go-v2/service/route53"
	route53types "github.com/aws/aws-sdk-go-v2/service/route53/types"
	"github.com/aws/aws-sdk-go-v2/service/ssm"
	"github.com/aws/smithy-go"

	cpaws "github.com/tidbcloud/controlplane9/internal/aws"
	"github.com/tidbcloud/controlplane9/internal/cluster"
	cp9errors "github.com/tidbcloud/controlplane9/internal/errors"
	"github.com/tidbcloud/controlplane9/internal/host"
	remotek3d "github.com/tidbcloud/controlplane9/internal/host/remote/k3d"
	"github.com/tidbcloud/controlplane9/internal/host/remoteimage"
	cpterraform "github.com/tidbcloud/controlplane9/internal/terraform"
	"github.com/tidbcloud/controlplane9/internal/verbose"
)

const (
	maxRootVolumeSizeGiB       = 16_384
	defaultRemoteHostAllowCIDR = "0.0.0.0/0"
	o11yRegionalHostPort       = 81
)

type ec2API interface {
	DescribeInstanceTypes(ctx context.Context, params *ec2.DescribeInstanceTypesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeInstanceTypesOutput, error)
	CreateSecurityGroup(ctx context.Context, params *ec2.CreateSecurityGroupInput,
		optFns ...func(*ec2.Options)) (*ec2.CreateSecurityGroupOutput, error)
	AuthorizeSecurityGroupIngress(ctx context.Context, params *ec2.AuthorizeSecurityGroupIngressInput,
		optFns ...func(*ec2.Options)) (*ec2.AuthorizeSecurityGroupIngressOutput, error)
	DescribeSubnets(ctx context.Context, params *ec2.DescribeSubnetsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeSubnetsOutput, error)
	DescribeNetworkInterfaces(ctx context.Context, params *ec2.DescribeNetworkInterfacesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeNetworkInterfacesOutput, error)
	DescribeImages(ctx context.Context, params *ec2.DescribeImagesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeImagesOutput, error)
	RunInstances(ctx context.Context, params *ec2.RunInstancesInput,
		optFns ...func(*ec2.Options)) (*ec2.RunInstancesOutput, error)
	DescribeInstances(ctx context.Context, params *ec2.DescribeInstancesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeInstancesOutput, error)
	TerminateInstances(ctx context.Context, params *ec2.TerminateInstancesInput,
		optFns ...func(*ec2.Options)) (*ec2.TerminateInstancesOutput, error)
	CreateVpcEndpointServiceConfiguration(ctx context.Context, params *ec2.CreateVpcEndpointServiceConfigurationInput,
		optFns ...func(*ec2.Options)) (*ec2.CreateVpcEndpointServiceConfigurationOutput, error)
	DescribeVpcEndpointServiceConfigurations(ctx context.Context, params *ec2.DescribeVpcEndpointServiceConfigurationsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeVpcEndpointServiceConfigurationsOutput, error)
	StartVpcEndpointServicePrivateDnsVerification(ctx context.Context, params *ec2.StartVpcEndpointServicePrivateDnsVerificationInput,
		optFns ...func(*ec2.Options)) (*ec2.StartVpcEndpointServicePrivateDnsVerificationOutput, error)
	DeleteVpcEndpointServiceConfigurations(ctx context.Context, params *ec2.DeleteVpcEndpointServiceConfigurationsInput,
		optFns ...func(*ec2.Options)) (*ec2.DeleteVpcEndpointServiceConfigurationsOutput, error)
}

type ssmAPI interface {
	GetParameter(ctx context.Context, params *ssm.GetParameterInput,
		optFns ...func(*ssm.Options)) (*ssm.GetParameterOutput, error)
}

type elbv2API interface {
	CreateLoadBalancer(ctx context.Context, params *elbv2.CreateLoadBalancerInput,
		optFns ...func(*elbv2.Options)) (*elbv2.CreateLoadBalancerOutput, error)
	CreateTargetGroup(ctx context.Context, params *elbv2.CreateTargetGroupInput,
		optFns ...func(*elbv2.Options)) (*elbv2.CreateTargetGroupOutput, error)
	RegisterTargets(ctx context.Context, params *elbv2.RegisterTargetsInput,
		optFns ...func(*elbv2.Options)) (*elbv2.RegisterTargetsOutput, error)
	CreateListener(ctx context.Context, params *elbv2.CreateListenerInput,
		optFns ...func(*elbv2.Options)) (*elbv2.CreateListenerOutput, error)
	DescribeListeners(ctx context.Context, params *elbv2.DescribeListenersInput,
		optFns ...func(*elbv2.Options)) (*elbv2.DescribeListenersOutput, error)
	DescribeLoadBalancers(ctx context.Context, params *elbv2.DescribeLoadBalancersInput,
		optFns ...func(*elbv2.Options)) (*elbv2.DescribeLoadBalancersOutput, error)
	DescribeTargetHealth(ctx context.Context, params *elbv2.DescribeTargetHealthInput,
		optFns ...func(*elbv2.Options)) (*elbv2.DescribeTargetHealthOutput, error)
	DeleteLoadBalancer(ctx context.Context, params *elbv2.DeleteLoadBalancerInput,
		optFns ...func(*elbv2.Options)) (*elbv2.DeleteLoadBalancerOutput, error)
	DeleteTargetGroup(ctx context.Context, params *elbv2.DeleteTargetGroupInput,
		optFns ...func(*elbv2.Options)) (*elbv2.DeleteTargetGroupOutput, error)
}

type route53API interface {
	ChangeResourceRecordSets(ctx context.Context, params *route53.ChangeResourceRecordSetsInput,
		optFns ...func(*route53.Options)) (*route53.ChangeResourceRecordSetsOutput, error)
	ListResourceRecordSets(ctx context.Context, params *route53.ListResourceRecordSetsInput,
		optFns ...func(*route53.Options)) (*route53.ListResourceRecordSetsOutput, error)
}

type awsConfigLoader func(ctx context.Context, profile, region string) (awssdk.Config, error)
type localCommandRunner func(cmd *exec.Cmd) ([]byte, error)
type remoteRunner func(ctx context.Context, selected *host.Session, stdin []byte, argv ...string) ([]byte, error)

// Options carries remote-EC2-k3d provisioning knobs from config.
type Options struct {
	Agents              int
	Memory              string
	K3sImage            string
	Profile             string
	Region              string
	RuntimeContractPath string
	OwnerAllowlist      map[string][]string
	TTL                 time.Duration
	InstanceType        string
	AMI                 string
	KeyPair             string
	SSHUser             string
	SSHPrivateKey       string
	SubnetID            string
	SecurityGroupIDs    []string
	InstanceProfile     string
	AllowedCIDR         string
	RootVolumeSizeGiB   int
	PublicHTTPIngress   bool
	O11YEndpointService bool
	Exposure            ExposureOptions
}

// ExposureOptions opts the session into public HTTPS exposure: an
// internet-facing ALB terminating TLS with the runtime account's wildcard
// certificate plus a per-session record in that account's delegated DNS zone.
// The ALB is required over the existing public NLB because the premium Kong
// force-https plugin needs an L7 hop that injects X-Forwarded-Proto: https —
// browsers cannot set that header themselves.
type ExposureOptions struct {
	Enabled        bool
	ZoneID         string
	ZoneName       string
	CertificateARN string
}

// Adapter manages the remote-EC2-k3d host implementation.
type Adapter struct {
	ec2       ec2API
	ssm       ssmAPI
	elbv2     elbv2API
	route53   route53API
	loadAWS   awsConfigLoader
	runLocal  localCommandRunner
	runRemote remoteRunner
}

// NewAdapter returns a remote-EC2-k3d host adapter backed by the real AWS SDK and ssh.
func NewAdapter() *Adapter {
	return &Adapter{
		loadAWS:   cpaws.LoadAWSConfig,
		runLocal:  verbose.RunCommand,
		runRemote: host.RunSSH,
	}
}

// CheckConflict currently accepts remote sessions by name; orphan cleanup is delegated to the reaper.
func (a *Adapter) CheckConflict(context.Context, *host.Session) error { return nil }

// Destroy terminates the selected remote EC2 instance when it exists.
func (a *Adapter) Destroy(ctx context.Context, selected *host.Session) error {
	if !selected.IsRemoteEC2() {
		return nil
	}
	if selected.Remote.O11YEndpointServiceID != "" {
		client, err := a.requireEC2Client(ctx, selected.Remote.Region, selected.Remote.Profile)
		if err != nil {
			return err
		}
		if err := a.destroyO11YEndpointService(ctx, client, selected); err != nil {
			return err
		}
	}
	if selected.Remote.O11YLoadBalancerARN != "" || selected.Remote.O11YLoadBalancerTargetGroup != "" {
		lbClient, err := a.requireELBV2Client(ctx, selected.Remote.Region, selected.Remote.Profile)
		if err != nil {
			return err
		}
		if err := a.destroyO11YLoadBalancer(ctx, lbClient, selected); err != nil {
			return err
		}
	}
	if selected.Remote.PublicLoadBalancerARN != "" || selected.Remote.PublicLoadBalancerTargetGroup != "" {
		lbClient, err := a.requireELBV2Client(ctx, selected.Remote.Region, selected.Remote.Profile)
		if err != nil {
			return err
		}
		if err := a.destroyPublicLoadBalancer(ctx, lbClient, selected); err != nil {
			return err
		}
	}
	if selected.Remote.PublicHTTPSLoadBalancerARN != "" || selected.Remote.PublicHTTPSTargetGroup != "" ||
		selected.Remote.PublicHTTPSRecordName != "" {
		lbClient, err := a.requireELBV2Client(ctx, selected.Remote.Region, selected.Remote.Profile)
		if err != nil {
			return err
		}
		ownerTag, err := persistedOwnerTag(selected.StateDir, selected.ClusterName)
		if err != nil {
			return err
		}
		dnsClient, err := a.requireRoute53Client(
			ctx,
			selected.Remote.Region,
			selected.Remote.Profile,
			selected.Remote.PublicHTTPSRecordRoleARN,
			ownerTag,
		)
		if err != nil {
			return err
		}
		if err := a.destroyPublicHTTPSExposure(ctx, lbClient, dnsClient, selected); err != nil {
			return err
		}
	}
	if strings.TrimSpace(selected.Remote.InstanceID) == "" {
		return nil
	}
	client, err := a.requireEC2Client(ctx, selected.Remote.Region, selected.Remote.Profile)
	if err != nil {
		return err
	}
	_, err = client.TerminateInstances(ctx, &ec2.TerminateInstancesInput{
		InstanceIds: []string{selected.Remote.InstanceID},
	})
	if err != nil {
		return fmt.Errorf("terminate remote EC2 instance %s: %w", selected.Remote.InstanceID, err)
	}
	return nil
}

// ImportImage streams a locally-built image into the remote host and imports it into k3d.
func (a *Adapter) ImportImage(ctx context.Context, selected *host.Session, image string) error {
	if !selected.IsRemoteEC2() {
		return cp9errors.NotImplemented("remote image import requires remote selected host")
	}
	baseArgs, err := host.SSHTransferBaseArgs(selected)
	if err != nil {
		return err
	}
	remoteImport := hostShellCmd(
		"docker", "load",
	) + " >/dev/null && " + hostShellCmd(
		"k3d", "image", "import", image, "-c", selected.ClusterName,
	)
	sshArgs := host.ShellEscapeArgs(append(baseArgs, host.SSHRemoteShellCommand(remoteImport))...)
	script := "docker save " + host.ShellEscape(image) + " | ssh " + sshArgs
	cmd := exec.CommandContext(ctx, "sh", "-lc", script)
	output, err := a.runLocal(cmd)
	if err != nil {
		return fmt.Errorf("stream image %s to remote host: %w\n%s", image, err, string(output))
	}
	return nil
}

// ClusterStep provisions the selected remote EC2 host and then creates the same k3d shape on it.
type ClusterStep struct {
	Selected             *host.Session
	Options              Options
	Adapter              *Adapter
	ResolveOwnerIdentity func(ctx context.Context) (*cpaws.SetupResult, error)
}

func (s *ClusterStep) Name() cluster.StepName { return cluster.StepClusterCreate }

func (s *ClusterStep) Run(ctx context.Context) error {
	if s.Selected == nil {
		return fmt.Errorf("remote host selected session is required")
	}
	s.ensureRemoteSession()
	if err := s.resolveRuntimeInstanceProfile(); err != nil {
		return err
	}
	client, err := s.Adapter.requireEC2Client(ctx, s.Options.Region, s.Options.Profile)
	if err != nil {
		return err
	}
	if err := s.ensureImagePlan(ctx, client); err != nil {
		return err
	}
	if s.Selected.Remote.InstanceID == "" {
		if err := s.provisionInstance(ctx, client); err != nil {
			return err
		}
	}
	if s.Selected.Remote.PublicDNS == "" && s.Selected.Remote.PublicIP == "" {
		if err := s.waitForInstanceReady(ctx, client, s.Selected.Remote.InstanceID); err != nil {
			return err
		}
		if err := s.persistSelected(); err != nil {
			return err
		}
	} else if err := s.refreshInstanceAddress(ctx, client, s.Selected.Remote.InstanceID); err != nil {
		return err
	}
	if s.Options.PublicHTTPIngress && strings.TrimSpace(s.Selected.Remote.PublicLoadBalancerDNS) == "" {
		lbClient, err := s.Adapter.requireELBV2Client(ctx, s.Options.Region, s.Options.Profile)
		if err != nil {
			return err
		}
		if err := s.ensurePublicLoadBalancer(ctx, client, lbClient); err != nil {
			return err
		}
		if err := s.persistSelected(); err != nil {
			return err
		}
	}
	if s.Options.Exposure.Enabled && strings.TrimSpace(s.Selected.Remote.PublicHTTPSURL) == "" {
		lbClient, err := s.Adapter.requireELBV2Client(ctx, s.Options.Region, s.Options.Profile)
		if err != nil {
			return err
		}
		ownerTag, err := s.resolveOwnerTag(ctx)
		if err != nil {
			return err
		}
		dnsClient, err := s.Adapter.requireRoute53Client(
			ctx,
			s.Options.Region,
			s.Options.Profile,
			"",
			ownerTag,
		)
		if err != nil {
			return err
		}
		if err := s.ensurePublicHTTPSExposure(ctx, client, lbClient, dnsClient); err != nil {
			return err
		}
		if err := s.persistSelected(); err != nil {
			return err
		}
	}
	var (
		o11yLBClient  elbv2API
		o11yDNSClient route53API
	)
	if s.Options.O11YEndpointService {
		if !s.Options.Exposure.Enabled {
			return fmt.Errorf("O11Y endpoint service requires public HTTPS exposure")
		}
		o11yLBClient, err = s.Adapter.requireELBV2Client(ctx, s.Options.Region, s.Options.Profile)
		if err != nil {
			return err
		}
		ownerTag, err := s.resolveOwnerTag(ctx)
		if err != nil {
			return err
		}
		o11yDNSClient, err = s.Adapter.requireRoute53Client(
			ctx,
			s.Options.Region,
			s.Options.Profile,
			"",
			ownerTag,
		)
		if err != nil {
			return err
		}
		// Create the NLB before remote bootstrap so AWS can make it Available
		// while SSH, tool installation and k3d proceed independently.
		if err := s.ensureO11YLoadBalancer(ctx, client, o11yLBClient); err != nil {
			return err
		}
		if err := s.persistSelected(); err != nil {
			return err
		}
	}
	o11yHostPort := 0
	if s.Options.O11YEndpointService {
		o11yHostPort = o11yRegionalHostPort
	}
	if err := (&remotek3d.Runtime{
		Selected: s.Selected,
		Options: remotek3d.Options{
			Agents:           s.Options.Agents,
			Memory:           s.Options.Memory,
			K3sImage:         s.Options.K3sImage,
			TTL:              s.Options.TTL,
			SSHUser:          s.Options.SSHUser,
			SSHPrivateKey:    s.Options.SSHPrivateKey,
			PublicHTTP:       s.publicHTTPIngressNeeded(),
			O11YHostPort:     o11yHostPort,
			AllowSSMFallback: true,
		},
		RunRemote: remotek3d.RemoteRunner(s.Adapter.runRemote),
	}).Prepare(ctx); err != nil {
		return err
	}
	if s.Options.O11YEndpointService {
		if err := s.ensureO11YEndpointService(ctx, client, o11yLBClient, o11yDNSClient); err != nil {
			return err
		}
	}
	return s.persistSelected()
}

// WaitForO11YEndpointServicePrivateDNS closes the final O11Y readiness gate
// after the independent Tilt graph has had time to converge.
func (s *ClusterStep) WaitForO11YEndpointServicePrivateDNS(ctx context.Context) error {
	if !s.Options.O11YEndpointService {
		return nil
	}
	selected, err := host.Load(s.Selected.StateDir)
	if err != nil {
		return fmt.Errorf("load remote host for O11Y endpoint readiness: %w", err)
	}
	if selected.Remote == nil || strings.TrimSpace(selected.Remote.O11YEndpointServiceID) == "" {
		return fmt.Errorf("O11Y endpoint service ID is not persisted")
	}
	client, err := s.Adapter.requireEC2Client(ctx, s.Options.Region, s.Options.Profile)
	if err != nil {
		return err
	}
	return waitForO11YEndpointServicePrivateDNS(
		ctx,
		client,
		selected.Remote.O11YEndpointServiceID,
		5*time.Second,
	)
}

func (s *ClusterStep) resolveRuntimeInstanceProfile() error {
	contractPath := strings.TrimSpace(s.Options.RuntimeContractPath)
	if contractPath == "" {
		return fmt.Errorf("resolve remote EC2 runtime identity: terraform runtime output contract path is required")
	}
	contract, err := cpterraform.LoadOutputContract(contractPath)
	if err != nil {
		return fmt.Errorf("resolve remote EC2 runtime identity: %w", err)
	}
	runtime := contract.RemoteEC2Runtime
	if runtime == nil || !runtime.Enabled || strings.TrimSpace(runtime.InstanceProfileName) == "" {
		return fmt.Errorf("resolve remote EC2 runtime identity: terraform output remote_ec2_runtime.instance_profile_name is required")
	}
	generated := strings.TrimSpace(runtime.InstanceProfileName)
	persisted := strings.TrimSpace(s.Selected.Remote.InstanceProfileName)
	if strings.TrimSpace(s.Selected.Remote.InstanceID) != "" && persisted == "" {
		return fmt.Errorf(
			"legacy remote session %q has EC2 instance %q without a persisted runtime instance profile; "+
				"this session cannot be migrated safely in place: destroy it and run cp9 up again",
			s.Selected.ClusterName,
			s.Selected.Remote.InstanceID,
		)
	}
	if persisted != "" && persisted != generated {
		return fmt.Errorf(
			"persisted remote runtime profile %q does not match generated remote runtime profile %q",
			persisted,
			generated,
		)
	}
	configured := strings.TrimSpace(s.Options.InstanceProfile)
	if configured != "" && configured != generated {
		return fmt.Errorf(
			"host.ec2.instance_profile %q does not match generated remote runtime profile %q",
			configured,
			generated,
		)
	}
	s.Options.InstanceProfile = generated
	s.Selected.Remote.InstanceProfileName = generated
	s.Selected.Remote.RuntimeRoleARN = strings.TrimSpace(runtime.RoleARN)
	return nil
}

// NewClusterStep constructs the real cluster_create implementation for remote-ec2-k3d.
func NewClusterStep(selected *host.Session, opts *Options) *ClusterStep {
	var resolved Options
	if opts != nil {
		resolved = *opts
	}
	return &ClusterStep{
		Selected: selected,
		Options:  resolved,
		Adapter:  NewAdapter(),
	}
}

// RefreshInstanceAddress re-reads and persists the public address for an
// existing remote session. It reports whether the address changed.
func RefreshInstanceAddress(ctx context.Context, selected *host.Session) (bool, error) {
	if !selected.IsRemoteEC2() || strings.TrimSpace(selected.Remote.InstanceID) == "" {
		return false, nil
	}
	adapter := NewAdapter()
	client, err := adapter.requireEC2Client(ctx, selected.Remote.Region, selected.Remote.Profile)
	if err != nil {
		return false, err
	}
	previousDNS, previousIP := selected.Remote.PublicDNS, selected.Remote.PublicIP
	step := &ClusterStep{Selected: selected, Adapter: adapter}
	if err := step.refreshInstanceAddress(ctx, client, selected.Remote.InstanceID); err != nil {
		return false, err
	}
	return previousDNS != selected.Remote.PublicDNS || previousIP != selected.Remote.PublicIP, nil
}

func (a *Adapter) requireEC2Client(ctx context.Context, region, profile string) (ec2API, error) {
	if a.ec2 != nil {
		return a.ec2, nil
	}
	cfg, err := a.loadAWS(ctx, profile, strings.TrimSpace(region))
	if err != nil {
		return nil, err
	}
	return ec2.NewFromConfig(cfg), nil
}

func (a *Adapter) requireSSMClient(ctx context.Context, region, profile string) (ssmAPI, error) {
	if a.ssm != nil {
		return a.ssm, nil
	}
	cfg, err := a.loadAWS(ctx, profile, strings.TrimSpace(region))
	if err != nil {
		return nil, err
	}
	return ssm.NewFromConfig(cfg), nil
}

func (a *Adapter) requireELBV2Client(ctx context.Context, region, profile string) (elbv2API, error) {
	if a.elbv2 != nil {
		return a.elbv2, nil
	}
	cfg, err := a.loadAWS(ctx, profile, strings.TrimSpace(region))
	if err != nil {
		return nil, err
	}
	return elbv2.NewFromConfig(cfg), nil
}

func (a *Adapter) requireRoute53Client(
	ctx context.Context,
	region, profile, roleARN, owner string,
) (route53API, error) {
	if a.route53 != nil {
		return a.route53, nil
	}
	cfg, err := a.loadAWS(ctx, profile, strings.TrimSpace(region))
	if err != nil {
		return nil, err
	}
	if strings.TrimSpace(roleARN) != "" {
		cfg = cpaws.ConfigWithAssumeRole(&cfg, roleARN, owner, nil)
	}
	return route53.NewFromConfig(cfg), nil
}

func persistedOwnerTag(stateDir, session string) (string, error) {
	state, err := cpaws.LoadSessionState(stateDir)
	if err != nil {
		return "", fmt.Errorf("load persisted AWS ownership tags for Route53: %w", err)
	}
	owner := strings.TrimSpace(state.SessionTags[cpaws.TagOwner])
	if owner == "" || strings.TrimSpace(state.SessionTags[cpaws.TagSession]) != strings.TrimSpace(session) {
		return "", fmt.Errorf("persisted AWS owner/session tags are invalid for Route53 session %q", session)
	}
	return owner, nil
}

func (s *ClusterStep) ensureRemoteSession() {
	if s.Selected.Remote == nil {
		s.Selected.Remote = &host.RemoteSession{}
	}
	if strings.TrimSpace(s.Selected.Remote.Profile) == "" {
		s.Selected.Remote.Profile = s.Options.Profile
	}
	if strings.TrimSpace(s.Selected.Remote.Region) == "" {
		s.Selected.Remote.Region = s.Options.Region
	}
}

func (s *ClusterStep) persistSelected() error {
	if err := host.Save(s.Selected.StateDir, s.Selected); err != nil {
		return fmt.Errorf("persist selected host: %w", err)
	}
	return nil
}

func (s *ClusterStep) ensureImagePlan(ctx context.Context, client ec2API) error {
	if s.Selected == nil || s.Selected.Remote == nil {
		return fmt.Errorf("remote host metadata is not available")
	}
	// Legacy config still carries an explicit AMI. Unified v1alpha2 never sets
	// this field and always follows the derived, persisted plan below.
	if strings.TrimSpace(s.Options.AMI) != "" && s.Selected.Remote.Image == nil {
		return nil
	}
	resolver := &remoteimage.Resolver{
		EC2:    client,
		Region: s.Options.Region,
	}
	if s.Selected.Remote.Image != nil {
		if err := resolver.ValidatePinned(ctx, s.Selected.Remote.Image, s.Options.InstanceType); err != nil {
			return fmt.Errorf("validate persisted remote image plan: %w", err)
		}
		s.Options.AMI = s.Selected.Remote.Image.AMI
		return nil
	}
	if strings.TrimSpace(s.Selected.Remote.InstanceID) != "" ||
		strings.TrimSpace(s.Selected.Remote.RemoteSecurityGroup) != "" {
		return fmt.Errorf(
			"legacy remote session %q has mutated AWS host resources without a persisted remote image plan; "+
				"this session cannot be migrated safely in place: destroy it and run cp9 up again",
			s.Selected.ClusterName,
		)
	}
	ssmClient, err := s.Adapter.requireSSMClient(ctx, s.Options.Region, s.Options.Profile)
	if err != nil {
		return fmt.Errorf("load SSM client for remote image plan: %w", err)
	}
	resolver.SSM = ssmClient
	plan, err := resolver.Resolve(ctx, s.Options.InstanceType)
	if err != nil {
		return fmt.Errorf("derive remote image plan: %w", err)
	}
	s.Selected.Remote.Image = plan
	s.Options.AMI = plan.AMI
	// This save is deliberately before owner lookup, security-group creation,
	// ingress changes, or RunInstances so every mutation and retry shares the pin.
	if err := s.persistSelected(); err != nil {
		return fmt.Errorf("persist remote image plan before host mutation: %w", err)
	}
	return nil
}

func (s *ClusterStep) provisionInstance(ctx context.Context, client ec2API) error {
	ownerTag, err := s.resolveOwnerTag(ctx)
	if err != nil {
		return err
	}
	sgID, err := s.ensureSecurityGroup(ctx, client, ownerTag)
	if err != nil {
		return err
	}
	blockDeviceMappings, err := s.rootBlockDeviceMappings(ctx, client)
	if err != nil {
		return err
	}
	ttlExpiresAt := elbv2TTLExpiresAt(s.Options.TTL)
	resp, err := client.RunInstances(ctx, &ec2.RunInstancesInput{
		ImageId:             awssdk.String(s.Options.AMI),
		InstanceType:        ec2types.InstanceType(s.Options.InstanceType),
		KeyName:             awssdk.String(s.Options.KeyPair),
		SubnetId:            awssdk.String(s.Options.SubnetID),
		BlockDeviceMappings: blockDeviceMappings,
		SecurityGroupIds: append(
			append([]string(nil), s.Options.SecurityGroupIDs...),
			sgID,
		),
		MinCount:           awssdk.Int32(1),
		MaxCount:           awssdk.Int32(1),
		IamInstanceProfile: iamInstanceProfile(s.Options.InstanceProfile),
		MetadataOptions: &ec2types.InstanceMetadataOptionsRequest{
			HttpEndpoint:            ec2types.InstanceMetadataEndpointStateEnabled,
			HttpTokens:              ec2types.HttpTokensStateRequired,
			HttpPutResponseHopLimit: awssdk.Int32(3),
		},
		InstanceInitiatedShutdownBehavior: ec2types.ShutdownBehaviorTerminate,
		TagSpecifications: []ec2types.TagSpecification{
			tagSpec(ec2types.ResourceTypeInstance, s.Selected.ClusterName, ownerTag, ttlExpiresAt),
			tagSpec(ec2types.ResourceTypeVolume, s.Selected.ClusterName, ownerTag, ttlExpiresAt),
			tagSpec(ec2types.ResourceTypeNetworkInterface, s.Selected.ClusterName, ownerTag, ttlExpiresAt),
		},
	})
	if err != nil {
		return fmt.Errorf("run remote EC2 host: %w", err)
	}
	if len(resp.Instances) == 0 {
		return fmt.Errorf("run remote EC2 host returned no instances")
	}
	instanceID := awssdk.ToString(resp.Instances[0].InstanceId)
	s.Selected.Remote.InstanceID = instanceID
	s.Selected.Remote.RemoteSecurityGroup = sgID
	if err := s.persistSelected(); err != nil {
		return err
	}
	return nil
}

func (s *ClusterStep) rootBlockDeviceMappings(
	ctx context.Context,
	client ec2API,
) ([]ec2types.BlockDeviceMapping, error) {
	if s.Options.RootVolumeSizeGiB <= 0 {
		return nil, nil
	}
	if s.Options.RootVolumeSizeGiB > maxRootVolumeSizeGiB {
		return nil, fmt.Errorf(
			"host.ec2.root_volume_size_gib must be at most %d GiB",
			maxRootVolumeSizeGiB,
		)
	}
	deviceName, err := s.rootDeviceName(ctx, client)
	if err != nil {
		return nil, err
	}
	return []ec2types.BlockDeviceMapping{{
		DeviceName: awssdk.String(deviceName),
		Ebs: &ec2types.EbsBlockDevice{
			VolumeSize:          awssdk.Int32(int32(s.Options.RootVolumeSizeGiB)),
			DeleteOnTermination: awssdk.Bool(true),
			VolumeType:          ec2types.VolumeTypeGp3,
		},
	}}, nil
}

func (s *ClusterStep) rootDeviceName(ctx context.Context, client ec2API) (string, error) {
	if s.Selected != nil && s.Selected.Remote != nil && s.Selected.Remote.Image != nil {
		rootDeviceName := strings.TrimSpace(s.Selected.Remote.Image.RootDeviceName)
		if rootDeviceName == "" {
			return "", fmt.Errorf("pinned remote EC2 AMI %s root device name is empty", s.Selected.Remote.Image.AMI)
		}
		return rootDeviceName, nil
	}
	out, err := client.DescribeImages(ctx, &ec2.DescribeImagesInput{ImageIds: []string{s.Options.AMI}})
	if err != nil {
		return "", fmt.Errorf("describe remote EC2 AMI %s root device: %w", s.Options.AMI, err)
	}
	if len(out.Images) == 0 {
		return "", fmt.Errorf("describe remote EC2 AMI %s root device: image not found", s.Options.AMI)
	}
	rootDeviceName := strings.TrimSpace(awssdk.ToString(out.Images[0].RootDeviceName))
	if rootDeviceName == "" {
		return "", fmt.Errorf("describe remote EC2 AMI %s root device: root device name is empty", s.Options.AMI)
	}
	return rootDeviceName, nil
}

func (s *ClusterStep) resolveOwnerIdentity(ctx context.Context) (*cpaws.SetupResult, error) {
	if s.ResolveOwnerIdentity != nil {
		return s.ResolveOwnerIdentity(ctx)
	}
	cfg, err := s.Adapter.loadAWS(ctx, s.Options.Profile, s.Options.Region)
	if err != nil {
		return nil, err
	}
	result, err := cpaws.ValidateAccess(ctx, cpaws.NewSTSClient(&cfg), s.Options.Profile, s.Options.Region)
	if err != nil {
		return nil, err
	}
	owner, err := cpaws.ResolveOwnerIdentity(result, s.Options.OwnerAllowlist)
	if err != nil {
		return nil, err
	}
	result.Owner = owner
	return result, nil
}

func (s *ClusterStep) resolveOwnerTag(ctx context.Context) (string, error) {
	state, err := cpaws.LoadSessionState(s.Selected.StateDir)
	if err == nil {
		if sessionID := strings.TrimSpace(state.SessionID); sessionID != s.Selected.ClusterName {
			return "", fmt.Errorf(
				"persisted AWS session %q does not match remote host session %q",
				sessionID,
				s.Selected.ClusterName,
			)
		}
		ownerTag := strings.TrimSpace(state.SessionTags[cpaws.TagOwner])
		sessionTag := strings.TrimSpace(state.SessionTags[cpaws.TagSession])
		if ownerTag == "" || sessionTag != s.Selected.ClusterName {
			return "", fmt.Errorf(
				"persisted AWS owner/session tags are invalid for remote host session %q",
				s.Selected.ClusterName,
			)
		}
		return ownerTag, nil
	}
	if !errors.Is(err, os.ErrNotExist) {
		return "", fmt.Errorf("load persisted AWS ownership tags for remote host: %w", err)
	}

	identity, err := s.resolveOwnerIdentity(ctx)
	if err != nil {
		return "", err
	}
	owner := identity.Owner
	tags, err := cpaws.DeriveSessionTags(owner, s.Selected.ClusterName)
	if err != nil {
		return "", fmt.Errorf("derive remote host ownership tags: %w", err)
	}
	return tags[cpaws.TagOwner], nil
}

func (s *ClusterStep) ensureSecurityGroup(ctx context.Context, client ec2API, owner string) (string, error) {
	if s.Selected.Remote != nil && strings.TrimSpace(s.Selected.Remote.RemoteSecurityGroup) != "" {
		sgID := s.Selected.Remote.RemoteSecurityGroup
		if err := s.ensureSecurityGroupIngress(ctx, client, sgID); err != nil {
			return "", err
		}
		return sgID, nil
	}

	out, err := client.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: []string{s.Options.SubnetID}})
	if err != nil {
		return "", fmt.Errorf("describe subnet %s: %w", s.Options.SubnetID, err)
	}
	if len(out.Subnets) == 0 || out.Subnets[0].VpcId == nil {
		return "", fmt.Errorf("subnet %s did not resolve a VPC", s.Options.SubnetID)
	}
	createOut, err := client.CreateSecurityGroup(ctx, &ec2.CreateSecurityGroupInput{
		GroupName:   awssdk.String("cp9-" + s.Selected.ClusterName + "-host"),
		Description: awssdk.String("cp9 remote host for " + s.Selected.ClusterName),
		VpcId:       out.Subnets[0].VpcId,
		TagSpecifications: []ec2types.TagSpecification{
			tagSpec(ec2types.ResourceTypeSecurityGroup, s.Selected.ClusterName, owner, elbv2TTLExpiresAt(s.Options.TTL)),
		},
	})
	if err != nil {
		return "", fmt.Errorf("create remote security group: %w", err)
	}
	sgID := awssdk.ToString(createOut.GroupId)
	s.Selected.Remote.RemoteSecurityGroup = sgID
	if err := s.persistSelected(); err != nil {
		return "", err
	}
	if err := s.ensureSecurityGroupIngress(ctx, client, sgID); err != nil {
		return "", err
	}
	return sgID, nil
}

func (s *ClusterStep) ensureSecurityGroupIngress(ctx context.Context, client ec2API, groupID string) error {
	cidr := strings.TrimSpace(s.Options.AllowedCIDR)
	if cidr == "" {
		cidr = defaultRemoteHostAllowCIDR
	}
	permissions := []ec2types.IpPermission{
		tcpPermission(22, cidr),
	}
	// Only the legacy public NLB path opens raw host 80/443 to the allowed
	// CIDR. HTTPS exposure must NOT: its public surface is the ALB's 443
	// listener alone, and host port 80 admits just the ALB security group
	// (ensureExposureSecurityGroup), so Kong stays unreachable over plain
	// http://<host>/ that would bypass TLS + X-Forwarded-Proto.
	if s.Options.PublicHTTPIngress {
		permissions = append(permissions, tcpPermission(80, cidr), tcpPermission(443, cidr))
		if subnetCIDR, subnetErr := s.subnetCIDR(ctx, client); subnetErr != nil {
			return subnetErr
		} else if subnetCIDR != "" && subnetCIDR != cidr {
			permissions = append(permissions, tcpPermission(80, subnetCIDR), tcpPermission(443, subnetCIDR))
		}
	}
	for i := range permissions {
		_, err := client.AuthorizeSecurityGroupIngress(ctx, &ec2.AuthorizeSecurityGroupIngressInput{
			GroupId:       awssdk.String(groupID),
			IpPermissions: []ec2types.IpPermission{permissions[i]},
		})
		if err != nil {
			if isDuplicateSecurityGroupIngress(err) {
				continue
			}
			return fmt.Errorf("authorize ingress on remote security group: %w", err)
		}
	}
	return nil
}

func (s *ClusterStep) subnetCIDR(ctx context.Context, client ec2API) (string, error) {
	out, err := client.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: []string{s.Options.SubnetID}})
	if err != nil {
		return "", fmt.Errorf("describe subnet %s for public ingress: %w", s.Options.SubnetID, err)
	}
	if len(out.Subnets) == 0 {
		return "", fmt.Errorf("subnet %s did not resolve for public ingress", s.Options.SubnetID)
	}
	return awssdk.ToString(out.Subnets[0].CidrBlock), nil
}

func tcpPermission(port int32, cidr string) ec2types.IpPermission {
	return ec2types.IpPermission{
		FromPort:   awssdk.Int32(port),
		ToPort:     awssdk.Int32(port),
		IpProtocol: awssdk.String("tcp"),
		IpRanges:   []ec2types.IpRange{{CidrIp: awssdk.String(cidr)}},
	}
}

func (s *ClusterStep) ensurePublicLoadBalancer(ctx context.Context, ec2Client ec2API, lbClient elbv2API) error {
	if s.Selected.Remote == nil || strings.TrimSpace(s.Selected.Remote.InstanceID) == "" {
		return fmt.Errorf("remote EC2 instance is required before public load balancer")
	}
	ownerTag, err := s.resolveOwnerTag(ctx)
	if err != nil {
		return err
	}
	subnets, err := ec2Client.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: []string{s.Options.SubnetID}})
	if err != nil {
		return fmt.Errorf("describe subnet %s for public load balancer: %w", s.Options.SubnetID, err)
	}
	if len(subnets.Subnets) == 0 || subnets.Subnets[0].VpcId == nil {
		return fmt.Errorf("subnet %s did not resolve a VPC for public load balancer", s.Options.SubnetID)
	}
	tags := elbv2Tags(s.Selected.ClusterName, ownerTag, elbv2TTLExpiresAt(s.Options.TTL))

	targetGroupARN := strings.TrimSpace(s.Selected.Remote.PublicLoadBalancerTargetGroup)
	if targetGroupARN == "" {
		tgOut, err := lbClient.CreateTargetGroup(ctx, &elbv2.CreateTargetGroupInput{
			Name:                awssdk.String(remoteTargetGroupName(s.Selected.ClusterName)),
			Port:                awssdk.Int32(80),
			Protocol:            elbv2types.ProtocolEnumTcp,
			HealthCheckProtocol: elbv2types.ProtocolEnumTcp,
			TargetType:          elbv2types.TargetTypeEnumInstance,
			VpcId:               subnets.Subnets[0].VpcId,
			Tags:                tags,
		})
		if err != nil {
			return fmt.Errorf("create remote public target group: %w", err)
		}
		if len(tgOut.TargetGroups) == 0 || tgOut.TargetGroups[0].TargetGroupArn == nil {
			return fmt.Errorf("create remote public target group returned no ARN")
		}
		targetGroupARN = awssdk.ToString(tgOut.TargetGroups[0].TargetGroupArn)
		s.Selected.Remote.PublicLoadBalancerTargetGroup = targetGroupARN
		if err := s.persistSelected(); err != nil {
			return err
		}
	}
	_, err = lbClient.RegisterTargets(ctx, &elbv2.RegisterTargetsInput{
		TargetGroupArn: awssdk.String(targetGroupARN),
		Targets: []elbv2types.TargetDescription{{
			Id:   awssdk.String(s.Selected.Remote.InstanceID),
			Port: awssdk.Int32(80),
		}},
	})
	if err != nil {
		return fmt.Errorf("register remote public target: %w", err)
	}
	s.Selected.Remote.PublicLoadBalancerTargets = []host.RemoteLoadBalancerTarget{{
		ID:   s.Selected.Remote.InstanceID,
		Port: 80,
	}}

	lbARN := strings.TrimSpace(s.Selected.Remote.PublicLoadBalancerARN)
	if lbARN == "" {
		lbOut, err := lbClient.CreateLoadBalancer(ctx, &elbv2.CreateLoadBalancerInput{
			Name:          awssdk.String(remoteLoadBalancerName(s.Selected.ClusterName)),
			Type:          elbv2types.LoadBalancerTypeEnumNetwork,
			Scheme:        elbv2types.LoadBalancerSchemeEnumInternetFacing,
			IpAddressType: elbv2types.IpAddressTypeIpv4,
			Subnets:       []string{s.Options.SubnetID},
			Tags:          tags,
		})
		if err != nil {
			return fmt.Errorf("create remote public load balancer: %w", err)
		}
		if len(lbOut.LoadBalancers) == 0 || lbOut.LoadBalancers[0].LoadBalancerArn == nil {
			return fmt.Errorf("create remote public load balancer returned no ARN")
		}
		lbARN = awssdk.ToString(lbOut.LoadBalancers[0].LoadBalancerArn)
		s.Selected.Remote.PublicLoadBalancerARN = lbARN
		s.Selected.Remote.PublicLoadBalancerDNS = awssdk.ToString(lbOut.LoadBalancers[0].DNSName)
		if err := s.persistSelected(); err != nil {
			return err
		}
	}

	listenerOut, err := lbClient.CreateListener(ctx, &elbv2.CreateListenerInput{
		LoadBalancerArn: awssdk.String(lbARN),
		Port:            awssdk.Int32(80),
		Protocol:        elbv2types.ProtocolEnumTcp,
		DefaultActions: []elbv2types.Action{{
			Type:           elbv2types.ActionTypeEnumForward,
			TargetGroupArn: awssdk.String(targetGroupARN),
		}},
	})
	if err != nil && !isAWSAPIErrorCode(err, "DuplicateListener") {
		return fmt.Errorf("create remote public load balancer listener: %w", err)
	}
	if err == nil && len(listenerOut.Listeners) > 0 {
		s.Selected.Remote.PublicLoadBalancerListener = remoteLoadBalancerListener(&listenerOut.Listeners[0])
	}
	if s.Selected.Remote.PublicLoadBalancerListener == nil {
		if err := s.refreshPublicLoadBalancerListener(ctx, lbClient, lbARN, targetGroupARN); err != nil {
			return err
		}
	}
	if strings.TrimSpace(s.Selected.Remote.PublicLoadBalancerDNS) != "" {
		return nil
	}
	describeOut, err := lbClient.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		LoadBalancerArns: []string{lbARN},
	})
	if err != nil {
		return fmt.Errorf("describe remote public load balancer: %w", err)
	}
	if len(describeOut.LoadBalancers) == 0 || strings.TrimSpace(awssdk.ToString(describeOut.LoadBalancers[0].DNSName)) == "" {
		return fmt.Errorf("remote public load balancer did not publish a DNS name")
	}
	s.Selected.Remote.PublicLoadBalancerDNS = awssdk.ToString(describeOut.LoadBalancers[0].DNSName)
	return nil
}

func (s *ClusterStep) ensureO11YLoadBalancer(ctx context.Context, ec2Client ec2API, lbClient elbv2API) error {
	if s.Selected.Remote == nil || strings.TrimSpace(s.Selected.Remote.InstanceID) == "" {
		return fmt.Errorf("remote EC2 instance is required before O11Y load balancer")
	}
	ttlExpiresAt := elbv2TTLExpiresAt(s.Options.TTL)
	if ttlExpiresAt == "" {
		return fmt.Errorf("O11Y load balancer requires a positive session TTL")
	}
	ownerTag, err := s.resolveOwnerTag(ctx)
	if err != nil {
		return err
	}
	subnets, err := ec2Client.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: []string{s.Options.SubnetID}})
	if err != nil {
		return fmt.Errorf("describe subnet %s for O11Y load balancer: %w", s.Options.SubnetID, err)
	}
	if len(subnets.Subnets) == 0 || subnets.Subnets[0].VpcId == nil {
		return fmt.Errorf("subnet %s did not resolve a VPC for O11Y load balancer", s.Options.SubnetID)
	}
	tags := elbv2Tags(s.Selected.ClusterName, ownerTag, ttlExpiresAt)

	targetGroupARN := strings.TrimSpace(s.Selected.Remote.O11YLoadBalancerTargetGroup)
	if targetGroupARN == "" {
		tgOut, err := lbClient.CreateTargetGroup(ctx, &elbv2.CreateTargetGroupInput{
			Name:                awssdk.String(remoteO11YTargetGroupName(s.Selected.ClusterName)),
			Port:                awssdk.Int32(o11yRegionalHostPort),
			Protocol:            elbv2types.ProtocolEnumTcp,
			HealthCheckProtocol: elbv2types.ProtocolEnumTcp,
			TargetType:          elbv2types.TargetTypeEnumInstance,
			VpcId:               subnets.Subnets[0].VpcId,
			Tags:                tags,
		})
		if err != nil {
			return fmt.Errorf("create O11Y target group: %w", err)
		}
		if len(tgOut.TargetGroups) == 0 || tgOut.TargetGroups[0].TargetGroupArn == nil {
			return fmt.Errorf("create O11Y target group returned no ARN")
		}
		targetGroupARN = awssdk.ToString(tgOut.TargetGroups[0].TargetGroupArn)
		s.Selected.Remote.O11YLoadBalancerTargetGroup = targetGroupARN
		if err := s.persistSelected(); err != nil {
			return err
		}
	}
	_, err = lbClient.RegisterTargets(ctx, &elbv2.RegisterTargetsInput{
		TargetGroupArn: awssdk.String(targetGroupARN),
		Targets: []elbv2types.TargetDescription{{
			Id:   awssdk.String(s.Selected.Remote.InstanceID),
			Port: awssdk.Int32(o11yRegionalHostPort),
		}},
	})
	if err != nil {
		return fmt.Errorf("register O11Y target: %w", err)
	}
	s.Selected.Remote.O11YLoadBalancerTargets = []host.RemoteLoadBalancerTarget{{
		ID:   s.Selected.Remote.InstanceID,
		Port: o11yRegionalHostPort,
	}}

	lbARN := strings.TrimSpace(s.Selected.Remote.O11YLoadBalancerARN)
	if lbARN == "" {
		lbOut, err := lbClient.CreateLoadBalancer(ctx, &elbv2.CreateLoadBalancerInput{
			Name:          awssdk.String(remoteO11YLoadBalancerName(s.Selected.ClusterName)),
			Type:          elbv2types.LoadBalancerTypeEnumNetwork,
			Scheme:        elbv2types.LoadBalancerSchemeEnumInternal,
			IpAddressType: elbv2types.IpAddressTypeIpv4,
			Subnets:       []string{s.Options.SubnetID},
			Tags:          tags,
		})
		if err != nil {
			return fmt.Errorf("create O11Y load balancer: %w", err)
		}
		if len(lbOut.LoadBalancers) == 0 || lbOut.LoadBalancers[0].LoadBalancerArn == nil {
			return fmt.Errorf("create O11Y load balancer returned no ARN")
		}
		lbARN = awssdk.ToString(lbOut.LoadBalancers[0].LoadBalancerArn)
		s.Selected.Remote.O11YLoadBalancerARN = lbARN
		s.Selected.Remote.O11YLoadBalancerDNS = awssdk.ToString(lbOut.LoadBalancers[0].DNSName)
		if err := s.persistSelected(); err != nil {
			return err
		}
	}

	listenerOut, err := lbClient.CreateListener(ctx, &elbv2.CreateListenerInput{
		LoadBalancerArn: awssdk.String(lbARN),
		Port:            awssdk.Int32(80),
		Protocol:        elbv2types.ProtocolEnumTcp,
		DefaultActions: []elbv2types.Action{{
			Type:           elbv2types.ActionTypeEnumForward,
			TargetGroupArn: awssdk.String(targetGroupARN),
		}},
	})
	if err != nil && !isAWSAPIErrorCode(err, "DuplicateListener") {
		return fmt.Errorf("create O11Y load balancer listener: %w", err)
	}
	if err == nil && len(listenerOut.Listeners) > 0 {
		s.Selected.Remote.O11YLoadBalancerListener = remoteLoadBalancerListener(&listenerOut.Listeners[0])
	}
	if s.Selected.Remote.O11YLoadBalancerListener == nil {
		listener, err := findListenerForTargetGroup(ctx, lbClient, lbARN, targetGroupARN)
		if err != nil {
			return fmt.Errorf("find O11Y load balancer listener: %w", err)
		}
		s.Selected.Remote.O11YLoadBalancerListener = listener
	}
	if strings.TrimSpace(s.Selected.Remote.O11YLoadBalancerDNS) != "" {
		return nil
	}
	describeOut, err := lbClient.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		LoadBalancerArns: []string{lbARN},
	})
	if err != nil {
		return fmt.Errorf("describe O11Y load balancer: %w", err)
	}
	if len(describeOut.LoadBalancers) == 0 || strings.TrimSpace(awssdk.ToString(describeOut.LoadBalancers[0].DNSName)) == "" {
		return fmt.Errorf("O11Y load balancer did not publish a DNS name")
	}
	s.Selected.Remote.O11YLoadBalancerDNS = awssdk.ToString(describeOut.LoadBalancers[0].DNSName)
	return nil
}

func (s *ClusterStep) ensureO11YEndpointService(
	ctx context.Context,
	client ec2API,
	lbClient elbv2API,
	dnsClient route53API,
) error {
	if s.Selected.Remote == nil {
		return fmt.Errorf("O11Y endpoint service requires a remote session")
	}
	if strings.TrimSpace(s.Selected.Remote.O11YLoadBalancerARN) == "" {
		return fmt.Errorf("O11Y endpoint service requires the O11Y network load balancer")
	}
	recordName := strings.TrimSpace(s.Selected.Remote.PublicHTTPSRecordName)
	zoneID := strings.TrimSpace(s.Selected.Remote.PublicHTTPSRecordZoneID)
	if recordName == "" || zoneID == "" {
		return fmt.Errorf("O11Y endpoint service requires the session public HTTPS DNS record")
	}
	privateDNSName := "o11y." + strings.TrimSuffix(recordName, ".")
	waiter := elbv2.NewLoadBalancerAvailableWaiter(lbClient)
	if err := waiter.Wait(
		ctx,
		&elbv2.DescribeLoadBalancersInput{
			LoadBalancerArns: []string{s.Selected.Remote.O11YLoadBalancerARN},
		},
		5*time.Minute,
	); err != nil {
		return fmt.Errorf("wait for O11Y load balancer before endpoint service: %w", err)
	}
	sourceCIDR, err := s.networkLoadBalancerSourceCIDR(ctx, client, s.Selected.Remote.O11YLoadBalancerARN)
	if err != nil {
		return err
	}
	s.Selected.Remote.O11YEndpointServiceSourceCIDR = sourceCIDR
	remoteSecurityGroup := strings.TrimSpace(s.Selected.Remote.RemoteSecurityGroup)
	if remoteSecurityGroup == "" {
		return fmt.Errorf("O11Y endpoint service requires the remote host security group")
	}
	_, err = client.AuthorizeSecurityGroupIngress(ctx, &ec2.AuthorizeSecurityGroupIngressInput{
		GroupId:       awssdk.String(remoteSecurityGroup),
		IpPermissions: []ec2types.IpPermission{tcpPermission(o11yRegionalHostPort, sourceCIDR)},
	})
	if err != nil && !isDuplicateSecurityGroupIngress(err) {
		return fmt.Errorf("authorize O11Y load balancer ingress on remote security group: %w", err)
	}
	if err := s.persistSelected(); err != nil {
		return err
	}
	ownerTag, err := s.resolveOwnerTag(ctx)
	if err != nil {
		return err
	}
	ttlExpiresAt := elbv2TTLExpiresAt(s.Options.TTL)
	serviceID := strings.TrimSpace(s.Selected.Remote.O11YEndpointServiceID)
	if serviceID == "" {
		out, err := client.CreateVpcEndpointServiceConfiguration(
			ctx,
			&ec2.CreateVpcEndpointServiceConfigurationInput{
				AcceptanceRequired:      awssdk.Bool(false),
				ClientToken:             awssdk.String(remoteEndpointServiceClientToken(s.Selected.ClusterName)),
				NetworkLoadBalancerArns: []string{s.Selected.Remote.O11YLoadBalancerARN},
				PrivateDnsName:          awssdk.String(privateDNSName),
				TagSpecifications: []ec2types.TagSpecification{
					endpointServiceTagSpec(
						s.Selected.ClusterName,
						ownerTag,
						ttlExpiresAt,
					),
				},
			},
		)
		if err != nil {
			return fmt.Errorf("create O11Y endpoint service: %w", err)
		}
		if out.ServiceConfiguration == nil ||
			strings.TrimSpace(awssdk.ToString(out.ServiceConfiguration.ServiceId)) == "" {
			return fmt.Errorf("create O11Y endpoint service returned no service ID")
		}
		serviceID = awssdk.ToString(out.ServiceConfiguration.ServiceId)
		s.Selected.Remote.O11YEndpointServiceID = serviceID
		s.Selected.Remote.O11YEndpointServiceName = awssdk.ToString(out.ServiceConfiguration.ServiceName)
		if err := s.persistSelected(); err != nil {
			return err
		}
	}

	configuration, err := describeO11YEndpointService(ctx, client, serviceID)
	if err != nil {
		return err
	}
	if !slices.Contains(configuration.NetworkLoadBalancerArns, s.Selected.Remote.O11YLoadBalancerARN) {
		return fmt.Errorf("O11Y endpoint service %s is not attached to the O11Y load balancer", serviceID)
	}
	if got := strings.TrimSuffix(strings.TrimSpace(awssdk.ToString(configuration.PrivateDnsName)), "."); got != privateDNSName {
		return fmt.Errorf(
			"O11Y endpoint service %s private DNS is %q, expected %q",
			serviceID,
			got,
			privateDNSName,
		)
	}
	if !endpointServiceTagsMatch(
		configuration.Tags,
		s.Selected.ClusterName,
		ownerTag,
	) {
		return fmt.Errorf("O11Y endpoint service %s lacks exact session ownership tags", serviceID)
	}
	if strings.TrimSpace(awssdk.ToString(configuration.ServiceName)) != "" {
		s.Selected.Remote.O11YEndpointServiceName = awssdk.ToString(configuration.ServiceName)
	}
	if configuration.PrivateDnsNameConfiguration == nil {
		return fmt.Errorf("O11Y endpoint service %s has no private DNS verification config", serviceID)
	}
	privateDNS := configuration.PrivateDnsNameConfiguration
	if privateDNS.State == ec2types.DnsNameStateVerified {
		return nil
	}
	verificationName := strings.TrimSuffix(strings.TrimSpace(awssdk.ToString(privateDNS.Name)), ".")
	verificationValue := strings.TrimSpace(awssdk.ToString(privateDNS.Value))
	if verificationName == "" || verificationValue == "" {
		return fmt.Errorf("O11Y endpoint service %s returned incomplete private DNS verification", serviceID)
	}
	if !strings.HasSuffix(verificationName, "."+privateDNSName) {
		verificationName += "." + privateDNSName
	}
	if err := changeTXTRecord(
		ctx,
		dnsClient,
		route53types.ChangeActionUpsert,
		zoneID,
		verificationName,
		verificationValue,
		"cp9 "+s.Selected.ClusterName+" O11Y PrivateLink verification",
	); err != nil {
		return fmt.Errorf("upsert O11Y endpoint service verification record: %w", err)
	}
	if _, err := client.StartVpcEndpointServicePrivateDnsVerification(
		ctx,
		&ec2.StartVpcEndpointServicePrivateDnsVerificationInput{ServiceId: awssdk.String(serviceID)},
	); err != nil {
		return fmt.Errorf("start O11Y endpoint service private DNS verification: %w", err)
	}
	return nil
}

func (s *ClusterStep) networkLoadBalancerSourceCIDR(
	ctx context.Context,
	client ec2API,
	loadBalancerARN string,
) (string, error) {
	parsed, err := awsarn.Parse(strings.TrimSpace(loadBalancerARN))
	if err != nil {
		return "", fmt.Errorf("parse O11Y network load balancer ARN: %w", err)
	}
	const resourcePrefix = "loadbalancer/"
	if !strings.HasPrefix(parsed.Resource, resourcePrefix) {
		return "", fmt.Errorf("O11Y network load balancer ARN has unexpected resource %q", parsed.Resource)
	}
	description := "ELB " + strings.TrimPrefix(parsed.Resource, resourcePrefix)
	out, err := client.DescribeNetworkInterfaces(ctx, &ec2.DescribeNetworkInterfacesInput{
		Filters: []ec2types.Filter{
			{Name: awssdk.String("description"), Values: []string{description}},
			{Name: awssdk.String("interface-type"), Values: []string{"network_load_balancer"}},
			{Name: awssdk.String("subnet-id"), Values: []string{s.Options.SubnetID}},
		},
	})
	if err != nil {
		return "", fmt.Errorf("describe O11Y network load balancer interface: %w", err)
	}
	if len(out.NetworkInterfaces) != 1 {
		return "", fmt.Errorf(
			"O11Y network load balancer %s resolved %d interfaces in subnet %s, expected 1",
			description,
			len(out.NetworkInterfaces),
			s.Options.SubnetID,
		)
	}
	address, err := netip.ParseAddr(strings.TrimSpace(awssdk.ToString(out.NetworkInterfaces[0].PrivateIpAddress)))
	if err != nil || !address.Is4() {
		return "", fmt.Errorf("O11Y network load balancer %s has invalid private IPv4 address", description)
	}
	return netip.PrefixFrom(address, 32).String(), nil
}

func describeO11YEndpointService(
	ctx context.Context,
	client ec2API,
	serviceID string,
) (*ec2types.ServiceConfiguration, error) {
	out, err := client.DescribeVpcEndpointServiceConfigurations(
		ctx,
		&ec2.DescribeVpcEndpointServiceConfigurationsInput{ServiceIds: []string{serviceID}},
	)
	if err != nil {
		return nil, fmt.Errorf("describe O11Y endpoint service %s: %w", serviceID, err)
	}
	if out == nil || len(out.ServiceConfigurations) != 1 {
		count := 0
		if out != nil {
			count = len(out.ServiceConfigurations)
		}
		return nil, fmt.Errorf(
			"describe O11Y endpoint service %s returned %d configurations",
			serviceID,
			count,
		)
	}
	configuration := &out.ServiceConfigurations[0]
	if awssdk.ToString(configuration.ServiceId) != serviceID {
		return nil, fmt.Errorf("describe O11Y endpoint service %s returned a conflicting identity", serviceID)
	}
	return configuration, nil
}

func waitForO11YEndpointServicePrivateDNS(
	ctx context.Context,
	client ec2API,
	serviceID string,
	pollInterval time.Duration,
) error {
	for {
		configuration, err := describeO11YEndpointService(ctx, client, serviceID)
		if err != nil {
			return err
		}
		privateDNS := configuration.PrivateDnsNameConfiguration
		if privateDNS != nil {
			switch privateDNS.State {
			case ec2types.DnsNameStateVerified:
				return nil
			case ec2types.DnsNameStateFailed:
				return fmt.Errorf("O11Y endpoint service %s private DNS verification failed", serviceID)
			}
		}
		if err := sleepWithContext(ctx, pollInterval); err != nil {
			return fmt.Errorf("wait for O11Y endpoint service private DNS verification: %w", err)
		}
	}
}

func (s *ClusterStep) refreshPublicLoadBalancerListener(
	ctx context.Context,
	lbClient elbv2API,
	loadBalancerARN string,
	targetGroupARN string,
) error {
	listeners, err := lbClient.DescribeListeners(ctx, &elbv2.DescribeListenersInput{
		LoadBalancerArn: awssdk.String(loadBalancerARN),
	})
	if err != nil {
		return fmt.Errorf("describe remote public load balancer listeners: %w", err)
	}
	for i := range listeners.Listeners {
		listener := &listeners.Listeners[i]
		for _, action := range listener.DefaultActions {
			if awssdk.ToString(action.TargetGroupArn) != targetGroupARN {
				continue
			}
			s.Selected.Remote.PublicLoadBalancerListener = remoteLoadBalancerListener(listener)
			return nil
		}
	}
	return fmt.Errorf("remote public load balancer listener forwarding to target group %s was not found", targetGroupARN)
}

// publicHTTPIngressNeeded reports whether the host must accept public HTTP
// traffic on port 80: either the legacy public NLB or the HTTPS exposure ALB
// forwards to the host's port 80, so both need the SG ingress and the k3d
// host-port mapping.
func (s *ClusterStep) publicHTTPIngressNeeded() bool {
	return s.Options.PublicHTTPIngress || s.Options.Exposure.Enabled
}

func (s *ClusterStep) ensurePublicHTTPSExposure(
	ctx context.Context,
	ec2Client ec2API,
	lbClient elbv2API,
	dnsClient route53API,
) error {
	if s.Selected.Remote == nil || strings.TrimSpace(s.Selected.Remote.InstanceID) == "" {
		return fmt.Errorf("remote EC2 instance is required before public HTTPS exposure")
	}
	certARN := strings.TrimSpace(s.Options.Exposure.CertificateARN)
	zoneID := strings.TrimSpace(s.Options.Exposure.ZoneID)
	zoneName := strings.TrimSuffix(strings.TrimSpace(s.Options.Exposure.ZoneName), ".")
	if certARN == "" || zoneID == "" || zoneName == "" {
		return fmt.Errorf("public HTTPS exposure requires zone_id, zone_name and certificate_arn")
	}
	ownerTag, err := s.resolveOwnerTag(ctx)
	if err != nil {
		return err
	}
	vpcID, subnetIDs, err := s.resolveExposureSubnets(ctx, ec2Client)
	if err != nil {
		return err
	}
	albSecurityGroup, err := s.ensureExposureSecurityGroup(ctx, ec2Client, vpcID, ownerTag)
	if err != nil {
		return err
	}
	tags := elbv2Tags(s.Selected.ClusterName, ownerTag, elbv2TTLExpiresAt(s.Options.TTL))

	targetGroupARN := strings.TrimSpace(s.Selected.Remote.PublicHTTPSTargetGroup)
	if targetGroupARN == "" {
		tgOut, err := lbClient.CreateTargetGroup(ctx, &elbv2.CreateTargetGroupInput{
			Name:                awssdk.String(remoteHTTPSTargetGroupName(s.Selected.ClusterName)),
			Port:                awssdk.Int32(80),
			Protocol:            elbv2types.ProtocolEnumHttp,
			HealthCheckProtocol: elbv2types.ProtocolEnumHttp,
			HealthCheckPath:     awssdk.String("/"),
			// Kong answers 404 on unmatched routes; any HTTP status proves the
			// ingress hop is alive, so widen the healthy matcher accordingly.
			Matcher:    &elbv2types.Matcher{HttpCode: awssdk.String("200-404")},
			TargetType: elbv2types.TargetTypeEnumInstance,
			VpcId:      awssdk.String(vpcID),
			Tags:       tags,
		})
		if err != nil {
			return fmt.Errorf("create public HTTPS target group: %w", err)
		}
		if len(tgOut.TargetGroups) == 0 || tgOut.TargetGroups[0].TargetGroupArn == nil {
			return fmt.Errorf("create public HTTPS target group returned no ARN")
		}
		targetGroupARN = awssdk.ToString(tgOut.TargetGroups[0].TargetGroupArn)
		s.Selected.Remote.PublicHTTPSTargetGroup = targetGroupARN
		if err := s.persistSelected(); err != nil {
			return err
		}
	}

	_, err = lbClient.RegisterTargets(ctx, &elbv2.RegisterTargetsInput{
		TargetGroupArn: awssdk.String(targetGroupARN),
		Targets: []elbv2types.TargetDescription{{
			Id:   awssdk.String(s.Selected.Remote.InstanceID),
			Port: awssdk.Int32(80),
		}},
	})
	if err != nil {
		return fmt.Errorf("register public HTTPS target: %w", err)
	}

	lbARN := strings.TrimSpace(s.Selected.Remote.PublicHTTPSLoadBalancerARN)
	if lbARN == "" {
		lbOut, err := lbClient.CreateLoadBalancer(ctx, &elbv2.CreateLoadBalancerInput{
			Name:           awssdk.String(remoteHTTPSLoadBalancerName(s.Selected.ClusterName)),
			Type:           elbv2types.LoadBalancerTypeEnumApplication,
			Scheme:         elbv2types.LoadBalancerSchemeEnumInternetFacing,
			IpAddressType:  elbv2types.IpAddressTypeIpv4,
			Subnets:        subnetIDs,
			SecurityGroups: []string{albSecurityGroup},
			Tags:           tags,
		})
		if err != nil {
			return fmt.Errorf("create public HTTPS load balancer: %w", err)
		}
		if len(lbOut.LoadBalancers) == 0 || lbOut.LoadBalancers[0].LoadBalancerArn == nil {
			return fmt.Errorf("create public HTTPS load balancer returned no ARN")
		}
		lbARN = awssdk.ToString(lbOut.LoadBalancers[0].LoadBalancerArn)
		s.Selected.Remote.PublicHTTPSLoadBalancerARN = lbARN
		s.Selected.Remote.PublicHTTPSLoadBalancerDNS = awssdk.ToString(lbOut.LoadBalancers[0].DNSName)
		if err := s.persistSelected(); err != nil {
			return err
		}
	}

	listenerOut, err := lbClient.CreateListener(ctx, &elbv2.CreateListenerInput{
		LoadBalancerArn: awssdk.String(lbARN),
		Port:            awssdk.Int32(443),
		Protocol:        elbv2types.ProtocolEnumHttps,
		Certificates:    []elbv2types.Certificate{{CertificateArn: awssdk.String(certARN)}},
		SslPolicy:       awssdk.String("ELBSecurityPolicy-TLS13-1-2-2021-06"),
		DefaultActions: []elbv2types.Action{{
			Type:           elbv2types.ActionTypeEnumForward,
			TargetGroupArn: awssdk.String(targetGroupARN),
		}},
	})
	if err != nil && !isAWSAPIErrorCode(err, "DuplicateListener") {
		return fmt.Errorf("create public HTTPS listener: %w", err)
	}
	if err == nil && len(listenerOut.Listeners) > 0 {
		s.Selected.Remote.PublicHTTPSListener = remoteLoadBalancerListener(&listenerOut.Listeners[0])
	}
	if s.Selected.Remote.PublicHTTPSListener == nil {
		listener, err := findListenerForTargetGroup(ctx, lbClient, lbARN, targetGroupARN)
		if err != nil {
			return err
		}
		s.Selected.Remote.PublicHTTPSListener = listener
	}

	if strings.TrimSpace(s.Selected.Remote.PublicHTTPSLoadBalancerDNS) == "" {
		describeOut, err := lbClient.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
			LoadBalancerArns: []string{lbARN},
		})
		if err != nil {
			return fmt.Errorf("describe public HTTPS load balancer: %w", err)
		}
		if len(describeOut.LoadBalancers) == 0 ||
			strings.TrimSpace(awssdk.ToString(describeOut.LoadBalancers[0].DNSName)) == "" {
			return fmt.Errorf("public HTTPS load balancer did not publish a DNS name")
		}
		s.Selected.Remote.PublicHTTPSLoadBalancerDNS = awssdk.ToString(describeOut.LoadBalancers[0].DNSName)
	}

	recordName := s.Selected.ClusterName + "." + zoneName
	if err := changePublicHTTPSRecord(
		ctx,
		dnsClient,
		route53types.ChangeActionUpsert,
		zoneID,
		recordName,
		s.Selected.Remote.PublicHTTPSLoadBalancerDNS,
		"cp9 "+s.Selected.ClusterName+" public HTTPS exposure",
	); err != nil {
		return fmt.Errorf("upsert public HTTPS record %s: %w", recordName, err)
	}
	s.Selected.Remote.PublicHTTPSRecordName = recordName
	s.Selected.Remote.PublicHTTPSRecordZoneID = zoneID
	s.Selected.Remote.PublicHTTPSURL = "https://" + recordName
	return nil
}

// ensureExposureSecurityGroup isolates the public HTTPS surface at the ALB:
// a dedicated, session-tagged security group admits only 443 from anywhere
// (TLS terminates at the ALB), and the host security group admits port 80
// from that group — not from a public CIDR — so the ALB nodes in every
// selected subnet reach Kong while plain http://<host>/ stays closed. Both
// groups are cp9-tagged; like the host group, teardown belongs to the
// tag-scoped reaper (destroy Stage 2), after the instance and ALB ENIs are
// gone.
func (s *ClusterStep) ensureExposureSecurityGroup(
	ctx context.Context,
	client ec2API,
	vpcID string,
	owner string,
) (string, error) {
	sgID := strings.TrimSpace(s.Selected.Remote.PublicHTTPSSecurityGroup)
	if sgID == "" {
		createOut, err := client.CreateSecurityGroup(ctx, &ec2.CreateSecurityGroupInput{
			GroupName:   awssdk.String("cp9-" + s.Selected.ClusterName + "-alb"),
			Description: awssdk.String("cp9 public HTTPS exposure ALB for " + s.Selected.ClusterName),
			VpcId:       awssdk.String(vpcID),
			TagSpecifications: []ec2types.TagSpecification{
				tagSpec(ec2types.ResourceTypeSecurityGroup, s.Selected.ClusterName, owner, elbv2TTLExpiresAt(s.Options.TTL)),
			},
		})
		if err != nil {
			return "", fmt.Errorf("create public HTTPS exposure security group: %w", err)
		}
		sgID = awssdk.ToString(createOut.GroupId)
		s.Selected.Remote.PublicHTTPSSecurityGroup = sgID
		if err := s.persistSelected(); err != nil {
			return "", err
		}
	}
	_, err := client.AuthorizeSecurityGroupIngress(ctx, &ec2.AuthorizeSecurityGroupIngressInput{
		GroupId:       awssdk.String(sgID),
		IpPermissions: []ec2types.IpPermission{tcpPermission(443, defaultRemoteHostAllowCIDR)},
	})
	if err != nil && !isDuplicateSecurityGroupIngress(err) {
		return "", fmt.Errorf("authorize 443 ingress on exposure security group: %w", err)
	}
	hostSG := strings.TrimSpace(s.Selected.Remote.RemoteSecurityGroup)
	if hostSG != "" {
		_, err := client.AuthorizeSecurityGroupIngress(ctx, &ec2.AuthorizeSecurityGroupIngressInput{
			GroupId: awssdk.String(hostSG),
			IpPermissions: []ec2types.IpPermission{{
				FromPort:   awssdk.Int32(80),
				ToPort:     awssdk.Int32(80),
				IpProtocol: awssdk.String("tcp"),
				UserIdGroupPairs: []ec2types.UserIdGroupPair{{
					GroupId: awssdk.String(sgID),
				}},
			}},
		})
		if err != nil && !isDuplicateSecurityGroupIngress(err) {
			return "", fmt.Errorf("authorize ALB-to-host port 80 ingress on remote security group: %w", err)
		}
	}
	return sgID, nil
}

// resolveExposureSubnets returns the host VPC and a two-subnet, two-AZ subnet
// pair for the ALB: the host subnet plus a deterministic sibling in a
// different availability zone (public-on-launch siblings preferred).
func (s *ClusterStep) resolveExposureSubnets(ctx context.Context, client ec2API) (string, []string, error) {
	out, err := client.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{SubnetIds: []string{s.Options.SubnetID}})
	if err != nil {
		return "", nil, fmt.Errorf("describe subnet %s for public HTTPS exposure: %w", s.Options.SubnetID, err)
	}
	if len(out.Subnets) == 0 || out.Subnets[0].VpcId == nil {
		return "", nil, fmt.Errorf("subnet %s did not resolve a VPC for public HTTPS exposure", s.Options.SubnetID)
	}
	hostSubnet := out.Subnets[0]
	vpcID := awssdk.ToString(hostSubnet.VpcId)
	hostAZ := awssdk.ToString(hostSubnet.AvailabilityZone)
	siblings, err := client.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{
		Filters: []ec2types.Filter{{Name: awssdk.String("vpc-id"), Values: []string{vpcID}}},
	})
	if err != nil {
		return "", nil, fmt.Errorf("describe VPC %s subnets for public HTTPS exposure: %w", vpcID, err)
	}
	type candidate struct {
		id     string
		public bool
	}
	var candidates []candidate
	for i := range siblings.Subnets {
		subnet := &siblings.Subnets[i]
		id := awssdk.ToString(subnet.SubnetId)
		az := awssdk.ToString(subnet.AvailabilityZone)
		if id == "" || id == s.Options.SubnetID || az == "" || az == hostAZ {
			continue
		}
		candidates = append(candidates, candidate{id: id, public: awssdk.ToBool(subnet.MapPublicIpOnLaunch)})
	}
	if len(candidates) == 0 {
		return "", nil, fmt.Errorf(
			"public HTTPS exposure requires a second subnet in a different availability zone in VPC %s (ALB minimum), none found",
			vpcID,
		)
	}
	slices.SortFunc(candidates, func(a, b candidate) int {
		if a.public != b.public {
			if a.public {
				return -1
			}
			return 1
		}
		return strings.Compare(a.id, b.id)
	})
	return vpcID, []string{s.Options.SubnetID, candidates[0].id}, nil
}

func findListenerForTargetGroup(
	ctx context.Context,
	lbClient elbv2API,
	loadBalancerARN string,
	targetGroupARN string,
) (*host.RemoteLoadBalancerListener, error) {
	listeners, err := lbClient.DescribeListeners(ctx, &elbv2.DescribeListenersInput{
		LoadBalancerArn: awssdk.String(loadBalancerARN),
	})
	if err != nil {
		return nil, fmt.Errorf("describe load balancer listeners: %w", err)
	}
	for i := range listeners.Listeners {
		listener := &listeners.Listeners[i]
		for _, action := range listener.DefaultActions {
			if awssdk.ToString(action.TargetGroupArn) != targetGroupARN {
				continue
			}
			return remoteLoadBalancerListener(listener), nil
		}
	}
	return nil, fmt.Errorf("load balancer listener forwarding to target group %s was not found", targetGroupARN)
}

// deleteSessionPublicRecords removes the session's records from the shared
// cp9-dns zone: the exposure record itself plus everything nested under it —
// the Dedicated regional stack writes per-cluster public endpoint records as
// children of <session>.<zone> (#703). Records are untaggable, so cleanup is
// a name-scoped sweep; the shared zone's other names are never touched.
func deleteSessionPublicRecords(
	ctx context.Context,
	dnsClient route53API,
	zoneID string,
	baseName string,
	sessionName string,
) error {
	base := strings.TrimSuffix(strings.TrimSpace(baseName), ".") + "."
	childSuffix := "." + base
	var changes []route53types.Change
	startName := awssdk.String(base)
	var startType route53types.RRType
	const maxPages = 20
	for range maxPages {
		input := &route53.ListResourceRecordSetsInput{
			HostedZoneId:    awssdk.String(zoneID),
			StartRecordName: startName,
		}
		if startType != "" {
			input.StartRecordType = startType
		}
		out, err := dnsClient.ListResourceRecordSets(ctx, input)
		if err != nil {
			return fmt.Errorf("list session public records under %s: %w", base, err)
		}
		done := false
		for i := range out.ResourceRecordSets {
			record := out.ResourceRecordSets[i]
			name := awssdk.ToString(record.Name)
			if name != base && !strings.HasSuffix(name, childSuffix) {
				// Route53 lists in DNS order: once past the session subtree,
				// nothing later can match.
				done = true
				break
			}
			if record.Type == route53types.RRTypeNs || record.Type == route53types.RRTypeSoa {
				continue
			}
			changes = append(changes, route53types.Change{
				Action:            route53types.ChangeActionDelete,
				ResourceRecordSet: &record,
			})
		}
		if done || !out.IsTruncated {
			break
		}
		startName = out.NextRecordName
		startType = out.NextRecordType
	}
	if len(changes) == 0 {
		return nil
	}
	_, err := dnsClient.ChangeResourceRecordSets(ctx, &route53.ChangeResourceRecordSetsInput{
		HostedZoneId: awssdk.String(zoneID),
		ChangeBatch: &route53types.ChangeBatch{
			Comment: awssdk.String("cp9 " + sessionName + " public exposure teardown"),
			Changes: changes,
		},
	})
	// InvalidChangeBatch means records are already gone (or drifted); the
	// zone itself is shared, so treat that as already-clean.
	if err != nil && !isAWSAPIErrorCode(err, "InvalidChangeBatch") {
		return fmt.Errorf("delete session public records under %s: %w", base, err)
	}
	return nil
}

func changePublicHTTPSRecord(
	ctx context.Context,
	dnsClient route53API,
	action route53types.ChangeAction,
	zoneID string,
	recordName string,
	loadBalancerDNS string,
	comment string,
) error {
	_, err := dnsClient.ChangeResourceRecordSets(ctx, &route53.ChangeResourceRecordSetsInput{
		HostedZoneId: awssdk.String(zoneID),
		ChangeBatch: &route53types.ChangeBatch{
			Comment: awssdk.String(comment),
			Changes: []route53types.Change{{
				Action: action,
				ResourceRecordSet: &route53types.ResourceRecordSet{
					Name: awssdk.String(recordName),
					Type: route53types.RRTypeCname,
					TTL:  awssdk.Int64(60),
					ResourceRecords: []route53types.ResourceRecord{{
						Value: awssdk.String(loadBalancerDNS),
					}},
				},
			}},
		},
	})
	return err
}

func changeTXTRecord(
	ctx context.Context,
	dnsClient route53API,
	action route53types.ChangeAction,
	zoneID string,
	recordName string,
	value string,
	comment string,
) error {
	_, err := dnsClient.ChangeResourceRecordSets(ctx, &route53.ChangeResourceRecordSetsInput{
		HostedZoneId: awssdk.String(zoneID),
		ChangeBatch: &route53types.ChangeBatch{
			Comment: awssdk.String(comment),
			Changes: []route53types.Change{{
				Action: action,
				ResourceRecordSet: &route53types.ResourceRecordSet{
					Name: awssdk.String(recordName),
					Type: route53types.RRTypeTxt,
					TTL:  awssdk.Int64(60),
					ResourceRecords: []route53types.ResourceRecord{{
						Value: awssdk.String(strconv.Quote(value)),
					}},
				},
			}},
		},
	})
	return err
}

func remoteLoadBalancerListener(listener *elbv2types.Listener) *host.RemoteLoadBalancerListener {
	if listener == nil {
		return nil
	}
	result := &host.RemoteLoadBalancerListener{
		ARN:      awssdk.ToString(listener.ListenerArn),
		Port:     awssdk.ToInt32(listener.Port),
		Protocol: string(listener.Protocol),
	}
	for _, action := range listener.DefaultActions {
		if strings.TrimSpace(awssdk.ToString(action.TargetGroupArn)) != "" {
			result.TargetGroupARN = awssdk.ToString(action.TargetGroupArn)
			break
		}
	}
	return result
}

func (a *Adapter) destroyPublicLoadBalancer(ctx context.Context, client elbv2API, selected *host.Session) error {
	if selected.Remote.PublicLoadBalancerARN != "" {
		_, err := client.DeleteLoadBalancer(ctx, &elbv2.DeleteLoadBalancerInput{
			LoadBalancerArn: awssdk.String(selected.Remote.PublicLoadBalancerARN),
		})
		if err != nil && !isAWSAPIErrorCode(err, "LoadBalancerNotFound") {
			return fmt.Errorf("delete remote public load balancer: %w", err)
		}
	}
	if selected.Remote.PublicLoadBalancerTargetGroup != "" {
		return deleteTargetGroupWithRetry(ctx, client, selected.Remote.PublicLoadBalancerTargetGroup)
	}
	return nil
}

func (a *Adapter) destroyO11YLoadBalancer(ctx context.Context, client elbv2API, selected *host.Session) error {
	if selected.Remote.O11YLoadBalancerARN != "" {
		_, err := client.DeleteLoadBalancer(ctx, &elbv2.DeleteLoadBalancerInput{
			LoadBalancerArn: awssdk.String(selected.Remote.O11YLoadBalancerARN),
		})
		if err != nil && !isAWSAPIErrorCode(err, "LoadBalancerNotFound") {
			return fmt.Errorf("delete O11Y load balancer: %w", err)
		}
	}
	if selected.Remote.O11YLoadBalancerTargetGroup != "" {
		return deleteTargetGroupWithRetry(ctx, client, selected.Remote.O11YLoadBalancerTargetGroup)
	}
	return nil
}

func (a *Adapter) destroyO11YEndpointService(ctx context.Context, client ec2API, selected *host.Session) error {
	serviceID := strings.TrimSpace(selected.Remote.O11YEndpointServiceID)
	if serviceID == "" {
		return nil
	}
	out, err := client.DeleteVpcEndpointServiceConfigurations(
		ctx,
		&ec2.DeleteVpcEndpointServiceConfigurationsInput{ServiceIds: []string{serviceID}},
	)
	if err != nil {
		if isAWSAPIErrorCode(
			err,
			"InvalidVpcEndpointServiceId.NotFound",
			"InvalidVpcEndpointServiceID.NotFound",
			"InvalidVpcEndpointService.NotFound",
		) {
			return nil
		}
		return fmt.Errorf("delete O11Y endpoint service %s: %w", serviceID, err)
	}
	if out == nil {
		return fmt.Errorf("delete O11Y endpoint service %s returned no result", serviceID)
	}
	if len(out.Unsuccessful) > 0 {
		message := "unknown AWS error"
		if out.Unsuccessful[0].Error != nil {
			message = awssdk.ToString(out.Unsuccessful[0].Error.Message)
		}
		return fmt.Errorf("delete O11Y endpoint service %s: %s", serviceID, message)
	}
	return nil
}

func deleteTargetGroupWithRetry(ctx context.Context, client elbv2API, targetGroupARN string) error {
	deadline := time.Now().Add(2 * time.Minute)
	for {
		_, err := client.DeleteTargetGroup(ctx, &elbv2.DeleteTargetGroupInput{
			TargetGroupArn: awssdk.String(targetGroupARN),
		})
		if err == nil || isAWSAPIErrorCode(err, "TargetGroupNotFound") {
			return nil
		}
		if !isAWSAPIErrorCode(err, "ResourceInUse") || time.Now().After(deadline) {
			return fmt.Errorf("delete remote public target group: %w", err)
		}
		if err := sleepWithContext(ctx, 5*time.Second); err != nil {
			return fmt.Errorf("delete remote public target group: %w", err)
		}
	}
}

// destroyPublicHTTPSExposure removes the per-session Route53 record, the
// HTTPS ALB and its target group. The account-level hosted zone and runtime
// account's bootstrap wildcard certificate are never touched here.
func (a *Adapter) destroyPublicHTTPSExposure(
	ctx context.Context,
	lbClient elbv2API,
	dnsClient route53API,
	selected *host.Session,
) error {
	remote := selected.Remote
	if remote.PublicHTTPSRecordName != "" && remote.PublicHTTPSRecordZoneID != "" {
		if err := deleteSessionPublicRecords(
			ctx,
			dnsClient,
			remote.PublicHTTPSRecordZoneID,
			remote.PublicHTTPSRecordName,
			selected.ClusterName,
		); err != nil {
			return err
		}
	}
	if remote.PublicHTTPSLoadBalancerARN != "" {
		_, err := lbClient.DeleteLoadBalancer(ctx, &elbv2.DeleteLoadBalancerInput{
			LoadBalancerArn: awssdk.String(remote.PublicHTTPSLoadBalancerARN),
		})
		if err != nil && !isAWSAPIErrorCode(err, "LoadBalancerNotFound") {
			return fmt.Errorf("delete public HTTPS load balancer: %w", err)
		}
	}
	if remote.PublicHTTPSTargetGroup != "" {
		return deleteTargetGroupWithRetry(ctx, lbClient, remote.PublicHTTPSTargetGroup)
	}
	return nil
}

// refreshInstanceAddress re-reads a recorded host's public address and persists it
// when EC2 has moved it. A reused instance gets a new public IP after a stop or
// start, which leaves the recorded address wrong but not empty, so the emptiness
// check above never fires and every later step keeps using the old value. What the
// operator sees is SSH hanging against an address nothing answers on, which reads
// as a broken tunnel rather than a stale record, and repeated repair runs cannot
// fix it because none of them looks the address up again.
//
// This is one describe rather than waitForInstanceReady's polling loop: the host is
// already known, so an instance that is not currently running should surface at the
// SSH boundary as before instead of blocking here for the ready timeout.
func (s *ClusterStep) refreshInstanceAddress(ctx context.Context, client ec2API, instanceID string) error {
	out, err := client.DescribeInstances(ctx, &ec2.DescribeInstancesInput{
		InstanceIds: []string{instanceID},
	})
	if err != nil {
		if isAWSAPIErrorCode(err, "InvalidInstanceID.NotFound") {
			return nil
		}
		return fmt.Errorf("describe remote EC2 host %s: %w", instanceID, err)
	}
	for _, reservation := range out.Reservations {
		for idx := range reservation.Instances {
			instance := &reservation.Instances[idx]
			if awssdk.ToString(instance.InstanceId) != instanceID {
				continue
			}
			publicDNS := awssdk.ToString(instance.PublicDnsName)
			publicIP := awssdk.ToString(instance.PublicIpAddress)
			// A stopped instance reports no address at all. Keep the recorded one
			// rather than blanking it, so the failure still names SSH.
			if publicDNS == "" && publicIP == "" {
				return nil
			}
			if publicDNS == s.Selected.Remote.PublicDNS && publicIP == s.Selected.Remote.PublicIP {
				return nil
			}
			s.Selected.Remote.PublicDNS = publicDNS
			s.Selected.Remote.PublicIP = publicIP
			return s.persistSelected()
		}
	}
	return nil
}

func (s *ClusterStep) waitForInstanceReady(ctx context.Context, client ec2API, instanceID string) error {
	deadline := time.Now().Add(5 * time.Minute)
	for time.Now().Before(deadline) {
		out, err := client.DescribeInstances(ctx, &ec2.DescribeInstancesInput{
			InstanceIds: []string{instanceID},
		})
		if err != nil {
			if !isAWSAPIErrorCode(err, "InvalidInstanceID.NotFound") {
				return fmt.Errorf("describe remote EC2 host %s: %w", instanceID, err)
			}
			out = &ec2.DescribeInstancesOutput{}
		}
		for _, reservation := range out.Reservations {
			for idx := range reservation.Instances {
				instance := &reservation.Instances[idx]
				if awssdk.ToString(instance.InstanceId) != instanceID {
					continue
				}
				if instance.State != nil && instance.State.Name == ec2types.InstanceStateNameRunning &&
					(awssdk.ToString(instance.PublicDnsName) != "" || awssdk.ToString(instance.PublicIpAddress) != "") {
					s.Selected.Remote.PublicDNS = awssdk.ToString(instance.PublicDnsName)
					s.Selected.Remote.PublicIP = awssdk.ToString(instance.PublicIpAddress)
					return nil
				}
			}
		}
		time.Sleep(5 * time.Second)
	}
	return fmt.Errorf("remote EC2 host %s did not become ready before timeout", instanceID)
}

func tagSpec(resourceType ec2types.ResourceType, sessionName, owner, ttlExpiresAt string) ec2types.TagSpecification {
	tags := []ec2types.Tag{
		{Key: awssdk.String(cpaws.TagManaged), Value: awssdk.String("true")},
		{Key: awssdk.String(cpaws.TagSession), Value: awssdk.String(sessionName)},
		{Key: awssdk.String(cpaws.TagOwner), Value: awssdk.String(owner)},
		{Key: awssdk.String("Name"), Value: awssdk.String(cpaws.SessionHostNameTag(sessionName))},
	}
	if strings.TrimSpace(ttlExpiresAt) != "" {
		tags = append(tags, ec2types.Tag{Key: awssdk.String(cpaws.TagTTL), Value: awssdk.String(ttlExpiresAt)})
	}
	return ec2types.TagSpecification{
		ResourceType: resourceType,
		Tags:         tags,
	}
}

func endpointServiceTagSpec(sessionName, owner, ttlExpiresAt string) ec2types.TagSpecification {
	return tagSpec(ec2types.ResourceTypeVpcEndpointService, sessionName, owner, ttlExpiresAt)
}

func endpointServiceTagsMatch(tags []ec2types.Tag, sessionName, owner string) bool {
	values := make(map[string]string, len(tags))
	for _, tag := range tags {
		values[awssdk.ToString(tag.Key)] = awssdk.ToString(tag.Value)
	}
	return values[cpaws.TagSession] == sessionName &&
		values[cpaws.TagOwner] == owner &&
		values[cpaws.TagManaged] == "true" &&
		strings.TrimSpace(values[cpaws.TagTTL]) != ""
}

func elbv2Tags(sessionName, owner, ttlExpiresAt string) []elbv2types.Tag {
	tags := []elbv2types.Tag{
		{Key: awssdk.String(cpaws.TagManaged), Value: awssdk.String("true")},
		{Key: awssdk.String(cpaws.TagSession), Value: awssdk.String(sessionName)},
		{Key: awssdk.String(cpaws.TagOwner), Value: awssdk.String(owner)},
		{Key: awssdk.String("Name"), Value: awssdk.String("cp9-" + sessionName)},
	}
	if strings.TrimSpace(ttlExpiresAt) != "" {
		tags = append(tags, elbv2types.Tag{
			Key:   awssdk.String(cpaws.TagTTL),
			Value: awssdk.String(ttlExpiresAt),
		})
	}
	return tags
}

func elbv2TTLExpiresAt(ttl time.Duration) string {
	if ttl <= 0 {
		return ""
	}
	return time.Now().UTC().Add(ttl).Format(time.RFC3339)
}

func isAWSAPIErrorCode(err error, codes ...string) bool {
	var apiErr smithy.APIError
	if !errors.As(err, &apiErr) {
		return false
	}
	got := apiErr.ErrorCode()
	return slices.Contains(codes, got)
}

func remoteLoadBalancerName(sessionName string) string {
	return remoteAWSName("cp9", sessionName)
}

func remoteO11YLoadBalancerName(sessionName string) string {
	return remoteAWSName("cp9-olb", sessionName)
}

func remoteEndpointServiceClientToken(sessionName string) string {
	return remoteAWSName("cp9-o11y", sessionName)
}

func remoteTargetGroupName(sessionName string) string {
	return remoteAWSName("cp9-tg", sessionName)
}

func remoteO11YTargetGroupName(sessionName string) string {
	return remoteAWSName("cp9-otg", sessionName)
}

func remoteHTTPSLoadBalancerName(sessionName string) string {
	return remoteAWSName("cp9-alb", sessionName)
}

func remoteHTTPSTargetGroupName(sessionName string) string {
	return remoteAWSName("cp9-atg", sessionName)
}

func remoteAWSName(prefix, sessionName string) string {
	// ELBv2 load-balancer and target-group names cap at 32 characters.
	const limit = 32
	hasher := fnv.New32a()
	_, _ = hasher.Write([]byte(sessionName))
	sum := hasher.Sum32()
	suffixBytes := []byte{
		byte(sum >> 24),
		byte(sum >> 16),
		byte(sum >> 8),
		byte(sum),
	}
	suffix := hex.EncodeToString(suffixBytes)
	baseLimit := max(limit-len(prefix)-len(suffix)-2, 1)
	base := sanitizeAWSName(sessionName)
	if len(base) > baseLimit {
		base = base[:baseLimit]
	}
	base = strings.Trim(base, "-")
	if base == "" {
		base = "session"
	}
	return prefix + "-" + base + "-" + suffix
}

func sanitizeAWSName(value string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	out := make([]byte, 0, len(value))
	for _, r := range value {
		switch {
		case r >= 'a' && r <= 'z':
			out = append(out, byte(r))
		case r >= '0' && r <= '9':
			out = append(out, byte(r))
		case r == '-':
			out = append(out, '-')
		default:
			out = append(out, '-')
		}
	}
	return string(out)
}

func iamInstanceProfile(name string) *ec2types.IamInstanceProfileSpecification {
	name = strings.TrimSpace(name)
	if name == "" {
		return nil
	}
	return &ec2types.IamInstanceProfileSpecification{Name: awssdk.String(name)}
}

func isDuplicateSecurityGroupIngress(err error) bool {
	if err == nil {
		return false
	}
	text := err.Error()
	return strings.Contains(text, "InvalidPermission.Duplicate") ||
		strings.Contains(text, "already exists")
}

func hostShellCmd(argv ...string) string {
	return host.ShellEscapeArgs(argv...)
}

func sleepWithContext(ctx context.Context, delay time.Duration) error {
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}