package aws

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"regexp"
	"slices"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/autoscaling"
	autoscalingtypes "github.com/aws/aws-sdk-go-v2/service/autoscaling/types"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go-v2/service/eks"
	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancing"
	elbtypes "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancing/types"
	"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/iam"
	iamtypes "github.com/aws/aws-sdk-go-v2/service/iam/types"
	"github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi"
	tagtypes "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi/types"

	"github.com/tidbcloud/controlplane9/internal/verbose"
)

const elbChildLookupTimeout = 30 * time.Second

var (
	staleTaggedEC2ResourceLookupTimeout = 2 * time.Second
	deterministicIAMLookupTimeout       = 5 * time.Second
	// taggedIAMLookupTimeout budgets the account-wide tagged-IAM scan, which issues
	// one ListRoleTags (and one ListInstanceProfileTags) call per IAM entity in the
	// account. That is a fundamentally different shape from the deterministic lookup
	// beside it, which resolves a handful of known names: in an account with hundreds
	// of roles the scan cannot finish in seconds, so it is budgeted like the other
	// heavy graph lookup rather than like a targeted one.
	taggedIAMLookupTimeout       = 2 * time.Minute
	trustedEKSGraphLookupTimeout = 2 * time.Minute
	ec2SingleAttemptOptions      = []func(*ec2.Options){
		func(o *ec2.Options) {
			o.RetryMaxAttempts = 1
		},
	}
	kubernetesSafeTagValueReplacer = regexp.MustCompile(`[^A-Za-z0-9_.-]+`)
	awsSafeTagValueReplacer        = regexp.MustCompile(`[^A-Za-z0-9-]+`)
)

// TaggingAPI is the subset of the Resource Groups Tagging API used by the tracker.
type TaggingAPI interface {
	GetResources(ctx context.Context, params *resourcegroupstaggingapi.GetResourcesInput,
		optFns ...func(*resourcegroupstaggingapi.Options)) (*resourcegroupstaggingapi.GetResourcesOutput, error)
}

// IAMTagAPI is the subset of IAM used to recover deterministic cp9 workload
// role/policy resources that the regional Tagging API does not reliably return.
type IAMTagAPI interface {
	GetRole(ctx context.Context, params *iam.GetRoleInput, optFns ...func(*iam.Options)) (*iam.GetRoleOutput, error)
	ListRoleTags(ctx context.Context, params *iam.ListRoleTagsInput,
		optFns ...func(*iam.Options)) (*iam.ListRoleTagsOutput, error)
	GetPolicy(ctx context.Context, params *iam.GetPolicyInput, optFns ...func(*iam.Options)) (*iam.GetPolicyOutput, error)
	ListPolicyTags(ctx context.Context, params *iam.ListPolicyTagsInput,
		optFns ...func(*iam.Options)) (*iam.ListPolicyTagsOutput, error)
}

type IAMListTagAPI interface {
	ListRoles(ctx context.Context, params *iam.ListRolesInput, optFns ...func(*iam.Options)) (*iam.ListRolesOutput, error)
	ListInstanceProfiles(ctx context.Context, params *iam.ListInstanceProfilesInput,
		optFns ...func(*iam.Options)) (*iam.ListInstanceProfilesOutput, error)
	ListPolicies(ctx context.Context, params *iam.ListPoliciesInput, optFns ...func(*iam.Options)) (*iam.ListPoliciesOutput, error)
	ListAttachedRolePolicies(ctx context.Context, params *iam.ListAttachedRolePoliciesInput,
		optFns ...func(*iam.Options)) (*iam.ListAttachedRolePoliciesOutput, error)
}

type iamInstanceProfileProofAPI interface {
	GetInstanceProfile(ctx context.Context, params *iam.GetInstanceProfileInput,
		optFns ...func(*iam.Options)) (*iam.GetInstanceProfileOutput, error)
}

// EC2ResourceGraphAPI is the direct EC2 discovery surface used to recover
// tag-complete VPC children that are not reliably returned by Resource Groups
// Tagging API.
type EC2ResourceGraphAPI interface {
	DescribeSubnets(ctx context.Context, params *ec2.DescribeSubnetsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeSubnetsOutput, error)
	DescribeRouteTables(ctx context.Context, params *ec2.DescribeRouteTablesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeRouteTablesOutput, error)
	DescribeInternetGateways(ctx context.Context, params *ec2.DescribeInternetGatewaysInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeInternetGatewaysOutput, error)
	DescribeNatGateways(ctx context.Context, params *ec2.DescribeNatGatewaysInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeNatGatewaysOutput, error)
	DescribeVpcEndpoints(ctx context.Context, params *ec2.DescribeVpcEndpointsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeVpcEndpointsOutput, error)
	DescribeVpcEndpointServiceConfigurations(ctx context.Context, params *ec2.DescribeVpcEndpointServiceConfigurationsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeVpcEndpointServiceConfigurationsOutput, error)
	DescribeNetworkInterfaces(ctx context.Context, params *ec2.DescribeNetworkInterfacesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeNetworkInterfacesOutput, error)
	DescribeSecurityGroups(ctx context.Context, params *ec2.DescribeSecurityGroupsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeSecurityGroupsOutput, error)
	DescribeInstances(ctx context.Context, params *ec2.DescribeInstancesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeInstancesOutput, error)
	DescribeVolumes(ctx context.Context, params *ec2.DescribeVolumesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeVolumesOutput, error)
	DescribeSnapshots(ctx context.Context, params *ec2.DescribeSnapshotsInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeSnapshotsOutput, error)
	DescribeAddresses(ctx context.Context, params *ec2.DescribeAddressesInput,
		optFns ...func(*ec2.Options)) (*ec2.DescribeAddressesOutput, error)
}

// ELBResourceGraphAPI is the direct ELBv2 discovery surface used to recover
// tag-complete Kubernetes load balancers and target groups inside a tagged VPC.
type ELBResourceGraphAPI interface {
	DescribeLoadBalancers(ctx context.Context, params *elasticloadbalancingv2.DescribeLoadBalancersInput,
		optFns ...func(*elasticloadbalancingv2.Options)) (*elasticloadbalancingv2.DescribeLoadBalancersOutput, error)
	DescribeTargetGroups(ctx context.Context, params *elasticloadbalancingv2.DescribeTargetGroupsInput,
		optFns ...func(*elasticloadbalancingv2.Options)) (*elasticloadbalancingv2.DescribeTargetGroupsOutput, error)
	DescribeTags(ctx context.Context, params *elasticloadbalancingv2.DescribeTagsInput,
		optFns ...func(*elasticloadbalancingv2.Options)) (*elasticloadbalancingv2.DescribeTagsOutput, error)
}

// ClassicELBResourceGraphAPI is the direct Classic ELB discovery surface used
// to recover tag-complete Kubernetes classic load balancers inside a tagged VPC.
type ClassicELBResourceGraphAPI interface {
	DescribeLoadBalancers(ctx context.Context, params *elasticloadbalancing.DescribeLoadBalancersInput,
		optFns ...func(*elasticloadbalancing.Options)) (*elasticloadbalancing.DescribeLoadBalancersOutput, error)
	DescribeTags(ctx context.Context, params *elasticloadbalancing.DescribeTagsInput,
		optFns ...func(*elasticloadbalancing.Options)) (*elasticloadbalancing.DescribeTagsOutput, error)
}

// AutoScalingGraphAPI recovers tagged Auto Scaling groups that Resource Groups
// Tagging API does not return even when DescribeAutoScalingGroups has the tags.
type AutoScalingGraphAPI interface {
	DescribeAutoScalingGroups(
		ctx context.Context,
		params *autoscaling.DescribeAutoScalingGroupsInput,
		optFns ...func(*autoscaling.Options),
	) (*autoscaling.DescribeAutoScalingGroupsOutput, error)
}

// EKSResourceGraphAPI is the direct EKS discovery surface used to recover
// clusters and nodegroups from a tag-verified deterministic CP9 service role.
// A cluster's RoleArn and a nodegroup's ClusterName are immutable containment
// edges, so this graph remains authoritative while the regional tag index is
// delayed after cleanup-tag assurance.
type EKSResourceGraphAPI interface {
	ListClusters(ctx context.Context, params *eks.ListClustersInput,
		optFns ...func(*eks.Options)) (*eks.ListClustersOutput, error)
	DescribeCluster(ctx context.Context, params *eks.DescribeClusterInput,
		optFns ...func(*eks.Options)) (*eks.DescribeClusterOutput, error)
	ListTagsForResource(ctx context.Context, params *eks.ListTagsForResourceInput,
		optFns ...func(*eks.Options)) (*eks.ListTagsForResourceOutput, error)
	ListNodegroups(ctx context.Context, params *eks.ListNodegroupsInput,
		optFns ...func(*eks.Options)) (*eks.ListNodegroupsOutput, error)
	DescribeNodegroup(ctx context.Context, params *eks.DescribeNodegroupInput,
		optFns ...func(*eks.Options)) (*eks.DescribeNodegroupOutput, error)
}

// EKSRoleProofAPI is the IAM read surface used to prove the deterministic EKS
// service role independently of Resource Groups Tagging API results.
type EKSRoleProofAPI interface {
	GetRole(ctx context.Context, params *iam.GetRoleInput,
		optFns ...func(*iam.Options)) (*iam.GetRoleOutput, error)
	ListRoleTags(ctx context.Context, params *iam.ListRoleTagsInput,
		optFns ...func(*iam.Options)) (*iam.ListRoleTagsOutput, error)
}

// TrustedEKSRoleGraph is a read-only EKS graph proven from one deterministic,
// owner/session-scoped service role. ParentByARN records the immutable
// relationship used for every discovered resource.
type TrustedEKSRoleGraph struct {
	RoleARN     string
	Resources   []Resource
	ParentByARN map[string]string
}

// ResourceTracker discovers cp9-managed AWS resources via the Tagging API.
type ResourceTracker struct {
	client                     TaggingAPI
	iam                        IAMTagAPI
	ec2                        EC2ResourceGraphAPI
	eks                        EKSResourceGraphAPI
	elbv2                      ELBResourceGraphAPI
	classicELB                 ClassicELBResourceGraphAPI
	autoscaling                AutoScalingGraphAPI
	region                     string
	accountID                  string
	requireTrustedEKSRoleGraph bool
	iamInventory               *IAMInventory
	sessionCollisionOnly       bool
}

// NewTracker creates a ResourceTracker backed by the given Tagging API client.
func NewTracker(client TaggingAPI) *ResourceTracker {
	return &ResourceTracker{client: client}
}

// NewTrackerForRegion creates a ResourceTracker bound to the given region.
func NewTrackerForRegion(client TaggingAPI, region string) *ResourceTracker {
	return &ResourceTracker{client: client, region: region}
}

// NewTrackerWithIAM creates a tracker that supplements Tagging API results with
// deterministic cp9 IAM role/policy discovery for the given region.
func NewTrackerWithIAM(client TaggingAPI, iamClient IAMTagAPI, region string) *ResourceTracker {
	return &ResourceTracker{client: client, iam: iamClient, region: region}
}

// NewTrackerWithIAMForAccount creates a tracker with an account hint for
// deterministic IAM policy lookup when the paired deterministic role is gone.
// Account-scoped destroy verification must also supply direct EKS discovery so
// a missing graph client cannot silently turn trusted-parent proof into zero.
func NewTrackerWithIAMForAccount(
	client TaggingAPI,
	iamClient IAMTagAPI,
	eksClient EKSResourceGraphAPI,
	region string,
	accountID string,
) *ResourceTracker {
	return &ResourceTracker{
		client:                     client,
		iam:                        iamClient,
		eks:                        eksClient,
		region:                     region,
		accountID:                  strings.TrimSpace(accountID),
		requireTrustedEKSRoleGraph: true,
	}
}

// WithEC2ResourceGraph enables direct VPC membership discovery for resources
// that are tag-complete but missing from Resource Groups Tagging API.
func (t *ResourceTracker) WithEC2ResourceGraph(client EC2ResourceGraphAPI) *ResourceTracker {
	t.ec2 = client
	return t
}

// WithIAMInventory serves tagged-IAM discovery from a snapshot of the account's
// IAM taken once by the caller, instead of re-scanning the account for every
// session. The snapshot is discovery-only: deletion still re-reads each entity's
// tags directly, so a stale snapshot cannot widen what gets deleted.
func (t *ResourceTracker) WithIAMInventory(inventory *IAMInventory) *ResourceTracker {
	t.iamInventory = inventory
	return t
}

// WithSessionCollisionDiscovery uses the bounded fresh-name existence scan
// instead of enumerating the session's full cleanup child graph.
func (t *ResourceTracker) WithSessionCollisionDiscovery() *ResourceTracker {
	t.sessionCollisionOnly = true
	return t
}

// WithEKSResourceGraph enables direct EKS relationship discovery after the
// deterministic service role has independently passed owner/session tag proof.
func (t *ResourceTracker) WithEKSResourceGraph(client EKSResourceGraphAPI) *ResourceTracker {
	t.eks = client
	return t
}

// WithELBResourceGraph enables direct VPC membership discovery for ELB resources
// that are tag-complete but missing from Resource Groups Tagging API.
func (t *ResourceTracker) WithELBResourceGraph(client ELBResourceGraphAPI) *ResourceTracker {
	t.elbv2 = client
	return t
}

// WithClassicELBResourceGraph enables direct VPC membership discovery for
// Classic ELB resources that are tag-complete but missing from Resource Groups
// Tagging API.
func (t *ResourceTracker) WithClassicELBResourceGraph(client ClassicELBResourceGraphAPI) *ResourceTracker {
	t.classicELB = client
	return t
}

// WithAutoScalingGraph enables direct Auto Scaling group discovery. Tagged
// groups are missing from Resource Groups Tagging API after their VPC is gone,
// so the tag index cannot be the only finder.
func (t *ResourceTracker) WithAutoScalingGraph(client AutoScalingGraphAPI) *ResourceTracker {
	t.autoscaling = client
	return t
}

// FindBySession returns all resources tagged with cp9-session=<sessionID>.
func (t *ResourceTracker) FindBySession(ctx context.Context, sessionID string) ([]Resource, error) {
	if t.sessionCollisionOnly {
		return t.findSessionCollision(ctx, sessionID)
	}
	resources, err := t.findByTags(ctx, []tagtypes.TagFilter{
		{Key: aws.String(TagSession), Values: []string{sessionID}},
	})
	if err != nil {
		return nil, err
	}
	fallback, err := t.findCP9ManagedFallback(ctx, "", sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, fallback)
	environmentFallback, err := t.findEnvironmentSessionNameFallback(ctx, "", sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, environmentFallback)
	vpcChildren, err := t.findVPCMembership(ctx, "", sessionID, resources)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, vpcChildren)
	iamResources, err := t.findDeterministicIAM(ctx, "", sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, iamResources)
	taggedIAMResources, err := t.findTaggedIAM(ctx, "", sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, taggedIAMResources)
	asgResources, err := t.findTaggedAutoScalingGroups(ctx, "", sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, asgResources)
	eksResources, err := t.findTrustedEKSRoleChildren(ctx, "", sessionID)
	if err != nil {
		return nil, err
	}
	return mergeTrackerResources(resources, eksResources), nil
}

// findSessionCollision proves whether any current CP9 resource can already own
// a fresh session name. Unlike cleanup, it does not enumerate the whole child
// graph: one live root is enough to reject the name. Independent roots are all
// collected so a stale tag-index row cannot hide a live fallback resource.
// IAM and EKS discovery narrows by CP9's session-prefixed name contract,
// avoiding a tag read and graph walk for every entity in the account.
func (t *ResourceTracker) findSessionCollision(ctx context.Context, sessionID string) ([]Resource, error) {
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		return nil, nil
	}
	resources, err := t.findByTags(ctx, []tagtypes.TagFilter{
		{Key: aws.String(TagSession), Values: []string{sessionID}},
	})
	if err != nil {
		return nil, err
	}

	for _, find := range []func(context.Context, string, string) ([]Resource, error){
		t.findDeterministicIAM,
		t.findSessionNamedIAM,
		t.findTaggedAutoScalingGroups,
		t.findCP9ManagedFallback,
		t.findEnvironmentSessionNameFallback,
	} {
		found, findErr := find(ctx, "", sessionID)
		if findErr != nil {
			return nil, findErr
		}
		resources = mergeTrackerResources(resources, found)
	}

	found, err := t.findSessionNamedEKSClusters(ctx, sessionID)
	if err != nil {
		return nil, err
	}
	return mergeTrackerResources(resources, found), nil
}

// FindByOwner returns all resources tagged with cp9-owner=<owner>.
func (t *ResourceTracker) FindByOwner(ctx context.Context, owner string) ([]Resource, error) {
	return t.findByTags(ctx, []tagtypes.TagFilter{
		{Key: aws.String(TagOwner), Values: ownerTagFilterValues(owner)},
	})
}

// FindByOwnerAndSession returns all resources matching both owner and session tags.
func (t *ResourceTracker) FindByOwnerAndSession(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	resources, err := t.findByTags(ctx, []tagtypes.TagFilter{
		{Key: aws.String(TagOwner), Values: ownerTagFilterValues(owner)},
		{Key: aws.String(TagSession), Values: []string{sessionID}},
	})
	if err != nil {
		return nil, err
	}
	fallback, err := t.findCP9ManagedFallback(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, fallback)
	environmentFallback, err := t.findEnvironmentSessionNameFallback(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, environmentFallback)
	kubernetesFallback, err := t.findKubernetesClusterNameFallback(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, kubernetesFallback)
	vpcChildren, err := t.findVPCMembership(ctx, owner, sessionID, resources)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, vpcChildren)
	iamResources, err := t.findDeterministicIAM(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, iamResources)
	taggedIAMResources, err := t.findTaggedIAM(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, taggedIAMResources)
	asgResources, err := t.findTaggedAutoScalingGroups(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	resources = mergeTrackerResources(resources, asgResources)
	eksResources, err := t.findTrustedEKSRoleChildren(ctx, owner, sessionID)
	if err != nil {
		return nil, err
	}
	return mergeTrackerResources(resources, eksResources), nil
}

// FindAll returns all resources tagged with cp9-session (any value).
func (t *ResourceTracker) FindAll(ctx context.Context) ([]Resource, error) {
	return t.findByTags(ctx, []tagtypes.TagFilter{
		{Key: aws.String(TagSession)},
	})
}

// SupportsKnownResourceVerification reports whether a resource type has an
// exact owning-service lookup for terminal destroy-handoff proof.
func SupportsKnownResourceVerification(resourceType ResourceType) bool {
	switch resourceType {
	case ResourceEC2Instance,
		ResourceSecurityGroup,
		ResourceIAMRole,
		ResourceIAMInstanceProfile,
		ResourceLoadBalancer,
		ResourceTargetGroup,
		ResourceVPCEndpointService,
		ResourceEBSVolume,
		ResourceEBSSnapshot,
		ResourceENI,
		ResourceVPCEndpoint,
		ResourceNATGateway,
		ResourceEIP:
		return true
	default:
		return false
	}
}

// VerifyKnownResource performs an exact, read-only service lookup for the
// resource types written to a Dev full destroy handoff independently of the
// eventually-consistent Resource Groups Tagging API index.
func (t *ResourceTracker) VerifyKnownResource(
	ctx context.Context,
	scope Scope,
	resource *Resource,
) (bool, error) {
	if resource == nil {
		return false, fmt.Errorf("known resource is nil")
	}
	scope = scope.Normalize()
	if err := scope.Validate(); err != nil {
		return false, err
	}
	if strings.TrimSpace(resource.ARN) == "" {
		return false, fmt.Errorf("known resource ARN is empty")
	}
	var (
		tags   map[string]string
		exists bool
		err    error
	)
	switch resource.Type {
	case ResourceEC2Instance:
		tags, exists, err = t.knownEC2InstanceTags(ctx, resource.ARN)
	case ResourceSecurityGroup:
		tags, exists, err = t.knownSecurityGroupTags(ctx, resource.ARN)
	case ResourceIAMRole:
		tags, exists, err = t.knownIAMRoleTags(ctx, resource.ARN)
	case ResourceIAMInstanceProfile:
		tags, exists, err = t.knownIAMInstanceProfileTags(ctx, resource.ARN)
	case ResourceLoadBalancer, ResourceTargetGroup:
		tags, exists, err = t.knownELBV2Tags(ctx, resource)
	case ResourceVPCEndpointService:
		tags, exists, err = t.knownVPCEndpointServiceTags(ctx, resource.ARN)
	case ResourceEBSVolume, ResourceEBSSnapshot, ResourceENI, ResourceVPCEndpoint, ResourceNATGateway, ResourceEIP:
		tags, exists, err = t.knownTaggedEC2ResourceTags(ctx, resource.Type, resource.ARN)
	default:
		return false, fmt.Errorf(
			"known resource %q type %q has no direct ownership verifier",
			resource.ARN,
			resource.Type,
		)
	}
	if err != nil {
		return false, err
	}
	if !exists {
		return false, nil
	}
	if !tagsMatchDeleteScope(tags, scope.Owner, scope.SessionID) {
		return false, fmt.Errorf(
			"known resource %q exists without exact owner/session/ttl proof",
			resource.ARN,
		)
	}
	return true, nil
}

func (t *ResourceTracker) knownEC2InstanceTags(ctx context.Context, arn string) (map[string]string, bool, error) {
	if t.ec2 == nil {
		return nil, false, fmt.Errorf("direct EC2 ownership verifier is not configured")
	}
	id := ec2IDFromARN(arn, "instance/")
	if id == "" {
		return nil, false, fmt.Errorf("invalid EC2 instance ARN %q", arn)
	}
	out, err := t.ec2.DescribeInstances(ctx, &ec2.DescribeInstancesInput{InstanceIds: []string{id}}, ec2SingleAttemptOptions...)
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("describe known EC2 instance %s: %w", id, err)
	}
	for i := range out.Reservations {
		for j := range out.Reservations[i].Instances {
			instance := out.Reservations[i].Instances[j]
			if aws.ToString(instance.InstanceId) != id ||
				(instance.State != nil && instance.State.Name == ec2types.InstanceStateNameTerminated) {
				continue
			}
			return ec2TagsToMap(instance.Tags), true, nil
		}
	}
	return nil, false, nil
}

func (t *ResourceTracker) knownSecurityGroupTags(ctx context.Context, arn string) (map[string]string, bool, error) {
	if t.ec2 == nil {
		return nil, false, fmt.Errorf("direct EC2 ownership verifier is not configured")
	}
	id := ec2IDFromARN(arn, "security-group/")
	if id == "" {
		return nil, false, fmt.Errorf("invalid security group ARN %q", arn)
	}
	out, err := t.ec2.DescribeSecurityGroups(ctx, &ec2.DescribeSecurityGroupsInput{GroupIds: []string{id}}, ec2SingleAttemptOptions...)
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("describe known security group %s: %w", id, err)
	}
	for i := range out.SecurityGroups {
		if aws.ToString(out.SecurityGroups[i].GroupId) == id {
			return ec2TagsToMap(out.SecurityGroups[i].Tags), true, nil
		}
	}
	return nil, false, nil
}

// knownTaggedEC2ResourceTags proves ownership of the remaining EC2 survivor
// types through one exact, single-attempt describe each.
//
// A destroy handoff persists whatever the reaper could not delete, without
// filtering by type, and any type missing from SupportsKnownResourceVerification
// makes the whole handoff fail closed. These six are the EC2 types that
// realistically survive a teardown -- an orphaned volume outlives its instance, a
// detached ENI or a released-but-still-held EIP outlives its subnet -- so each
// needs a verifier of its own rather than only an absence check.
//
// Absence is decided exactly as the stale-index check beside it decides it, so
// the two cannot disagree about whether a resource is gone: an empty result or a
// NotFound is absent, and a NAT gateway is additionally absent once its state is
// terminal, because AWS keeps deleted NAT gateways describable for hours.
func (t *ResourceTracker) knownTaggedEC2ResourceTags(
	ctx context.Context,
	resourceType ResourceType,
	arn string,
) (map[string]string, bool, error) {
	if t.ec2 == nil {
		return nil, false, fmt.Errorf("direct EC2 ownership verifier is not configured")
	}
	prefix, ok := ec2KnownResourceARNPrefixes[resourceType]
	if !ok {
		return nil, false, fmt.Errorf("EC2 type %q has no known-resource ARN form", resourceType)
	}
	id := ec2IDFromARN(arn, prefix)
	if id == "" {
		return nil, false, fmt.Errorf("invalid %s ARN %q", resourceType, arn)
	}
	tags, exists, err := t.describeKnownEC2Resource(ctx, resourceType, id)
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("describe known %s %s: %w", resourceType, id, err)
	}
	return tags, exists, nil
}

// ec2KnownResourceARNPrefixes maps each verified EC2 type to the ARN resource
// prefix its id is carried in, matching what ResourceCleaner parses when it
// deletes the same resource.
var ec2KnownResourceARNPrefixes = map[ResourceType]string{
	ResourceEBSVolume:   "volume/",
	ResourceEBSSnapshot: "snapshot/",
	ResourceENI:         "network-interface/",
	ResourceVPCEndpoint: "vpc-endpoint/",
	ResourceNATGateway:  "natgateway/",
	ResourceEIP:         "elastic-ip/",
}

func (t *ResourceTracker) describeKnownEC2Resource(
	ctx context.Context,
	resourceType ResourceType,
	id string,
) (map[string]string, bool, error) {
	switch resourceType {
	case ResourceEBSVolume:
		out, err := t.ec2.DescribeVolumes(ctx,
			&ec2.DescribeVolumesInput{VolumeIds: []string{id}}, ec2SingleAttemptOptions...)
		if err != nil {
			return nil, false, err
		}
		for i := range out.Volumes {
			if aws.ToString(out.Volumes[i].VolumeId) == id {
				return ec2TagsToMap(out.Volumes[i].Tags), true, nil
			}
		}
	case ResourceEBSSnapshot:
		out, err := t.ec2.DescribeSnapshots(ctx,
			&ec2.DescribeSnapshotsInput{SnapshotIds: []string{id}}, ec2SingleAttemptOptions...)
		if err != nil {
			return nil, false, err
		}
		for i := range out.Snapshots {
			if aws.ToString(out.Snapshots[i].SnapshotId) == id {
				return ec2TagsToMap(out.Snapshots[i].Tags), true, nil
			}
		}
	case ResourceENI:
		out, err := t.ec2.DescribeNetworkInterfaces(ctx,
			&ec2.DescribeNetworkInterfacesInput{NetworkInterfaceIds: []string{id}}, ec2SingleAttemptOptions...)
		if err != nil {
			return nil, false, err
		}
		for i := range out.NetworkInterfaces {
			if aws.ToString(out.NetworkInterfaces[i].NetworkInterfaceId) == id {
				// An ENI carries its tags in TagSet, not Tags.
				return ec2TagsToMap(out.NetworkInterfaces[i].TagSet), true, nil
			}
		}
	case ResourceVPCEndpoint:
		out, err := t.ec2.DescribeVpcEndpoints(ctx,
			&ec2.DescribeVpcEndpointsInput{VpcEndpointIds: []string{id}}, ec2SingleAttemptOptions...)
		if err != nil {
			return nil, false, err
		}
		for i := range out.VpcEndpoints {
			if aws.ToString(out.VpcEndpoints[i].VpcEndpointId) == id {
				return ec2TagsToMap(out.VpcEndpoints[i].Tags), true, nil
			}
		}
	case ResourceNATGateway:
		out, err := t.ec2.DescribeNatGateways(ctx,
			&ec2.DescribeNatGatewaysInput{NatGatewayIds: []string{id}}, ec2SingleAttemptOptions...)
		if err != nil {
			return nil, false, err
		}
		for i := range out.NatGateways {
			gateway := out.NatGateways[i]
			if aws.ToString(gateway.NatGatewayId) != id ||
				gateway.State == ec2types.NatGatewayStateDeleted {
				continue
			}
			return ec2TagsToMap(gateway.Tags), true, nil
		}
	case ResourceEIP:
		out, err := t.ec2.DescribeAddresses(ctx,
			&ec2.DescribeAddressesInput{AllocationIds: []string{id}}, ec2SingleAttemptOptions...)
		if err != nil {
			return nil, false, err
		}
		for i := range out.Addresses {
			if aws.ToString(out.Addresses[i].AllocationId) == id {
				return ec2TagsToMap(out.Addresses[i].Tags), true, nil
			}
		}
	default:
		return nil, false, fmt.Errorf("EC2 type %q has no known-resource describe", resourceType)
	}
	return nil, false, nil
}

// knownVPCEndpointServiceTags proves ownership of a VPC endpoint service the
// same way as its EC2 siblings. A PrivateLink service outlives the VPC that the
// trusted-parent walk starts from, so a destroy handoff can carry one as a
// survivor and it must have a direct verifier of its own.
func (t *ResourceTracker) knownVPCEndpointServiceTags(
	ctx context.Context,
	arn string,
) (map[string]string, bool, error) {
	if t.ec2 == nil {
		return nil, false, fmt.Errorf("direct EC2 ownership verifier is not configured")
	}
	id := ec2IDFromARN(arn, "vpc-endpoint-service/")
	if id == "" {
		return nil, false, fmt.Errorf("invalid VPC endpoint service ARN %q", arn)
	}
	out, err := t.ec2.DescribeVpcEndpointServiceConfigurations(ctx, &ec2.DescribeVpcEndpointServiceConfigurationsInput{
		ServiceIds: []string{id},
	}, ec2SingleAttemptOptions...)
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("describe known VPC endpoint service %s: %w", id, err)
	}
	for i := range out.ServiceConfigurations {
		service := out.ServiceConfigurations[i]
		// A service still in Deleting keeps its identity and tags, so report it as
		// present: the caller re-reads until it is terminal rather than counting a
		// half-deleted service as zero.
		if aws.ToString(service.ServiceId) != id || service.ServiceState == ec2types.ServiceStateDeleted {
			continue
		}
		return ec2TagsToMap(service.Tags), true, nil
	}
	return nil, false, nil
}

func (t *ResourceTracker) knownIAMRoleTags(ctx context.Context, arn string) (map[string]string, bool, error) {
	if t.iam == nil {
		return nil, false, fmt.Errorf("direct IAM ownership verifier is not configured")
	}
	name := arnResourceName(arn, "role/")
	if name == "" {
		return nil, false, fmt.Errorf("invalid IAM role ARN %q", arn)
	}
	out, err := t.iam.GetRole(ctx, &iam.GetRoleInput{RoleName: aws.String(name)})
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("get known IAM role %s: %w", name, err)
	}
	if out.Role == nil || aws.ToString(out.Role.Arn) != arn {
		return nil, false, fmt.Errorf("known IAM role %q returned a conflicting identity", arn)
	}
	tags, err := t.listRoleTags(ctx, name)
	if err != nil {
		return nil, false, fmt.Errorf("list known IAM role tags %s: %w", name, err)
	}
	return tags, true, nil
}

func (t *ResourceTracker) knownIAMInstanceProfileTags(
	ctx context.Context,
	arn string,
) (map[string]string, bool, error) {
	client, ok := t.iam.(iamInstanceProfileProofAPI)
	if !ok {
		return nil, false, fmt.Errorf("direct IAM instance-profile verifier is not configured")
	}
	name := arnResourceName(arn, "instance-profile/")
	if name == "" {
		return nil, false, fmt.Errorf("invalid IAM instance profile ARN %q", arn)
	}
	out, err := client.GetInstanceProfile(ctx, &iam.GetInstanceProfileInput{InstanceProfileName: aws.String(name)})
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("get known IAM instance profile %s: %w", name, err)
	}
	if out.InstanceProfile == nil || aws.ToString(out.InstanceProfile.Arn) != arn {
		return nil, false, fmt.Errorf("known IAM instance profile %q returned a conflicting identity", arn)
	}
	tags := make(map[string]string, len(out.InstanceProfile.Tags))
	copyIAMTags(tags, out.InstanceProfile.Tags)
	return tags, true, nil
}

func (t *ResourceTracker) knownELBV2Tags(
	ctx context.Context,
	resource *Resource,
) (map[string]string, bool, error) {
	if resource == nil {
		return nil, false, fmt.Errorf("known ELB resource is nil")
	}
	if t.elbv2 == nil {
		return nil, false, fmt.Errorf("direct ELB ownership verifier is not configured")
	}
	var err error
	switch resource.Type {
	case ResourceLoadBalancer:
		var out *elasticloadbalancingv2.DescribeLoadBalancersOutput
		out, err = t.elbv2.DescribeLoadBalancers(ctx, &elasticloadbalancingv2.DescribeLoadBalancersInput{
			LoadBalancerArns: []string{resource.ARN},
		})
		if err == nil && (len(out.LoadBalancers) != 1 || aws.ToString(out.LoadBalancers[0].LoadBalancerArn) != resource.ARN) {
			return nil, false, fmt.Errorf("known load balancer %q returned a conflicting identity", resource.ARN)
		}
	case ResourceTargetGroup:
		var out *elasticloadbalancingv2.DescribeTargetGroupsOutput
		out, err = t.elbv2.DescribeTargetGroups(ctx, &elasticloadbalancingv2.DescribeTargetGroupsInput{
			TargetGroupArns: []string{resource.ARN},
		})
		if err == nil && (len(out.TargetGroups) != 1 || aws.ToString(out.TargetGroups[0].TargetGroupArn) != resource.ARN) {
			return nil, false, fmt.Errorf("known target group %q returned a conflicting identity", resource.ARN)
		}
	}
	if err != nil {
		if isNotFoundErr(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("describe known ELB resource %s: %w", resource.ARN, err)
	}
	tags, err := t.elbTagsByARN(ctx, []string{resource.ARN})
	if err != nil {
		return nil, false, fmt.Errorf("describe known ELB tags %s: %w", resource.ARN, err)
	}
	return tags[resource.ARN], true, nil
}

func arnResourceName(arn, marker string) string {
	_, after, ok := strings.Cut(arn, marker)
	if !ok {
		return ""
	}
	name := strings.TrimSpace(after)
	if slash := strings.LastIndex(name, "/"); slash >= 0 {
		name = name[slash+1:]
	}
	return strings.TrimSpace(name)
}

func (t *ResourceTracker) findByTags(ctx context.Context, filters []tagtypes.TagFilter) ([]Resource, error) {
	return t.findByTagsWithStaleCheck(ctx, filters, true)
}

func (t *ResourceTracker) findByTagsWithStaleCheck(
	ctx context.Context,
	filters []tagtypes.TagFilter,
	checkStale bool,
) ([]Resource, error) {
	tagged, err := t.findTaggedResources(ctx, filters, checkStale)
	if err != nil {
		return nil, err
	}
	resources := make([]Resource, 0, len(tagged))
	for i := range tagged {
		resources = append(resources, tagged[i].Resource)
	}
	return resources, nil
}

// taggedResource carries a discovered resource together with its full tag set, so
// discovery paths that infer ownership from a non-cp9 tag can read that tag's value.
type taggedResource struct {
	Resource

	Tags map[string]string
}

func (t *ResourceTracker) findTaggedResources(
	ctx context.Context,
	filters []tagtypes.TagFilter,
	checkStale bool,
) ([]taggedResource, error) {
	var resources []taggedResource
	var token *string

	for {
		verbose.Logf("AWS tagging:GetResources region=%s filters=%v page_token=%q", t.region, filters, aws.ToString(token))
		out, err := t.client.GetResources(ctx, &resourcegroupstaggingapi.GetResourcesInput{
			TagFilters:       filters,
			PaginationToken:  token,
			ResourcesPerPage: aws.Int32(100),
		})
		if err != nil {
			return nil, fmt.Errorf("tagging api GetResources: %w", err)
		}
		verbose.Logf("AWS tagging:GetResources region=%s returned=%d next_page_token=%q",
			t.region, len(out.ResourceTagMappingList), aws.ToString(out.PaginationToken))

		for _, mapping := range out.ResourceTagMappingList {
			r := Resource{
				ARN:    aws.ToString(mapping.ResourceARN),
				Type:   ParseResourceType(aws.ToString(mapping.ResourceARN)),
				Region: firstNonEmpty(ParseResourceRegion(aws.ToString(mapping.ResourceARN)), t.region),
			}
			tags := make(map[string]string, len(mapping.Tags))
			for _, tag := range mapping.Tags {
				key := aws.ToString(tag.Key)
				tags[key] = aws.ToString(tag.Value)
				switch key {
				case TagSession:
					r.SessionID = aws.ToString(tag.Value)
				case TagOwner:
					r.Owner = aws.ToString(tag.Value)
				}
			}
			if checkStale && t.isStaleTaggedEC2Resource(ctx, &r) {
				continue
			}
			resources = append(resources, taggedResource{Resource: r, Tags: tags})
		}

		token = out.PaginationToken
		if token == nil || *token == "" {
			break
		}
	}

	return resources, nil
}

func (t *ResourceTracker) filterStaleTaggedEC2Resources(ctx context.Context, resources []Resource) []Resource {
	filtered := resources[:0]
	for i := range resources {
		if t.isStaleTaggedEC2Resource(ctx, &resources[i]) {
			continue
		}
		filtered = append(filtered, resources[i])
	}
	return filtered
}

func (t *ResourceTracker) isStaleTaggedEC2Resource(ctx context.Context, resource *Resource) bool {
	if t.ec2 == nil || resource == nil {
		return false
	}
	ctx, cancel := staleTaggedEC2ResourceContext(ctx)
	defer cancel()
	result := make(chan bool, 1)
	go func() {
		result <- t.isStaleTaggedEC2ResourceOnce(ctx, resource)
	}()
	select {
	case stale := <-result:
		return stale
	case <-ctx.Done():
		verbose.Logf("AWS stale EC2 lookup timed out for %s: %v", resource.ARN, ctx.Err())
		return false
	}
}

func (t *ResourceTracker) isStaleTaggedEC2ResourceOnce(ctx context.Context, resource *Resource) bool {
	switch resource.Type {
	case ResourceEC2Instance:
		instanceID := ec2IDFromARN(resource.ARN, "instance/")
		if instanceID == "" {
			return false
		}
		out, err := t.ec2.DescribeInstances(
			ctx,
			&ec2.DescribeInstancesInput{InstanceIds: []string{instanceID}},
			ec2SingleAttemptOptions...,
		)
		if err != nil {
			return isNotFoundErr(err)
		}
		for i := range out.Reservations {
			for j := range out.Reservations[i].Instances {
				state := out.Reservations[i].Instances[j].State
				if state != nil && state.Name == ec2types.InstanceStateNameTerminated {
					continue
				}
				return false
			}
		}
		return true
	case ResourceNATGateway:
		id := ec2IDFromARN(resource.ARN, "natgateway/")
		if id == "" {
			return false
		}
		out, err := t.ec2.DescribeNatGateways(
			ctx,
			&ec2.DescribeNatGatewaysInput{NatGatewayIds: []string{id}},
			ec2SingleAttemptOptions...,
		)
		if err != nil {
			return isNotFoundErr(err)
		}
		if len(out.NatGateways) == 0 {
			return true
		}
		return out.NatGateways[0].State == ec2types.NatGatewayStateDeleted
	case ResourceVPCEndpoint:
		id := ec2IDFromARN(resource.ARN, "vpc-endpoint/")
		if id == "" {
			return false
		}
		out, err := t.ec2.DescribeVpcEndpoints(
			ctx,
			&ec2.DescribeVpcEndpointsInput{VpcEndpointIds: []string{id}},
			ec2SingleAttemptOptions...,
		)
		if err != nil {
			return isNotFoundErr(err)
		}
		return len(out.VpcEndpoints) == 0
	case ResourceVPCEndpointService:
		id := ec2IDFromARN(resource.ARN, "vpc-endpoint-service/")
		if id == "" {
			return false
		}
		out, err := t.ec2.DescribeVpcEndpointServiceConfigurations(ctx, &ec2.DescribeVpcEndpointServiceConfigurationsInput{
			ServiceIds: []string{id},
		}, ec2SingleAttemptOptions...)
		if err != nil {
			return isNotFoundErr(err)
		}
		return len(out.ServiceConfigurations) == 0
	case ResourceENI:
		id := ec2IDFromARN(resource.ARN, "network-interface/")
		if id == "" {
			return false
		}
		out, err := t.ec2.DescribeNetworkInterfaces(
			ctx,
			&ec2.DescribeNetworkInterfacesInput{NetworkInterfaceIds: []string{id}},
			ec2SingleAttemptOptions...,
		)
		if err != nil {
			return isNotFoundErr(err)
		}
		return len(out.NetworkInterfaces) == 0
	case ResourceEBSVolume:
		id := ec2IDFromARN(resource.ARN, "volume/")
		if id == "" {
			return false
		}
		out, err := t.ec2.DescribeVolumes(
			ctx,
			&ec2.DescribeVolumesInput{VolumeIds: []string{id}},
			ec2SingleAttemptOptions...,
		)
		if err != nil {
			return isNotFoundErr(err)
		}
		return len(out.Volumes) == 0
	case ResourceEBSSnapshot:
		id := ec2IDFromARN(resource.ARN, "snapshot/")
		if id == "" {
			return false
		}
		out, err := t.ec2.DescribeSnapshots(
			ctx,
			&ec2.DescribeSnapshotsInput{SnapshotIds: []string{id}},
			ec2SingleAttemptOptions...,
		)
		if err != nil {
			return isNotFoundErr(err)
		}
		return len(out.Snapshots) == 0
	default:
		return false
	}
}

func staleTaggedEC2ResourceContext(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(ctx, staleTaggedEC2ResourceLookupTimeout)
}

func (t *ResourceTracker) findCP9ManagedFallback(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	if strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	resources, err := t.findByTagsWithStaleCheck(ctx, []tagtypes.TagFilter{
		{Key: aws.String("environment"), Values: []string{"cp9"}},
		{Key: aws.String(TagManaged), Values: []string{"true"}},
	}, false)
	if err != nil {
		return nil, err
	}
	var scoped []Resource
	for _, resource := range resources {
		if resourceMatchesFallbackScope(&resource, owner, sessionID) {
			// Fallback discovery already attributed the resource to this scope by
			// its CP9-managed tags and session token. Keep the returned scope
			// canonical so downstream cleanup accepts product-derived regional
			// tags such as "<session>-r2".
			if strings.TrimSpace(owner) != "" {
				resource.Owner = owner
			}
			resource.SessionID = sessionID
			scoped = append(scoped, resource)
		}
	}
	return t.filterStaleTaggedEC2Resources(ctx, scoped), nil
}

func resourceMatchesFallbackScope(resource *Resource, owner, sessionID string) bool {
	if resource == nil {
		return false
	}
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		return false
	}
	resourceOwner := strings.TrimSpace(resource.Owner)
	resourceSession := strings.TrimSpace(resource.SessionID)
	owner = strings.TrimSpace(owner)
	if resourceSession == sessionID {
		return owner == "" || resourceOwner == "" || ownerTagMatches(owner, resourceOwner)
	}
	if resourceSession != "" && !regionalChildSessionMatches(sessionID, resourceSession) {
		return false
	}
	if resource.Type == ResourceEKSCluster || resource.Type == ResourceEKSNodegroup {
		if !EKSARNMatchesSession(resource.ARN, sessionID, resource.Region) {
			return false
		}
		return owner == "" || resourceOwner == "" || ownerTagMatches(owner, resourceOwner)
	}
	if arnContainsSessionToken(resource.ARN, sessionID) {
		return owner == "" || resourceOwner == "" || ownerTagMatches(owner, resourceOwner)
	}
	return false
}

func regionalChildSessionMatches(parent, child string) bool {
	suffix, ok := strings.CutPrefix(child, parent+"-r")
	if !ok || suffix == "" {
		return false
	}
	for _, ch := range suffix {
		if ch < '0' || ch > '9' {
			return false
		}
	}
	return true
}

func (t *ResourceTracker) findEnvironmentSessionNameFallback(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	if strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	resources, err := t.findByTagsWithStaleCheck(ctx, []tagtypes.TagFilter{
		{Key: aws.String("environment"), Values: []string{"cp9"}},
	}, false)
	if err != nil {
		return nil, err
	}
	var scoped []Resource
	for _, resource := range resources {
		existing := strings.TrimSpace(resource.SessionID)
		if existing != "" && existing != sessionID && !regionalChildSessionMatches(sessionID, existing) {
			continue
		}
		if !resourceMatchesEnvironmentSessionNameFallback(&resource, sessionID) {
			continue
		}
		requestedOwner := strings.TrimSpace(owner)
		existingOwner := strings.TrimSpace(resource.Owner)
		if existingOwner != "" && (requestedOwner == "" || !ownerTagMatches(requestedOwner, existingOwner)) {
			continue
		}
		if requestedOwner != "" {
			resource.Owner = requestedOwner
		}
		resource.SessionID = sessionID
		scoped = append(scoped, resource)
	}
	return t.filterStaleTaggedEC2Resources(ctx, scoped), nil
}

// kubernetesClusterTagKey is the tag the AWS EBS CSI driver and the in-tree AWS
// cloud provider stamp on resources they create for a cluster, carrying the cluster
// name as its value. Dataplane-created resources — TiKV and PD data volumes above
// all — carry it while carrying no cp9 ownership tag at all, because the creating
// service is product code that knows nothing about cp9 sessions.
const kubernetesClusterTagKey = "KubernetesCluster"

// kubernetesClusterNameMatchesSession reports whether a Kubernetes cluster name
// belongs to exactly this cp9 session.
//
// Matching is deliberately restricted to the cluster-name shapes cp9 actually
// produces, and anchored on what follows the session name, because a loose token
// match is unsafe here. Session names nest: with only a token-boundary check,
// session cp9-ui would claim cp9-ui-trains-debug1's cluster. These resources carry
// no cp9 ownership tag whether their session is dead or alive, so the cluster name
// is the only discriminator and over-claiming would delete a live session's data
// volumes. Anything that does not match a known shape is left alone.
func kubernetesClusterNameMatchesSession(clusterName, sessionID, region string) bool {
	clusterName = strings.TrimSpace(clusterName)
	sessionID = strings.TrimSpace(sessionID)
	if clusterName == "" || sessionID == "" {
		return false
	}
	// The session's own regional cluster.
	if clusterName == sessionID {
		return true
	}
	// Starter regional cluster: "<session>-starter-regional".
	if clusterName == sessionID+"-starter-regional" {
		return true
	}
	// Dedicated dataplane cluster: "dedicated-<session>-<region>-<hash>...".
	// Session names may contain hyphens, including the region string, so a
	// HasPrefix on "dedicated-<session>-<region>-" would let session "abc"
	// claim "dedicated-abc-us-west-2-us-west-2-...". Parse session as the
	// exact token between "dedicated-" and the last "-<region>-".
	if region = strings.TrimSpace(region); region != "" {
		const dedicatedPrefix = "dedicated-"
		anchor := "-" + region + "-"
		if strings.HasPrefix(clusterName, dedicatedPrefix) {
			rest := clusterName[len(dedicatedPrefix):]
			if idx := strings.LastIndex(rest, anchor); idx >= 0 && rest[:idx] == sessionID {
				return true
			}
		}
	}
	return false
}

// findKubernetesClusterNameFallback recovers dataplane resources whose only link to
// a cp9 session is the Kubernetes cluster name in their KubernetesCluster tag.
//
// Without this, an EKS dataplane that is deleted without its CSI controller draining
// first leaves its EBS volumes behind carrying no cp9-owner/cp9-session tag, which
// makes them invisible to the tag-scoped reaper and to aws-nuke: a silent, unbounded
// cost leak. Claiming them here lets the existing tag-assurance step stamp ownership
// so the normal reaper reclaims them.
//
// The session token must appear in the cluster name on token boundaries, so a session
// named cp9-ui cannot claim cp9-ui-trains-debug1's cluster. A resource already tagged
// for a different cp9 session is never claimed.
func (t *ResourceTracker) findKubernetesClusterNameFallback(
	ctx context.Context,
	owner, sessionID string,
) ([]Resource, error) {
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		return nil, nil
	}
	tagged, err := t.findTaggedResources(ctx, []tagtypes.TagFilter{
		{Key: aws.String(kubernetesClusterTagKey)},
	}, false)
	if err != nil {
		return nil, err
	}
	var scoped []Resource
	for i := range tagged {
		clusterName := strings.TrimSpace(tagged[i].Tags[kubernetesClusterTagKey])
		if !kubernetesClusterNameMatchesSession(clusterName, sessionID, t.region) {
			continue
		}
		if existing := strings.TrimSpace(tagged[i].SessionID); existing != "" && existing != sessionID {
			continue
		}
		resource := tagged[i].Resource
		verbose.Logf(
			"AWS ownership recovered from %s=%q for session=%s: %s",
			kubernetesClusterTagKey,
			clusterName,
			sessionID,
			resource.ARN,
		)
		resource.Owner = owner
		resource.SessionID = sessionID
		scoped = append(scoped, resource)
	}
	return t.filterStaleTaggedEC2Resources(ctx, scoped), nil
}

func resourceMatchesEnvironmentSessionNameFallback(resource *Resource, sessionID string) bool {
	if resource == nil {
		return false
	}
	switch resource.Type {
	case ResourceEKSCluster, ResourceEKSNodegroup:
	default:
		return false
	}
	return EKSARNMatchesSession(resource.ARN, sessionID, resource.Region)
}

// EKSARNMatchesSession reports whether an EKS cluster or nodegroup ARN belongs
// to exactly this cp9 session. Matching is the Recycle ownerless-name shapes
// from ADR 0001: the session name, <session>-starter-regional, or
// dedicated-<session>-<region>-….
func EKSARNMatchesSession(arn, sessionID, region string) bool {
	clusterName := eksClusterNameFromARN(arn)
	if clusterName == "" {
		clusterName, _ = eksNodegroupNamesFromARN(arn)
	}
	return kubernetesClusterNameMatchesSession(clusterName, sessionID, firstNonEmpty(region, ParseResourceRegion(arn)))
}

func (t *ResourceTracker) findTrustedEKSRoleChildren(
	ctx context.Context,
	owner string,
	sessionID string,
) ([]Resource, error) {
	if strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	if t.requireTrustedEKSRoleGraph && (t.eks == nil || t.iam == nil) {
		return nil, fmt.Errorf("trusted EKS parent proof is required but its IAM/EKS graph clients are not configured")
	}
	if t.eks == nil || t.iam == nil {
		return nil, nil
	}
	lookupCtx, cancel := context.WithTimeout(ctx, trustedEKSGraphLookupTimeout)
	defer cancel()
	type result struct {
		graph TrustedEKSRoleGraph
		err   error
	}
	results := make(chan result, 1)
	go func() {
		graph, err := DiscoverTrustedEKSRoleGraph(
			lookupCtx,
			t.eks,
			t.iam,
			owner,
			sessionID,
			t.region,
			t.accountID,
		)
		results <- result{graph: graph, err: err}
	}()
	select {
	case result := <-results:
		if result.err != nil {
			return nil, result.err
		}
		return result.graph.Resources, nil
	case <-lookupCtx.Done():
		return nil, fmt.Errorf(
			"trusted EKS parent proof timed out for session %q: %w",
			sessionID,
			lookupCtx.Err(),
		)
	}
}

func (t *ResourceTracker) findSessionNamedEKSClusters(
	ctx context.Context,
	sessionID string,
) ([]Resource, error) {
	if t.eks == nil || strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	names, err := listTrustedEKSClusters(ctx, t.eks, t.region)
	if err != nil {
		return nil, err
	}
	for _, name := range names {
		if !kubernetesClusterNameMatchesSession(name, sessionID, t.region) {
			continue
		}
		if strings.TrimSpace(t.region) == "" || strings.TrimSpace(t.accountID) == "" {
			return nil, fmt.Errorf("session-named EKS collision has incomplete account or region identity")
		}
		return []Resource{{
			ARN:       fmt.Sprintf("arn:aws:eks:%s:%s:cluster/%s", t.region, t.accountID, name),
			Type:      ResourceEKSCluster,
			Region:    t.region,
			SessionID: sessionID,
		}}, nil
	}
	return nil, nil
}

// DiscoverTrustedEKSRoleGraph follows immutable EKS relationships from an
// exact deterministic service role. It independently re-reads the IAM role and
// requires its account/owner/session/managed tags before inspecting EKS. It is
// strictly read-only and validates the whole graph before returning resources.
func DiscoverTrustedEKSRoleGraph(
	ctx context.Context,
	client EKSResourceGraphAPI,
	roleClient EKSRoleProofAPI,
	owner string,
	sessionID string,
	region string,
	accountID string,
) (TrustedEKSRoleGraph, error) {
	graph := TrustedEKSRoleGraph{ParentByARN: map[string]string{}}
	if client == nil || roleClient == nil || strings.TrimSpace(sessionID) == "" {
		return graph, nil
	}
	roleARN, scopeOwner, err := trustedDeterministicEKSServiceRole(
		ctx,
		roleClient,
		owner,
		sessionID,
		accountID,
	)
	if err != nil {
		return TrustedEKSRoleGraph{}, err
	}
	if roleARN == "" {
		return graph, nil
	}
	graph.RoleARN = roleARN

	clusterNames, err := listTrustedEKSClusters(ctx, client, region)
	if err != nil {
		return TrustedEKSRoleGraph{}, err
	}
	for _, clusterName := range clusterNames {
		verbose.Logf("AWS eks:DescribeCluster trusted-parent region=%s cluster=%s", region, clusterName)
		out, err := client.DescribeCluster(ctx, &eks.DescribeClusterInput{Name: aws.String(clusterName)})
		if err != nil {
			if isNotFoundErr(err) {
				continue
			}
			return TrustedEKSRoleGraph{}, fmt.Errorf(
				"describe EKS cluster %s for trusted-parent discovery: %w",
				clusterName,
				err,
			)
		}
		cluster := out.Cluster
		if cluster == nil {
			return TrustedEKSRoleGraph{}, fmt.Errorf(
				"describe EKS cluster %s for trusted-parent discovery returned no cluster",
				clusterName,
			)
		}
		if strings.TrimSpace(aws.ToString(cluster.RoleArn)) != roleARN {
			continue
		}
		if describedName := strings.TrimSpace(aws.ToString(cluster.Name)); describedName != "" && describedName != clusterName {
			return TrustedEKSRoleGraph{}, fmt.Errorf(
				"EKS cluster %s trusted-parent describe returned name %q",
				clusterName,
				describedName,
			)
		}
		clusterARN := strings.TrimSpace(aws.ToString(cluster.Arn))
		if clusterARN == "" {
			return TrustedEKSRoleGraph{}, fmt.Errorf(
				"EKS cluster %s matched trusted role %s but returned no ARN",
				clusterName,
				roleARN,
			)
		}
		clusterTags, err := trustedEKSTags(ctx, client, clusterARN, region)
		if err != nil {
			if isNotFoundErr(err) {
				continue
			}
			return TrustedEKSRoleGraph{}, err
		}
		if err := validateTrustedEKSGraphTags(clusterARN, clusterTags, owner, sessionID, true); err != nil {
			return TrustedEKSRoleGraph{}, err
		}
		resourceRegion := firstNonEmpty(ParseResourceRegion(clusterARN), region)
		clusterResource := Resource{
			ARN:       clusterARN,
			Type:      ResourceEKSCluster,
			Region:    resourceRegion,
			SessionID: sessionID,
			Owner:     scopeOwner,
		}

		nodegroupNames, err := listTrustedEKSNodegroups(ctx, client, clusterName, region)
		if err != nil {
			if isNotFoundErr(err) {
				continue
			}
			return TrustedEKSRoleGraph{}, err
		}
		var nodeResources []Resource
		for _, nodegroupName := range nodegroupNames {
			verbose.Logf(
				"AWS eks:DescribeNodegroup trusted-parent region=%s cluster=%s nodegroup=%s",
				region,
				clusterName,
				nodegroupName,
			)
			nodeOut, err := client.DescribeNodegroup(ctx, &eks.DescribeNodegroupInput{
				ClusterName:   aws.String(clusterName),
				NodegroupName: aws.String(nodegroupName),
			})
			if err != nil {
				if isNotFoundErr(err) {
					continue
				}
				return TrustedEKSRoleGraph{}, fmt.Errorf(
					"describe EKS nodegroup %s/%s for trusted-parent discovery: %w",
					clusterName,
					nodegroupName,
					err,
				)
			}
			nodegroup := nodeOut.Nodegroup
			if nodegroup == nil || strings.TrimSpace(aws.ToString(nodegroup.ClusterName)) != clusterName {
				return TrustedEKSRoleGraph{}, fmt.Errorf(
					"EKS nodegroup %s/%s did not prove cluster containment",
					clusterName,
					nodegroupName,
				)
			}
			if describedName := strings.TrimSpace(aws.ToString(nodegroup.NodegroupName)); describedName != "" && describedName != nodegroupName {
				return TrustedEKSRoleGraph{}, fmt.Errorf(
					"EKS nodegroup %s/%s trusted-parent describe returned name %q",
					clusterName,
					nodegroupName,
					describedName,
				)
			}
			nodeARN := strings.TrimSpace(aws.ToString(nodegroup.NodegroupArn))
			if nodeARN == "" {
				return TrustedEKSRoleGraph{}, fmt.Errorf(
					"EKS nodegroup %s/%s returned no ARN",
					clusterName,
					nodegroupName,
				)
			}
			nodeTags, err := trustedEKSTags(ctx, client, nodeARN, region)
			if err != nil {
				if isNotFoundErr(err) {
					continue
				}
				return TrustedEKSRoleGraph{}, err
			}
			if err := validateTrustedEKSGraphTags(nodeARN, nodeTags, owner, sessionID, false); err != nil {
				return TrustedEKSRoleGraph{}, err
			}
			nodeResources = append(nodeResources, Resource{
				ARN:       nodeARN,
				Type:      ResourceEKSNodegroup,
				Region:    firstNonEmpty(ParseResourceRegion(nodeARN), resourceRegion),
				SessionID: sessionID,
				Owner:     scopeOwner,
			})
		}
		graph.Resources = append(graph.Resources, clusterResource)
		graph.ParentByARN[clusterARN] = roleARN
		for i := range nodeResources {
			graph.Resources = append(graph.Resources, nodeResources[i])
			graph.ParentByARN[nodeResources[i].ARN] = clusterARN
		}
	}
	return graph, nil
}

func trustedDeterministicEKSServiceRole(
	ctx context.Context,
	client EKSRoleProofAPI,
	owner string,
	sessionID string,
	accountID string,
) (roleARN, scopeOwner string, err error) {
	roleName := deterministicEKSServiceRoleName(sessionID)
	if client == nil || roleName == "" {
		return "", "", nil
	}
	owner = strings.TrimSpace(owner)
	sessionID = strings.TrimSpace(sessionID)
	accountID = strings.TrimSpace(accountID)
	verbose.Logf("AWS iam:GetRole trusted-eks-parent role=%s", roleName)
	roleOut, err := client.GetRole(ctx, &iam.GetRoleInput{RoleName: aws.String(roleName)})
	if err != nil {
		if isNotFoundErr(err) {
			return "", "", nil
		}
		return "", "", fmt.Errorf("get trusted EKS parent role %s: %w", roleName, err)
	}
	if roleOut.Role == nil {
		return "", "", fmt.Errorf("get trusted EKS parent role %s returned no role", roleName)
	}
	if describedName := strings.TrimSpace(aws.ToString(roleOut.Role.RoleName)); describedName != "" && describedName != roleName {
		return "", "", fmt.Errorf("trusted EKS parent role %s returned name %q", roleName, describedName)
	}
	roleARN = strings.TrimSpace(aws.ToString(roleOut.Role.Arn))
	roleAccount, rolePath, ok := parseRoleARN(roleARN)
	if !ok || rolePath != roleName {
		return "", "", fmt.Errorf("trusted EKS parent role %s returned invalid ARN %q", roleName, roleARN)
	}
	if accountID != "" && roleAccount != accountID {
		return "", "", fmt.Errorf(
			"refusing EKS trusted-parent role %s from account %q, want %q",
			roleARN,
			roleAccount,
			accountID,
		)
	}

	tags := map[string]string{}
	var marker *string
	for {
		verbose.Logf(
			"AWS iam:ListRoleTags trusted-eks-parent role=%s marker=%q",
			roleName,
			aws.ToString(marker),
		)
		tagOut, err := client.ListRoleTags(ctx, &iam.ListRoleTagsInput{
			RoleName: aws.String(roleName),
			Marker:   marker,
		})
		if err != nil {
			if isNotFoundErr(err) {
				return "", "", nil
			}
			return "", "", fmt.Errorf("list trusted EKS parent role tags %s: %w", roleName, err)
		}
		copyIAMTags(tags, tagOut.Tags)
		if !tagOut.IsTruncated || tagOut.Marker == nil || aws.ToString(tagOut.Marker) == "" {
			break
		}
		marker = tagOut.Marker
	}
	resourceOwner := strings.TrimSpace(tags[TagOwner])
	if resourceOwner == "" || !tagsMatchScope(tags, owner, sessionID) {
		return "", "", fmt.Errorf(
			"refusing EKS trusted-parent role %s with owner=%q session=%q managed=%q, want owner=%q session=%q managed=true",
			roleARN,
			resourceOwner,
			strings.TrimSpace(tags[TagSession]),
			strings.TrimSpace(tags[TagManaged]),
			owner,
			sessionID,
		)
	}
	return roleARN, resourceOwner, nil
}

func listTrustedEKSClusters(ctx context.Context, client EKSResourceGraphAPI, region string) ([]string, error) {
	var names []string
	var token *string
	for {
		verbose.Logf("AWS eks:ListClusters trusted-parent region=%s page_token=%q", region, aws.ToString(token))
		out, err := client.ListClusters(ctx, &eks.ListClustersInput{NextToken: token})
		if err != nil {
			return nil, fmt.Errorf("list EKS clusters for trusted-parent discovery: %w", err)
		}
		names = append(names, out.Clusters...)
		token = out.NextToken
		if token == nil || aws.ToString(token) == "" {
			break
		}
	}
	slices.Sort(names)
	return names, nil
}

func listTrustedEKSNodegroups(
	ctx context.Context,
	client EKSResourceGraphAPI,
	clusterName string,
	region string,
) ([]string, error) {
	var names []string
	var token *string
	for {
		verbose.Logf(
			"AWS eks:ListNodegroups trusted-parent region=%s cluster=%s page_token=%q",
			region,
			clusterName,
			aws.ToString(token),
		)
		out, err := client.ListNodegroups(ctx, &eks.ListNodegroupsInput{
			ClusterName: aws.String(clusterName),
			NextToken:   token,
		})
		if err != nil {
			return nil, fmt.Errorf("list EKS nodegroups for cluster %s: %w", clusterName, err)
		}
		names = append(names, out.Nodegroups...)
		token = out.NextToken
		if token == nil || aws.ToString(token) == "" {
			break
		}
	}
	slices.Sort(names)
	return names, nil
}

func trustedEKSTags(
	ctx context.Context,
	client EKSResourceGraphAPI,
	arn string,
	region string,
) (map[string]string, error) {
	verbose.Logf("AWS eks:ListTagsForResource trusted-parent region=%s arn=%s", region, arn)
	out, err := client.ListTagsForResource(ctx, &eks.ListTagsForResourceInput{ResourceArn: aws.String(arn)})
	if err != nil {
		return nil, fmt.Errorf("list EKS tags for trusted-parent resource %s: %w", arn, err)
	}
	tags := make(map[string]string, len(out.Tags))
	maps.Copy(tags, out.Tags)
	return tags, nil
}

func validateTrustedEKSGraphTags(
	arn string,
	tags map[string]string,
	owner string,
	sessionID string,
	requireEnvironment bool,
) error {
	environment := strings.TrimSpace(tags["environment"])
	if requireEnvironment && environment != "cp9" {
		return fmt.Errorf("refusing EKS trusted-parent resource %s without environment=cp9 (got %q)", arn, environment)
	}
	if !requireEnvironment && environment != "" && environment != "cp9" {
		return fmt.Errorf("refusing EKS trusted-parent resource %s with environment=%q", arn, environment)
	}
	resourceOwner := strings.TrimSpace(tags[TagOwner])
	resourceSession := strings.TrimSpace(tags[TagSession])
	managed := strings.TrimSpace(tags[TagManaged])
	if (resourceOwner != "" && !ownerTagMatches(owner, resourceOwner)) ||
		(resourceSession != "" && resourceSession != sessionID) ||
		(managed != "" && managed != "true") {
		return fmt.Errorf(
			"refusing EKS trusted-parent resource %s with conflicting cleanup scope owner=%q session=%q managed=%q",
			arn,
			resourceOwner,
			resourceSession,
			managed,
		)
	}
	return nil
}

func arnContainsSessionToken(arn, sessionID string) bool {
	start := strings.Index(arn, sessionID)
	for start >= 0 {
		end := start + len(sessionID)
		if sessionTokenBoundary(arn, start-1) && sessionTokenBoundary(arn, end) {
			return true
		}
		next := strings.Index(arn[end:], sessionID)
		if next < 0 {
			return false
		}
		start = end + next
	}
	return false
}

func sessionTokenBoundary(value string, index int) bool {
	if index < 0 || index >= len(value) {
		return true
	}
	ch := value[index]
	if ch >= 'a' && ch <= 'z' {
		return false
	}
	if ch >= 'A' && ch <= 'Z' {
		return false
	}
	if ch >= '0' && ch <= '9' {
		return false
	}
	return true
}

func (t *ResourceTracker) findVPCMembership(ctx context.Context, owner, sessionID string, roots []Resource) ([]Resource, error) {
	if t.ec2 == nil || strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	var resources []Resource
	seenVPC := map[string]struct{}{}
	for i := range roots {
		root := roots[i]
		if root.Type != ResourceVPC {
			continue
		}
		vpcID := ec2IDFromARN(root.ARN, "vpc/")
		if vpcID == "" {
			continue
		}
		if _, ok := seenVPC[vpcID]; ok {
			continue
		}
		seenVPC[vpcID] = struct{}{}
		children, err := t.findVPCChildren(ctx, &root, vpcID, owner, sessionID)
		if err != nil {
			return nil, err
		}
		resources = mergeTrackerResources(resources, children)
		elbChildren, err := t.findELBChildren(ctx, &root, vpcID, owner, sessionID)
		if err != nil {
			if errors.Is(err, context.DeadlineExceeded) {
				verbose.Logf("AWS elbv2: skipping VPC %s child lookup after timeout: %v", vpcID, err)
				continue
			}
			return nil, err
		}
		resources = mergeTrackerResources(resources, elbChildren)
	}
	return resources, nil
}

func (t *ResourceTracker) findELBChildren(
	ctx context.Context,
	root *Resource,
	vpcID string,
	owner string,
	sessionID string,
) ([]Resource, error) {
	if t.elbv2 == nil && t.classicELB == nil {
		return nil, nil
	}
	if elbChildLookupTimeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, elbChildLookupTimeout)
		defer cancel()
	}
	region := firstNonEmpty(root.Region, t.region)
	scopeOwner := firstNonEmpty(strings.TrimSpace(owner), root.Owner)
	scopeSession := firstNonEmpty(strings.TrimSpace(sessionID), root.SessionID)
	resource := func(resourceType ResourceType, arn string) Resource {
		return Resource{
			ARN:       arn,
			Type:      resourceType,
			Region:    region,
			SessionID: scopeSession,
			Owner:     scopeOwner,
		}
	}

	var resources []Resource
	if t.elbv2 != nil {
		lbs, err := t.elbv2.DescribeLoadBalancers(ctx, &elasticloadbalancingv2.DescribeLoadBalancersInput{})
		if err != nil {
			return nil, fmt.Errorf("describe VPC %s load balancers: %w", vpcID, err)
		}
		var lbARNs []string
		for i := range lbs.LoadBalancers {
			lb := &lbs.LoadBalancers[i]
			if aws.ToString(lb.VpcId) == vpcID && aws.ToString(lb.LoadBalancerArn) != "" {
				lbARNs = append(lbARNs, aws.ToString(lb.LoadBalancerArn))
			}
		}
		lbTags, err := t.elbTagsByARN(ctx, lbARNs)
		if err != nil {
			return nil, fmt.Errorf("describe VPC %s load balancer tags: %w", vpcID, err)
		}
		for _, arn := range lbARNs {
			if tagsMatchDeleteScope(lbTags[arn], scopeOwner, scopeSession) {
				resources = append(resources, resource(ResourceLoadBalancer, arn))
			}
		}

		tgs, err := t.elbv2.DescribeTargetGroups(ctx, &elasticloadbalancingv2.DescribeTargetGroupsInput{})
		if err != nil {
			return nil, fmt.Errorf("describe VPC %s target groups: %w", vpcID, err)
		}
		var tgARNs []string
		for i := range tgs.TargetGroups {
			tg := &tgs.TargetGroups[i]
			if aws.ToString(tg.VpcId) == vpcID && aws.ToString(tg.TargetGroupArn) != "" {
				tgARNs = append(tgARNs, aws.ToString(tg.TargetGroupArn))
			}
		}
		tgTags, err := t.elbTagsByARN(ctx, tgARNs)
		if err != nil {
			return nil, fmt.Errorf("describe VPC %s target group tags: %w", vpcID, err)
		}
		for _, arn := range tgARNs {
			if tagsMatchDeleteScope(tgTags[arn], scopeOwner, scopeSession) {
				resources = append(resources, resource(ResourceTargetGroup, arn))
			}
		}
	}
	if t.classicELB != nil {
		lbs, err := t.classicELB.DescribeLoadBalancers(ctx, &elasticloadbalancing.DescribeLoadBalancersInput{})
		if err != nil {
			return nil, fmt.Errorf("describe VPC %s classic load balancers: %w", vpcID, err)
		}
		var lbNames []string
		for i := range lbs.LoadBalancerDescriptions {
			lb := &lbs.LoadBalancerDescriptions[i]
			if aws.ToString(lb.VPCId) == vpcID && aws.ToString(lb.LoadBalancerName) != "" {
				lbNames = append(lbNames, aws.ToString(lb.LoadBalancerName))
			}
		}
		lbTags, err := t.classicELBTagsByName(ctx, lbNames)
		if err != nil {
			return nil, fmt.Errorf("describe VPC %s classic load balancer tags: %w", vpcID, err)
		}
		accountID := accountIDFromARN(root.ARN)
		for _, name := range lbNames {
			if tagsMatchDeleteScope(lbTags[name], scopeOwner, scopeSession) {
				resources = append(resources, resource(ResourceLoadBalancer, classicELBARN(region, accountID, name)))
			}
		}
	}
	return resources, nil
}

func (t *ResourceTracker) elbTagsByARN(ctx context.Context, arns []string) (map[string]map[string]string, error) {
	tags := make(map[string]map[string]string, len(arns))
	if len(arns) == 0 {
		return tags, nil
	}
	out, err := t.elbv2.DescribeTags(ctx, &elasticloadbalancingv2.DescribeTagsInput{
		ResourceArns: arns,
	})
	if err != nil {
		return nil, err
	}
	for _, description := range out.TagDescriptions {
		arn := aws.ToString(description.ResourceArn)
		if arn == "" {
			continue
		}
		tags[arn] = elbTagsToMap(description.Tags)
	}
	return tags, nil
}

func (t *ResourceTracker) classicELBTagsByName(ctx context.Context, names []string) (map[string]map[string]string, error) {
	tags := make(map[string]map[string]string, len(names))
	if len(names) == 0 {
		return tags, nil
	}
	out, err := t.classicELB.DescribeTags(ctx, &elasticloadbalancing.DescribeTagsInput{
		LoadBalancerNames: names,
	})
	if err != nil {
		return nil, err
	}
	for _, description := range out.TagDescriptions {
		name := aws.ToString(description.LoadBalancerName)
		if name == "" {
			continue
		}
		tags[name] = classicELBTagsToMap(description.Tags)
	}
	return tags, nil
}

func (t *ResourceTracker) findVPCChildren(
	ctx context.Context,
	root *Resource,
	vpcID string,
	owner string,
	sessionID string,
) ([]Resource, error) {
	region := firstNonEmpty(root.Region, t.region)
	accountID := firstNonEmpty(accountIDFromARN(root.ARN), t.accountID)
	if region == "" || accountID == "" {
		return nil, nil
	}
	scopeOwner := firstNonEmpty(strings.TrimSpace(owner), root.Owner)
	scopeSession := firstNonEmpty(strings.TrimSpace(sessionID), root.SessionID)
	vpcFilter := []ec2types.Filter{{Name: aws.String("vpc-id"), Values: []string{vpcID}}}
	resource := func(resourceType ResourceType, path string) Resource {
		return Resource{
			ARN:       fmt.Sprintf("arn:aws:ec2:%s:%s:%s", region, accountID, path),
			Type:      resourceType,
			Region:    region,
			SessionID: scopeSession,
			Owner:     scopeOwner,
		}
	}

	var resources []Resource
	subnets, err := t.ec2.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{Filters: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s subnets: %w", vpcID, err)
	}
	for i := range subnets.Subnets {
		if id := aws.ToString(subnets.Subnets[i].SubnetId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(subnets.Subnets[i].Tags), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceSubnet, "subnet/"+id))
		}
	}

	routeTables, err := t.ec2.DescribeRouteTables(ctx, &ec2.DescribeRouteTablesInput{Filters: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s route tables: %w", vpcID, err)
	}
	for i := range routeTables.RouteTables {
		if id := aws.ToString(routeTables.RouteTables[i].RouteTableId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(routeTables.RouteTables[i].Tags), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceRouteTable, "route-table/"+id))
		}
	}

	gateways, err := t.ec2.DescribeInternetGateways(ctx, &ec2.DescribeInternetGatewaysInput{
		Filters: []ec2types.Filter{{Name: aws.String("attachment.vpc-id"), Values: []string{vpcID}}},
	})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s internet gateways: %w", vpcID, err)
	}
	for i := range gateways.InternetGateways {
		if id := aws.ToString(gateways.InternetGateways[i].InternetGatewayId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(gateways.InternetGateways[i].Tags), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceInternetGateway, "internet-gateway/"+id))
		}
	}

	natGateways, err := t.ec2.DescribeNatGateways(ctx, &ec2.DescribeNatGatewaysInput{Filter: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s NAT gateways: %w", vpcID, err)
	}
	for i := range natGateways.NatGateways {
		if natGateways.NatGateways[i].State == ec2types.NatGatewayStateDeleted {
			continue
		}
		if id := aws.ToString(natGateways.NatGateways[i].NatGatewayId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(natGateways.NatGateways[i].Tags), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceNATGateway, "natgateway/"+id))
		}
	}

	endpoints, err := t.ec2.DescribeVpcEndpoints(ctx, &ec2.DescribeVpcEndpointsInput{Filters: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s endpoints: %w", vpcID, err)
	}
	for i := range endpoints.VpcEndpoints {
		if id := aws.ToString(endpoints.VpcEndpoints[i].VpcEndpointId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(endpoints.VpcEndpoints[i].Tags), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceVPCEndpoint, "vpc-endpoint/"+id))
		}
	}

	interfaces, err := t.ec2.DescribeNetworkInterfaces(ctx, &ec2.DescribeNetworkInterfacesInput{Filters: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s network interfaces: %w", vpcID, err)
	}
	for i := range interfaces.NetworkInterfaces {
		if id := aws.ToString(interfaces.NetworkInterfaces[i].NetworkInterfaceId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(interfaces.NetworkInterfaces[i].TagSet), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceENI, "network-interface/"+id))
		}
	}

	securityGroups, err := t.ec2.DescribeSecurityGroups(ctx, &ec2.DescribeSecurityGroupsInput{Filters: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s security groups: %w", vpcID, err)
	}
	for i := range securityGroups.SecurityGroups {
		if id := aws.ToString(securityGroups.SecurityGroups[i].GroupId); id != "" &&
			tagsMatchDeleteScope(ec2TagsToMap(securityGroups.SecurityGroups[i].Tags), scopeOwner, scopeSession) {
			resources = append(resources, resource(ResourceSecurityGroup, "security-group/"+id))
		}
	}

	instances, err := t.ec2.DescribeInstances(ctx, &ec2.DescribeInstancesInput{Filters: vpcFilter})
	if err != nil {
		return nil, fmt.Errorf("describe VPC %s instances: %w", vpcID, err)
	}
	for i := range instances.Reservations {
		for j := range instances.Reservations[i].Instances {
			instance := &instances.Reservations[i].Instances[j]
			if instance.State != nil && instance.State.Name == ec2types.InstanceStateNameTerminated {
				continue
			}
			if id := aws.ToString(instance.InstanceId); id != "" &&
				tagsMatchDeleteScope(ec2TagsToMap(instance.Tags), scopeOwner, scopeSession) {
				resources = append(resources, resource(ResourceEC2Instance, "instance/"+id))
			}
		}
	}

	return resources, nil
}

func ec2TagsToMap(tags []ec2types.Tag) map[string]string {
	out := make(map[string]string, len(tags))
	for _, tag := range tags {
		out[aws.ToString(tag.Key)] = aws.ToString(tag.Value)
	}
	return out
}

func elbTagsToMap(tags []elbv2types.Tag) map[string]string {
	out := make(map[string]string, len(tags))
	for _, tag := range tags {
		out[aws.ToString(tag.Key)] = aws.ToString(tag.Value)
	}
	return out
}

func classicELBTagsToMap(tags []elbtypes.Tag) map[string]string {
	out := make(map[string]string, len(tags))
	for _, tag := range tags {
		out[aws.ToString(tag.Key)] = aws.ToString(tag.Value)
	}
	return out
}

func tagsMatchDeleteScope(tags map[string]string, owner, sessionID string) bool {
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" || strings.TrimSpace(tags[TagSession]) != sessionID {
		return false
	}
	tagOwner := strings.TrimSpace(tags[TagOwner])
	if tagOwner == "" {
		return false
	}
	owner = strings.TrimSpace(owner)
	if owner != "" && !ownerTagMatches(owner, tagOwner) {
		return false
	}
	return strings.TrimSpace(tags[TagTTL]) != ""
}

func (t *ResourceTracker) findDeterministicIAM(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	if t.iam == nil || strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	ctx, cancel := context.WithTimeout(ctx, deterministicIAMLookupTimeout)
	defer cancel()
	type result struct {
		resources []Resource
		err       error
	}
	results := make(chan result, 1)
	go func() {
		resources, err := t.findDeterministicIAMOnce(ctx, owner, sessionID)
		results <- result{resources: resources, err: err}
	}()
	select {
	case result := <-results:
		return result.resources, result.err
	case <-ctx.Done():
		verbose.Logf("AWS deterministic IAM lookup timed out for session=%s: %v", sessionID, ctx.Err())
		return nil, nil
	}
}

func (t *ResourceTracker) findTaggedAutoScalingGroups(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	if t.autoscaling == nil || strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	var resources []Resource
	var token *string
	for {
		verbose.Logf("AWS autoscaling:DescribeAutoScalingGroups page_token=%q", aws.ToString(token))
		out, err := t.autoscaling.DescribeAutoScalingGroups(ctx, &autoscaling.DescribeAutoScalingGroupsInput{
			NextToken: token,
		})
		if err != nil {
			return nil, fmt.Errorf("list Auto Scaling groups for tagged discovery: %w", err)
		}
		for i := range out.AutoScalingGroups {
			group := &out.AutoScalingGroups[i]
			tags := AutoScalingTagMap(group.Tags)
			if !tagsMatchScope(tags, owner, sessionID) {
				continue
			}
			arn := strings.TrimSpace(aws.ToString(group.AutoScalingGroupARN))
			if arn == "" {
				continue
			}
			resources = append(resources, Resource{
				ARN:       arn,
				Type:      ResourceAutoScalingGroup,
				Region:    firstNonEmpty(ParseResourceRegion(arn), t.region),
				SessionID: tags[TagSession],
				Owner:     tags[TagOwner],
			})
		}
		token = out.NextToken
		if token == nil || aws.ToString(token) == "" {
			break
		}
	}
	return resources, nil
}

// AutoScalingTagMap copies Auto Scaling tag descriptions into a key/value map.
func AutoScalingTagMap(tags []autoscalingtypes.TagDescription) map[string]string {
	out := make(map[string]string, len(tags))
	for _, tag := range tags {
		out[aws.ToString(tag.Key)] = aws.ToString(tag.Value)
	}
	return out
}

func (t *ResourceTracker) findTaggedIAM(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	if t.iam == nil || strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	lister, ok := t.iam.(IAMListTagAPI)
	if !ok {
		return nil, nil
	}
	// A shared account snapshot answers this from memory. Without it every
	// candidate re-scans the whole account's IAM, which is where the throttling
	// and the timeout budget go.
	ctx, cancel := context.WithTimeout(ctx, taggedIAMLookupTimeout)
	defer cancel()
	if t.iamInventory != nil {
		// Filtering is in memory, but following a matched instance profile's roles
		// still calls IAM, so keep the same budget the live scan had.
		return t.iamInventory.resourcesForScope(ctx, t, lister, owner, sessionID)
	}

	var resources []Resource
	profileResources, err := t.findTaggedIAMInstanceProfiles(ctx, lister, owner, sessionID)
	if err != nil {
		// Degrade on timeout exactly like the role and policy lookups below. Without
		// this, a slow account-wide instance-profile scan is the one branch here that
		// hard-fails, which blocks the whole account's cleanup even though every
		// session's deletes succeeded.
		if ctx.Err() != nil {
			verbose.Logf("AWS tagged IAM instance profile lookup timed out for session=%s: %v", sessionID, err)
			return resources, nil
		}
		return nil, err
	}
	resources = append(resources, profileResources...)

	roleResources, err := t.findTaggedIAMRoles(ctx, lister, owner, sessionID)
	if err != nil {
		if ctx.Err() != nil {
			verbose.Logf("AWS tagged IAM role lookup timed out for session=%s: %v", sessionID, err)
			return resources, nil
		}
		return nil, err
	}
	resources = append(resources, roleResources...)

	policyResources, err := t.findTaggedIAMPolicies(ctx, lister, owner, sessionID)
	if err != nil {
		if ctx.Err() != nil {
			verbose.Logf("AWS tagged IAM policy lookup timed out for session=%s: %v", sessionID, err)
			return resources, nil
		}
		return nil, err
	}
	resources = append(resources, policyResources...)

	return resources, nil
}

func (t *ResourceTracker) findSessionNamedIAM(
	ctx context.Context,
	owner string,
	sessionID string,
) ([]Resource, error) {
	if t.iam == nil || strings.TrimSpace(sessionID) == "" {
		return nil, nil
	}
	lister, ok := t.iam.(IAMListTagAPI)
	if !ok {
		return nil, nil
	}
	resources, err := t.findTaggedIAMInstanceProfiles(ctx, lister, owner, sessionID)
	if err != nil {
		return nil, err
	}
	if len(resources) > 0 {
		return resources, nil
	}
	match := func(name string) bool {
		return name == sessionID || strings.HasPrefix(name, sessionID+"-") ||
			name == "cp9-"+sessionID || strings.HasPrefix(name, "cp9-"+sessionID+"-")
	}
	resources, err = t.findTaggedIAMRolesMatching(ctx, lister, owner, sessionID, match)
	if err != nil {
		return nil, err
	}
	if len(resources) > 0 {
		return resources, nil
	}
	return t.findTaggedIAMPoliciesMatching(ctx, lister, owner, sessionID, match)
}

func (t *ResourceTracker) findTaggedIAMRoles(
	ctx context.Context,
	lister IAMListTagAPI,
	owner string,
	sessionID string,
) ([]Resource, error) {
	return t.findTaggedIAMRolesMatching(ctx, lister, owner, sessionID, nil)
}

func (t *ResourceTracker) findTaggedIAMRolesMatching(
	ctx context.Context,
	lister IAMListTagAPI,
	owner string,
	sessionID string,
	match func(string) bool,
) ([]Resource, error) {
	var resources []Resource
	roleInput := &iam.ListRolesInput{}
	for {
		out, err := lister.ListRoles(ctx, roleInput)
		if err != nil {
			if isAuthorizationErr(err) {
				verbose.Logf("AWS tagged IAM role list unavailable: %v", err)
				break
			}
			return nil, fmt.Errorf("list tagged IAM roles: %w", err)
		}
		for _, role := range out.Roles {
			roleName := aws.ToString(role.RoleName)
			roleARN := aws.ToString(role.Arn)
			if roleName == "" || roleARN == "" || (match != nil && !match(roleName)) {
				continue
			}
			roleTags, err := t.listRoleTags(ctx, roleName)
			if err != nil {
				if isNotFoundErr(err) || isAuthorizationErr(err) {
					continue
				}
				return nil, fmt.Errorf("list tagged IAM role tags %s: %w", roleName, err)
			}
			if !tagsMatchScope(roleTags, owner, sessionID) {
				continue
			}
			resources = append(resources, Resource{
				ARN:       roleARN,
				Type:      ResourceIAMRole,
				Region:    t.region,
				SessionID: roleTags[TagSession],
				Owner:     roleTags[TagOwner],
			})
		}
		if !out.IsTruncated || out.Marker == nil {
			break
		}
		roleInput.Marker = out.Marker
	}
	return resources, nil
}

func (t *ResourceTracker) findTaggedIAMInstanceProfiles(
	ctx context.Context,
	lister IAMListTagAPI,
	owner string,
	sessionID string,
) ([]Resource, error) {
	var resources []Resource
	profileInput := &iam.ListInstanceProfilesInput{}
	for {
		out, err := lister.ListInstanceProfiles(ctx, profileInput)
		if err != nil {
			if isAuthorizationErr(err) {
				verbose.Logf("AWS tagged IAM instance profile list unavailable: %v", err)
				break
			}
			return nil, fmt.Errorf("list tagged IAM instance profiles: %w", err)
		}
		for i := range out.InstanceProfiles {
			profile := &out.InstanceProfiles[i]
			profileARN := aws.ToString(profile.Arn)
			if profileARN == "" {
				continue
			}
			tags := map[string]string{}
			copyIAMTags(tags, profile.Tags)
			if !tagsMatchScope(tags, owner, sessionID) {
				continue
			}
			resources = append(resources, Resource{
				ARN:       profileARN,
				Type:      ResourceIAMInstanceProfile,
				Region:    t.region,
				SessionID: tags[TagSession],
				Owner:     tags[TagOwner],
			})
			profileResources, err := t.findIAMResourcesForTaggedInstanceProfile(ctx, lister, profile, tags)
			if err != nil {
				return nil, err
			}
			resources = append(resources, profileResources...)
		}
		if !out.IsTruncated || out.Marker == nil {
			break
		}
		profileInput.Marker = out.Marker
	}
	return resources, nil
}

func (t *ResourceTracker) findTaggedIAMPolicies(
	ctx context.Context,
	lister IAMListTagAPI,
	owner string,
	sessionID string,
) ([]Resource, error) {
	return t.findTaggedIAMPoliciesMatching(ctx, lister, owner, sessionID, nil)
}

func (t *ResourceTracker) findTaggedIAMPoliciesMatching(
	ctx context.Context,
	lister IAMListTagAPI,
	owner string,
	sessionID string,
	match func(string) bool,
) ([]Resource, error) {
	var resources []Resource
	policyInput := &iam.ListPoliciesInput{Scope: iamtypes.PolicyScopeTypeLocal}
	for {
		out, err := lister.ListPolicies(ctx, policyInput)
		if err != nil {
			if isAuthorizationErr(err) {
				verbose.Logf("AWS tagged IAM policy list unavailable: %v", err)
				break
			}
			return nil, fmt.Errorf("list tagged IAM policies: %w", err)
		}
		for _, policy := range out.Policies {
			if match != nil && !match(aws.ToString(policy.PolicyName)) {
				continue
			}
			policyARN := aws.ToString(policy.Arn)
			if policyARN == "" {
				continue
			}
			tags, err := t.listPolicyTags(ctx, policyARN)
			if err != nil {
				if isNotFoundErr(err) || isAuthorizationErr(err) {
					continue
				}
				return nil, fmt.Errorf("list tagged IAM policy tags %s: %w", policyARN, err)
			}
			if !tagsMatchScope(tags, owner, sessionID) {
				continue
			}
			resources = append(resources, Resource{
				ARN:       policyARN,
				Type:      ResourceIAMPolicy,
				Region:    t.region,
				SessionID: tags[TagSession],
				Owner:     tags[TagOwner],
			})
		}
		if !out.IsTruncated || out.Marker == nil {
			break
		}
		policyInput.Marker = out.Marker
	}

	return resources, nil
}

func (t *ResourceTracker) findIAMResourcesForTaggedInstanceProfile(
	ctx context.Context,
	lister IAMListTagAPI,
	profile *iamtypes.InstanceProfile,
	scopeTags map[string]string,
) ([]Resource, error) {
	if profile == nil {
		return nil, nil
	}
	profileName := aws.ToString(profile.InstanceProfileName)
	sessionID := scopeTags[TagSession]
	owner := scopeTags[TagOwner]
	var resources []Resource
	for _, role := range profile.Roles {
		roleName := aws.ToString(role.RoleName)
		roleARN := aws.ToString(role.Arn)
		if roleName == "" || roleARN == "" {
			continue
		}
		if !taggedInstanceProfileOwnsRole(profileName, roleName, sessionID) {
			continue
		}
		resources = append(resources, Resource{
			ARN:       roleARN,
			Type:      ResourceIAMRole,
			Region:    t.region,
			SessionID: sessionID,
			Owner:     owner,
		})
		policies, err := t.findLocalPoliciesAttachedToTaggedRole(ctx, lister, roleName, sessionID)
		if err != nil {
			return nil, err
		}
		for _, policyARN := range policies {
			resources = append(resources, Resource{
				ARN:       policyARN,
				Type:      ResourceIAMPolicy,
				Region:    t.region,
				SessionID: sessionID,
				Owner:     owner,
			})
		}
	}
	return resources, nil
}

func (t *ResourceTracker) findLocalPoliciesAttachedToTaggedRole(
	ctx context.Context,
	lister IAMListTagAPI,
	roleName string,
	sessionID string,
) ([]string, error) {
	var policies []string
	input := &iam.ListAttachedRolePoliciesInput{RoleName: aws.String(roleName)}
	for {
		out, err := lister.ListAttachedRolePolicies(ctx, input)
		if err != nil {
			if isNotFoundErr(err) || isAuthorizationErr(err) {
				return policies, nil
			}
			return nil, fmt.Errorf("list tagged IAM profile role policies %s: %w", roleName, err)
		}
		for _, policy := range out.AttachedPolicies {
			policyARN := aws.ToString(policy.PolicyArn)
			policyName := aws.ToString(policy.PolicyName)
			if !taggedInstanceProfileOwnsPolicy(policyARN, policyName, roleName, sessionID) {
				continue
			}
			policies = append(policies, policyARN)
		}
		if !out.IsTruncated || out.Marker == nil {
			break
		}
		input.Marker = out.Marker
	}
	return policies, nil
}

func taggedInstanceProfileOwnsRole(profileName, roleName, sessionID string) bool {
	profileName = strings.TrimSpace(profileName)
	roleName = strings.TrimSpace(roleName)
	sessionID = strings.TrimSpace(sessionID)
	if profileName == "" || roleName == "" || sessionID == "" {
		return false
	}
	return roleName == profileName || strings.Contains(roleName, sessionID)
}

func taggedInstanceProfileOwnsPolicy(policyARN, policyName, roleName, sessionID string) bool {
	policyARN = strings.TrimSpace(policyARN)
	policyName = strings.TrimSpace(policyName)
	roleName = strings.TrimSpace(roleName)
	sessionID = strings.TrimSpace(sessionID)
	if policyARN == "" || strings.HasPrefix(policyARN, "arn:aws:iam::aws:policy/") {
		return false
	}
	return strings.Contains(policyName, roleName) || strings.Contains(policyName, sessionID)
}

func (t *ResourceTracker) findDeterministicIAMOnce(ctx context.Context, owner, sessionID string) ([]Resource, error) {
	var resources []Resource
	accountID := strings.TrimSpace(t.accountID)
	for _, roleName := range deterministicIAMRoleNames(sessionID) {
		verbose.Logf("AWS iam:GetRole deterministic-discovery role=%s", roleName)
		roleOut, err := t.iam.GetRole(ctx, &iam.GetRoleInput{RoleName: aws.String(roleName)})
		switch {
		case err == nil:
		case isNotFoundErr(err):
			continue
		case isAuthorizationErr(err):
			verbose.Logf("AWS deterministic IAM role metadata unavailable for %s: %v", roleName, err)
			continue
		default:
			return nil, fmt.Errorf("get deterministic IAM role %s: %w", roleName, err)
		}
		if roleOut.Role == nil || aws.ToString(roleOut.Role.Arn) == "" {
			continue
		}
		roleTags, err := t.listRoleTags(ctx, roleName)
		if err != nil {
			return nil, fmt.Errorf("list deterministic IAM role tags %s: %w", roleName, err)
		}
		if !tagsMatchScope(roleTags, owner, sessionID) {
			continue
		}
		roleARN := aws.ToString(roleOut.Role.Arn)
		if parsedAccountID := accountIDFromARN(roleARN); parsedAccountID != "" {
			accountID = parsedAccountID
		}
		resources = append(resources, Resource{
			ARN:       roleARN,
			Type:      ResourceIAMRole,
			Region:    t.region,
			SessionID: roleTags[TagSession],
			Owner:     roleTags[TagOwner],
		})
	}

	if accountID == "" {
		return resources, nil
	}
	for _, policyName := range []string{
		fmt.Sprintf("cp9-%s-workload", sessionID),
		fmt.Sprintf("cp9-%s-dataplane-manager", sessionID),
	} {
		policyARN := fmt.Sprintf("arn:aws:iam::%s:policy/%s", accountID, policyName)
		policy, ok, err := t.findDeterministicPolicy(ctx, policyARN, owner, sessionID)
		if err != nil {
			return nil, err
		}
		if ok {
			resources = append(resources, policy)
		}
	}
	return resources, nil
}

func deterministicIAMRoleNames(sessionID string) []string {
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		return nil
	}
	roleNames := []string{
		fmt.Sprintf("cp9-%s-workload", sessionID),
		fmt.Sprintf("cp9-%s-dataplane-manager", sessionID),
	}
	if serviceRoleName := deterministicEKSServiceRoleName(sessionID); serviceRoleName != "" {
		roleNames = append(roleNames, serviceRoleName)
	}
	return roleNames
}

func deterministicEKSServiceRoleName(sessionID string) string {
	const maxIAMRoleNameLen = 64
	sessionID = sanitizeIAMRoleNameSegment(sessionID)
	if sessionID == "" {
		return ""
	}
	name := strings.Trim(sessionID+"-eks-service", "-")
	if len(name) <= maxIAMRoleNameLen {
		return name
	}
	return strings.Trim(name[:maxIAMRoleNameLen], "-")
}

func sanitizeIAMRoleNameSegment(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return ""
	}
	var b strings.Builder
	lastHyphen := false
	for _, r := range value {
		ok := r >= 'a' && r <= 'z' ||
			r >= 'A' && r <= 'Z' ||
			r >= '0' && r <= '9'
		if ok {
			_, _ = b.WriteRune(r)
			lastHyphen = false
			continue
		}
		if !lastHyphen {
			_ = b.WriteByte('-')
			lastHyphen = true
		}
	}
	return strings.Trim(b.String(), "-")
}

func (t *ResourceTracker) listRoleTags(ctx context.Context, roleName string) (map[string]string, error) {
	tags := map[string]string{}
	var marker *string
	for {
		verbose.Logf(
			"AWS iam:ListRoleTags deterministic-discovery role=%s marker=%q",
			roleName,
			aws.ToString(marker),
		)
		out, err := t.iam.ListRoleTags(ctx, &iam.ListRoleTagsInput{
			RoleName: aws.String(roleName),
			Marker:   marker,
		})
		if err != nil {
			if isNotFoundErr(err) {
				return map[string]string{}, nil
			}
			return nil, err
		}
		copyIAMTags(tags, out.Tags)
		if !out.IsTruncated || out.Marker == nil {
			break
		}
		marker = out.Marker
	}
	return tags, nil
}

func (t *ResourceTracker) findDeterministicPolicy(
	ctx context.Context,
	policyARN string,
	owner string,
	sessionID string,
) (Resource, bool, error) {
	out, err := t.iam.GetPolicy(ctx, &iam.GetPolicyInput{PolicyArn: aws.String(policyARN)})
	if err != nil {
		if isNotFoundErr(err) {
			return Resource{}, false, nil
		}
		return Resource{}, false, fmt.Errorf("get deterministic IAM policy %s: %w", policyARN, err)
	}
	if out.Policy == nil || aws.ToString(out.Policy.Arn) == "" {
		return Resource{}, false, nil
	}
	tags, err := t.listPolicyTags(ctx, policyARN)
	if err != nil {
		return Resource{}, false, fmt.Errorf("list deterministic IAM policy tags %s: %w", policyARN, err)
	}
	if !tagsMatchScope(tags, owner, sessionID) {
		return Resource{}, false, nil
	}
	return Resource{
		ARN:       aws.ToString(out.Policy.Arn),
		Type:      ResourceIAMPolicy,
		Region:    t.region,
		SessionID: tags[TagSession],
		Owner:     tags[TagOwner],
	}, true, nil
}

func (t *ResourceTracker) listPolicyTags(ctx context.Context, policyARN string) (map[string]string, error) {
	tags := map[string]string{}
	var marker *string
	for {
		out, err := t.iam.ListPolicyTags(ctx, &iam.ListPolicyTagsInput{
			PolicyArn: aws.String(policyARN),
			Marker:    marker,
		})
		if err != nil {
			if isNotFoundErr(err) {
				return map[string]string{}, nil
			}
			return nil, err
		}
		copyIAMTags(tags, out.Tags)
		if !out.IsTruncated || out.Marker == nil {
			break
		}
		marker = out.Marker
	}
	return tags, nil
}

func copyIAMTags(dst map[string]string, tags []iamtypes.Tag) {
	for _, tag := range tags {
		dst[aws.ToString(tag.Key)] = aws.ToString(tag.Value)
	}
}

func tagsMatchScope(tags map[string]string, owner, sessionID string) bool {
	if strings.TrimSpace(tags[TagSession]) != strings.TrimSpace(sessionID) {
		return false
	}
	if strings.TrimSpace(owner) != "" && !ownerTagMatches(owner, tags[TagOwner]) {
		return false
	}
	switch strings.TrimSpace(tags[TagManaged]) {
	case TagManagedTerraformSubstrate, "true":
		return true
	default:
		return false
	}
}

func ownerTagFilterValues(owner string) []string {
	owner = strings.TrimSpace(owner)
	if owner == "" {
		return []string{owner}
	}
	values := []string{owner}
	for _, safe := range []string{kubernetesSafeTagValue(owner), awsSafeTagValue(owner)} {
		if safe == "" || containsFold(values, safe) {
			continue
		}
		values = append(values, safe)
	}
	return values
}

// OwnerTagMatches reports whether an AWS cp9-owner tag value matches the raw owner.
// Regional-server-originated AWS tags may contain the sanitized value that was
// first valid as a Kubernetes label and then forwarded through regional-server's
// stricter tag helper.
func OwnerTagMatches(want, got string) bool {
	return ownerTagMatches(want, got)
}

func ownerTagMatches(want, got string) bool {
	want = strings.TrimSpace(want)
	got = strings.TrimSpace(got)
	if want == "" {
		return true
	}
	if strings.EqualFold(got, want) {
		return true
	}
	for _, safe := range []string{kubernetesSafeTagValue(want), awsSafeTagValue(want)} {
		if strings.EqualFold(got, safe) {
			return true
		}
	}
	return false
}

func containsFold(values []string, want string) bool {
	for _, value := range values {
		if strings.EqualFold(value, want) {
			return true
		}
	}
	return false
}

func kubernetesSafeTagValue(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return "cp9"
	}
	value = kubernetesSafeTagValueReplacer.ReplaceAllString(value, "-")
	value = strings.Trim(value, "-_.")
	if value == "" {
		return "cp9"
	}
	if len(value) > 63 {
		value = value[:63]
		value = strings.Trim(value, "-_.")
		if value == "" {
			return "cp9"
		}
	}
	return value
}

func awsSafeTagValue(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return "cp9"
	}
	value = awsSafeTagValueReplacer.ReplaceAllString(value, "-")
	value = strings.Trim(value, "-")
	if value == "" {
		return "cp9"
	}
	if len(value) > 35 {
		value = value[:35]
		value = strings.Trim(value, "-")
		if value == "" {
			return "cp9"
		}
	}
	return value
}

func accountIDFromARN(arn string) string {
	parts := strings.SplitN(arn, ":", 6)
	if len(parts) < 5 {
		return ""
	}
	return parts[4]
}

func classicELBARN(region, accountID, name string) string {
	return fmt.Sprintf("arn:aws:elasticloadbalancing:%s:%s:loadbalancer/%s", region, accountID, name)
}

func mergeTrackerResources(primary, extra []Resource) []Resource {
	if len(extra) == 0 {
		return primary
	}
	merged := append([]Resource(nil), primary...)
	seen := make(map[string]struct{}, len(merged)+len(extra))
	for i := range merged {
		seen[resourceDedupeKey(&merged[i])] = struct{}{}
	}
	for i := range extra {
		resource := extra[i]
		key := resourceDedupeKey(&resource)
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		merged = append(merged, resource)
	}
	return merged
}

func resourceDedupeKey(resource *Resource) string {
	switch resource.Type {
	case ResourceIAMInstanceProfile, ResourceIAMRole, ResourceIAMPolicy:
		return string(resource.Type) + "\x00" + resource.ARN
	default:
		return resource.Region + "\x00" + string(resource.Type) + "\x00" + resource.ARN
	}
}

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