map2-auto-tagger-optimized.yaml

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

AWSTemplateFormatVersion: '2010-09-09'
Description: >
  MAP 2.0 Auto-Tagger v21.0.6 - Auto-tags AWS resources with map-migrated for MAP 2.0
  credit eligibility. 190+ resource types. Deploy once; tagging happens within 60-90 s
  of resource creation. Daily reconciliation Lambda (RGTA-based) catches any tags the
  live Lambda missed. Three-path error classifier + TagFailureByClass CloudWatch metric
  distinguishes transient / permanent-ignorable / permanent-actionable failures.
  Pipeline: EventBridge -> SQS (14-day retention) -> Lambda with DLQ + SNS alarm.
  See CHANGELOG.md for full history.

Parameters:
  MpeId:
    Type: String
    Description: MAP 2.0 Migration Program Engagement ID (e.g., mig1234567890)
    AllowedPattern: ^mig[a-zA-Z0-9]+$
    MaxLength: 20
    ConstraintDescription: Must start with 'mig' followed by alphanumeric characters
  AgreementStartDate:
    Type: String
    Description: MAP 2.0 agreement start date (YYYY-MM-DD)
    AllowedPattern: ^(19|20)\d{2}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$
  AgreementEndDate:
    Type: String
    Description: MAP 2.0 agreement end date (YYYY-MM-DD). Resources created after this date will NOT be tagged.
    AllowedPattern: ^(19|20)\d{2}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$
    Default: "2099-12-31"
  ScopeMode:
    Type: String
    Description: How to determine which resources to tag
    Default: account
    AllowedValues:
      - account
      - vpc
  ScopedAccountIds:
    Type: CommaDelimitedList
    Description: >
      Comma-separated AWS account IDs in MAP scope (used when ScopeMode=account).
      Leave as 'ALL' to tag all resources in this account.
    Default: ALL
  ScopedVpcIds:
    Type: CommaDelimitedList
    Description: >
      Comma-separated VPC IDs in MAP scope (used when ScopeMode=vpc).
      Only resources in these VPCs will be tagged.
    Default: NONE
  ReconciliationInterval:
    Type: String
    Description: >
      Schedule for the reconciliation Lambda (daily safety-net that enumerates
      in-scope resources via RGTA GetResources and re-enqueues any whose
      map-migrated tag is missing or wrong). EventBridge schedule expression.
    Default: 'rate(24 hours)'
    AllowedPattern: '^(rate\(\d+ (hour|hours|day|days)\)|cron\(.+\))$'

Conditions:
  IsAccountAll: !Equals [!Join [',', !Ref ScopedAccountIds], 'ALL']
  IsVpcNone: !Equals [!Join [',', !Ref ScopedVpcIds], 'NONE']

Resources:

  # SSM Parameter Store - pinned template version (read by ops, upgrade.sh, and
  # the Lambda at cold start). Zero outbound network calls — version is written
  # at deploy time and discovered locally via ssm:GetParameter.
  MapVersion:
    Type: AWS::SSM::Parameter
    Properties:
      Name: !Sub '/auto-map-tagger/${MpeId}/version'
      Type: String
      Description: MAP 2.0 Auto-Tagger template version pinned at deploy time
      Value: v21.0.6

  # SSM Parameter Store - single source of truth for config.
  # Tier: Intelligent-Tiering leaves the parameter in the free Standard tier
  # (4KB limit) until the Value crosses the threshold, at which point AWS
  # auto-upgrades to Advanced (8KB limit, $0.05/parameter/month). Closes
  # §1.60: customers with ~240+ accounts in `scoped_account_ids` generate a
  # Value > 4KB; prior behavior (default Standard) silently failed stack
  # create with `ParameterMaxSizeExceeded`.
  MapConfig:
    Type: AWS::SSM::Parameter
    Properties:
      Name: !Sub '/auto-map-tagger/${MpeId}/config'
      Tier: Intelligent-Tiering
      Type: String
      Description: MAP 2.0 Auto-Tagger configuration
      Value: !Sub
        - |
          {
            "mpe_id": "${MpeId}",
            "agreement_start_date": "${AgreementStartDate}",
            "agreement_end_date": "${AgreementEndDate}",
            "scope_mode": "${ScopeMode}",
            "scoped_account_ids": ${AccountIdsArray},
            "scoped_vpc_ids": ${VpcIdsArray}
          }
        - AccountIdsArray: !If
            - IsAccountAll
            - '["ALL"]'
            - !Sub
                - '["${Joined}"]'
                - Joined: !Join ['","', !Ref ScopedAccountIds]
          VpcIdsArray: !If
            - IsVpcNone
            - '[]'
            - !Sub
                - '["${Joined}"]'
                - Joined: !Join ['","', !Ref ScopedVpcIds]

  # Lambda execution role
  AutoTaggerRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub 'map-auto-tagger-role-${MpeId}-${AWS::Region}'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: map-auto-tagger-policy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              # SECURITY NOTE: Resource: '*' is required for tagging actions.
              # AWS Tag Editor API (tag:TagResources) does not support resource-level
              # permissions — it must be scoped to '*'. This is consistent with
              # AWS-managed tagging solutions (e.g., AWS Tag Editor console, MAP Taggr).
              # The Lambda only applies the map-migrated tag and has no create/delete/update
              # permissions outside of tagging. IAM Access Analyzer: 0 findings.
              - Sid: UniversalTagging
                Effect: Allow
                Action:
                  - tag:TagResources
                  - tag:GetResources
                Resource: '*'
              - Sid: ServiceSpecificTagging
                Effect: Allow
                Action:
                  # --- Compute ---
                  - ec2:CreateTags
                  - autoscaling:CreateOrUpdateTags
                  - ecs:TagResource
                  - eks:TagResource
                  - lambda:TagResource
                  - elasticbeanstalk:AddTags
                  - emr-serverless:TagResource
                  - elasticmapreduce:AddTags
                  # AWS Batch (§1.27) — job queues, compute environments, job
                  # definitions. RGTA dispatches tag:TagResources on these to
                  # batch:TagResource per the AWS service-authorization matrix;
                  # without this grant every Batch CreateJobQueue / CreateComputeEnvironment
                  # / RegisterJobDefinition fell through to AccessDenied.
                  - batch:TagResource
                  # --- Storage ---
                  - s3:PutBucketTagging
                  - s3:GetBucketTagging
                  - elasticfilesystem:TagResource
                  - fsx:TagResource
                  - ecr:TagResource
                  - backup:TagResource
                  # --- Database ---
                  - rds:AddTagsToResource
                  - dynamodb:TagResource
                  - elasticache:AddTagsToResource
                  - memorydb:TagResource
                  - redshift:CreateTags
                  - redshift-serverless:TagResource
                  - es:AddTags
                  - kafka:TagResource
                  - dms:AddTagsToResource
                  - cassandra:TagResource
                  - mq:CreateTags
                  # --- Networking ---
                  - elasticloadbalancing:AddTags
                  - globalaccelerator:TagResource
                  - cloudfront:TagResource
                  - route53:ChangeTagsForResource
                  - network-firewall:TagResource
                  - directconnect:TagResource
                  - appmesh:TagResource
                  # VPC Lattice: RGTA dispatches to vpc-lattice:TagResource per
                  # AWS auth matrix. Without this grant, COVERAGE.md's VPC
                  # Lattice claim was backed by RGTA fallthrough that silently
                  # AccessDenied on every CreateServiceNetwork event (D7).
                  - vpc-lattice:TagResource
                  # --- Analytics ---
                  - kinesis:AddTagsToStream
                  - firehose:TagDeliveryStream
                  - kinesisanalytics:TagResource
                  - glue:TagResource
                  - glue:GetDatabase
                  - databrew:TagResource
                  - athena:TagResource
                  # --- Integration ---
                  - sns:TagResource
                  - sqs:TagQueue
                  - states:TagResource
                  - appsync:TagResource
                  # SECURITY NOTE: API Gateway v1 (REST APIs) tags via PUT /tags/{arn}.
                  # AWS maps this to 'apigateway:PUT' — no narrower IAM action exists for v1 tagging.
                  # Lambda code only calls tagging APIs; no other API GW mutation code paths exist.
                  - apigateway:PUT
                  # v1 REST APIs also use PATCH (AWS routes tag:TagResources
                  # through PATCH /tags/{arn} internally).
                  - apigateway:PATCH
                  # v2 HTTP APIs actually use POST /tags/{arn} (confirmed via
                  # AccessDenied testing on PR #8 iteration 2). The v1/v2
                  # routing for tagging is POST for v2, PATCH for v1.
                  - apigateway:POST
                  # --- Management & Monitoring ---
                  - logs:TagResource
                  - cloudwatch:TagResource
                  - ssm:AddTagsToResource
                  - secretsmanager:TagResource
                  - servicediscovery:TagResource
                  # --- Security & Identity ---
                  - kms:TagResource
                  - acm:AddTagsToCertificate
                  - acm-pca:TagCertificateAuthority
                  - cognito-idp:TagResource
                  - cognito-identity:TagResource
                  - securityhub:TagResource
                  - wafv2:TagResource
                  # --- Developer Tools ---
                  - codepipeline:TagResource
                  - codedeploy:TagResource
                  - cloud9:TagResource
                  # SECURITY NOTE: CodeBuild has no standalone TagResource IAM action.
                  # AWS requires UpdateProject to apply tags (confirmed via AccessDenied testing).
                  # BatchGetProjects resolves project ARNs. Lambda code only calls tagging APIs.
                  - codebuild:UpdateProject
                  - codebuild:BatchGetProjects
                  # SECURITY NOTE: AWS routes tag:TagResources on CF stacks through UpdateStack
                  # internally (confirmed via AccessDenied testing). Lambda code only calls
                  # tag:TagResources; UpdateStack is never invoked directly by the Lambda.
                  - cloudformation:TagResource
                  - cloudformation:UpdateStack
                  - cloudformation:UpdateStackSet
                  # SECURITY NOTE: AWS routes Service Catalog tagging through UpdateProduct /
                  # UpdatePortfolio internally (confirmed via AccessDenied testing in v14).
                  # Lambda code only calls tag:TagResources; Update* is never invoked directly.
                  - servicecatalog:TagResource
                  - servicecatalog:UpdatePortfolio
                  - servicecatalog:UpdateProduct
                  # --- Migration & Transfer ---
                  - transfer:TagResource
                  - datasync:TagResource
                  - storagegateway:AddTagsToResource
                  # --- ML / AI ---
                  - sagemaker:AddTags
                  - comprehend:TagResource
                  - kendra:TagResource
                  - bedrock:TagResource
                  - bedrock-agentcore:TagResource
                  - braket:TagResource
                  # --- IoT ---
                  - iot:TagResource
                  - iotanalytics:TagResource
                  - iotevents:TagResource
                  # EventBridge (events:*) — distinct service from IoT Events (iotevents:*).
                  # Needed to tag newly-created Event rules / buses / schedules / connections.
                  - events:TagResource
                  - iotsitewise:TagResource
                  # --- Media ---
                  - mediaconvert:TagResource
                  - medialive:CreateTags
                  - mediapackage:TagResource
                  # --- Other ---
                  - ram:TagResource
                  - appstream:TagResource
                  - workspaces:CreateTags
                  - workspaces-web:TagResource
                  - quicksight:TagResource
                  - connect:TagResource
                  - managedblockchain:TagResource
                  - gamelift:TagResource
                  - timestream:TagResource
                  - healthlake:TagResource
                  - omics:TagResource
                  - dax:TagResource
                  - drs:TagResource
                  - deadline:TagResource
                  - kinesisvideo:TagStream
                  - dsql:TagResource
                  - payment-cryptography:TagResource
                  - networkmanager:TagResource
                  # --- Tier 1 MAP services (PR #25) ---
                  # Keyspaces (Cassandra-compatible) — RGTA support is unreliable
                  # for cassandra resources, so Lambda calls keyspaces:TagResource
                  # directly. cassandra:TagResource already present above; adding
                  # cassandra:Alter (required by the service-authorization matrix
                  # for keyspaces:TagResource — AccessDenied at runtime without it,
                  # live-confirmed as §1.99 in the remediation plan) + ds + cloudhsm.
                  - cassandra:Alter
                  - ds:AddTagsToResource
                  - cloudhsm:TagResource
                Resource: '*'
              # NOTE: This template deploys a single-account Lambda. No sts:AssumeRole or
              # cross-account permissions are present or required. For multi-account deployments,
              # a separate CloudFormation StackSet deploys this same template to each account
              # independently via CloudFormation SERVICE_MANAGED permissions (AWS service-linked
              # roles), not direct cross-account role assumption by the Lambda function itself.
              - Sid: MinimalReads
                Effect: Allow
                Action:
                  # VPC scope checking (only used when ScopeMode=vpc)
                  - ec2:DescribeInstances
                  - ec2:DescribeVolumes
                  # Determine local account ID for cross-account tagging logic
                  - sts:GetCallerIdentity
                  # SSM TagResource validates the param name via GetParameters
                  # internally (AccessDenied without it). Lambda code only
                  # calls tag:TagResources.
                  - ssm:GetParameters
                  # CFN TagResource validates the stack ARN via DescribeStacks
                  # internally (AccessDenied without it). Lambda code only
                  # calls tag:TagResources.
                  - cloudformation:DescribeStacks
                  # IAM role tagging (iam:TagRole not covered by tag:TagResources)
                  - iam:TagRole
                Resource: '*'
              - Sid: SqsEventSource
                Effect: Allow
                Action:
                  - sqs:ReceiveMessage
                  - sqs:DeleteMessage
                  - sqs:GetQueueAttributes
                  - sqs:SendMessage
                Resource:
                  - !GetAtt EventQueue.Arn
                  - !GetAtt EventDLQ.Arn
              - Sid: AlertPublish
                Effect: Allow
                Action:
                  - sns:Publish
                Resource: !Ref AlertTopic
              - Sid: EmitClassifierMetrics
                Effect: Allow
                Action:
                  - cloudwatch:PutMetricData
                # PutMetricData does not support resource-level IAM; '*' is
                # standard. Constrained to the MapAutoTagger namespace via
                # IAM Condition below.
                Resource: '*'
                Condition:
                  StringEquals:
                    cloudwatch:namespace: MapAutoTagger
              - Sid: ReadConfig
                Effect: Allow
                Action:
                  - ssm:GetParameter
                Resource: !Sub 'arn:aws:ssm:${AWS::Region}:${AWS::AccountId}:parameter/auto-map-tagger/${MpeId}/config'
              # Peer-tagger detection at cold start. Lists map-auto-tagger-mig*
              # stacks in this account/region to surface concurrent taggers
              # (§1.108 Phase 16). ListStacks has no resource-level IAM per AWS
              # IAM Service Authorization Reference; scope is implicitly the
              # caller's account. Read-only.
              - Sid: PeerTaggerDetect
                Effect: Allow
                Action:
                  - cloudformation:ListStacks
                Resource: '*'
              - Sid: Logging
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:*

  # ── Preflight: peer-tagger scope-overlap guard at stack-instance creation ──
  # Runs BEFORE AutoTaggerFunction is created. If another map-auto-tagger-mig*
  # stack exists in this account+region with an overlapping scope, fail the
  # Custom Resource → CFN rolls back this stack instance → no AutoTaggerFunction
  # is ever provisioned. Closes the §1.108 temporal race (Phase 16 Test 5,
  # evidence 0/50) that configurator.html preflight cannot catch: StackSet
  # AutoDeployment into newly-joined OU accounts, and member-account deploys
  # via multi-account StackSet where the member already has a peer stack.
  # Fail-open on any internal error (throttle / IAM propagation / region issue)
  # so a legitimate deploy is never blocked by a transient AWS condition;
  # PR #60's runtime detector + PeerTaggerDetectedAlarm remain as the second
  # backstop if the preflight's fail-open path let a real conflict through.
  PreflightRole:
    Type: AWS::IAM::Role
    Properties:
      # Name kept short (≤64 chars worst-case: 21+20+1+14 = 56) to match
      # AutoTaggerRole's budget. Regional suffix is necessary because IAM
      # role names are account-global and same-MPE stacks may deploy to
      # multiple regions in the same account.
      RoleName: !Sub 'map-preflight-role-${MpeId}-${AWS::Region}'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: map-auto-tagger-preflight-policy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: ListOwnStacks
                Effect: Allow
                Action:
                  - cloudformation:ListStacks
                Resource: '*'  # ListStacks has no resource-level IAM per AWS auth matrix
              - Sid: ReadPeerConfigs
                Effect: Allow
                Action:
                  - ssm:GetParameter
                Resource: !Sub 'arn:aws:ssm:${AWS::Region}:${AWS::AccountId}:parameter/auto-map-tagger/*/config'
              - Sid: Logging
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:*'

  PreflightLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      LogGroupName: !Sub '/aws/lambda/map-auto-tagger-preflight-${MpeId}'
      RetentionInDays: 14

  PreflightFunction:
    Type: AWS::Lambda::Function
    DependsOn: PreflightLogGroup
    Properties:
      FunctionName: !Sub 'map-auto-tagger-preflight-${MpeId}'
      Runtime: python3.12
      Handler: index.handler
      Role: !GetAtt PreflightRole.Arn
      Timeout: 60
      MemorySize: 256
      Code:
        ZipFile: |
          import json, os, boto3
          from urllib.request import urlopen, Request

          def respond(event, context, status, reason=''):
              body = json.dumps({
                  'Status': status,
                  'Reason': reason or 'See CloudWatch logs',
                  'PhysicalResourceId': context.log_stream_name,
                  'StackId': event['StackId'],
                  'RequestId': event['RequestId'],
                  'LogicalResourceId': event['LogicalResourceId'],
              })
              req = Request(event['ResponseURL'], data=body.encode(), method='PUT')
              req.add_header('Content-Type', '')
              req.add_header('Content-Length', str(len(body)))
              urlopen(req)

          def scope_overlap(new_mode, new_accts, new_vpcs, peer_mode, peer_accts, peer_vpcs, this_account):
              """Return conflict reason string if scopes overlap; '' otherwise.

              Mirrors the 5-case logic at configurator.html preflight
              (lines ~6473–6514). Both sides are already in the same
              account+region; overlap decisions reduce to scope dimensions.
              """
              new_accts = set(new_accts or [])
              peer_accts = set(peer_accts or [])
              new_vpcs = set(new_vpcs or [])
              peer_vpcs = set(peer_vpcs or [])
              if new_mode == 'account' and peer_mode == 'account':
                  if 'ALL' in peer_accts:
                      return f'peer scope=account/ALL dominates {this_account}'
                  if 'ALL' in new_accts:
                      return f'our scope=account/ALL dominates peer in {this_account}'
                  if this_account in peer_accts and (this_account in new_accts or 'ALL' in new_accts):
                      return f'peer scope includes {this_account}'
                  return ''
              if new_mode == 'account' and peer_mode == 'vpc':
                  # Our side claims whole account; peer claims specific VPCs in same account.
                  if 'ALL' in new_accts or this_account in new_accts:
                      return 'our account-mode dominates peer VPC-scope on shared VPCs'
                  return ''
              if new_mode == 'vpc' and peer_mode == 'account':
                  if 'ALL' in peer_accts or this_account in peer_accts:
                      return f'peer account-mode dominates our VPC-scope'
                  return ''
              # Both VPC-scoped
              overlap = new_vpcs & peer_vpcs
              if overlap:
                  return f'shared VPC(s): {sorted(overlap)}'
              return ''

          def check_peers(props, account, region):
              own_mpe = props['MpeId']
              new_mode = props.get('ScopeMode', 'account')
              new_accts = props.get('ScopedAccountIds') or ['ALL']
              if isinstance(new_accts, str):
                  new_accts = [s.strip() for s in new_accts.split(',') if s.strip()]
              new_vpcs = props.get('ScopedVpcIds') or []
              if isinstance(new_vpcs, str):
                  new_vpcs = [s.strip() for s in new_vpcs.split(',') if s.strip() and s.strip() != 'NONE']

              cfn = boto3.client('cloudformation')
              ssm = boto3.client('ssm')
              conflicts = []
              active = ['CREATE_COMPLETE', 'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_COMPLETE']
              for page in cfn.get_paginator('list_stacks').paginate(StackStatusFilter=active):
                  for s in page.get('StackSummaries', []):
                      name = s.get('StackName', '')
                      if name.startswith('StackSet-map-auto-tagger-mig'):
                          peer_mpe = name[len('StackSet-map-auto-tagger-'):]
                      elif name.startswith('map-auto-tagger-mig'):
                          peer_mpe = name[len('map-auto-tagger-'):]
                      else:
                          continue
                      import re
                      peer_mpe = re.sub(r'-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', '', peer_mpe)
                      if peer_mpe == own_mpe:
                          continue  # same MPE = in-place update, not a conflict
                      try:
                          p = ssm.get_parameter(Name=f'/auto-map-tagger/{peer_mpe}/config')
                          cfg = json.loads(p['Parameter']['Value'])
                      except Exception as e:
                          conflicts.append(f'peer {name} config unreadable ({e.__class__.__name__})')
                          continue
                      peer_mode = cfg.get('scope_mode', 'account')
                      peer_accts = cfg.get('scoped_account_ids') or ['ALL']
                      peer_vpcs = cfg.get('scoped_vpc_ids') or []
                      reason = scope_overlap(
                          new_mode, new_accts, new_vpcs,
                          peer_mode, peer_accts, peer_vpcs,
                          account,
                      )
                      if reason:
                          conflicts.append(f'{name} (MPE {peer_mpe}): {reason}')
              return conflicts

          def handler(event, context):
              try:
                  rt = event.get('RequestType', '')
                  if rt in ('Update', 'Delete'):
                      return respond(event, context, 'SUCCESS', f'{rt}: no-op')
                  # Detect upgrade: CFN sends Create for a new resource, but if a
                  # tagger stack with this MpeId already exists then the peer-Lambda
                  # is already live — this is a template upgrade, not a fresh deploy.
                  # Uses ListStacks (already in PreflightRole IAM) instead of
                  # DescribeStacks to avoid needing an extra IAM grant.
                  props = event.get('ResourceProperties', {})
                  own_mpe = props.get('MpeId', '')
                  if own_mpe:
                      own_stack = f'map-auto-tagger-{own_mpe}'
                      cfn_client = boto3.client('cloudformation')
                      active = ['CREATE_COMPLETE', 'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_COMPLETE', 'UPDATE_IN_PROGRESS']
                      for page in cfn_client.get_paginator('list_stacks').paginate(StackStatusFilter=active):
                          for s in page.get('StackSummaries', []):
                              if s['StackName'] == own_stack:
                                  return respond(event, context, 'SUCCESS',
                                                 f'Upgrade detected ({own_stack} exists): skip preflight')
                      # stack not found — genuine first deploy, proceed to check
                  account = context.invoked_function_arn.split(':')[4]
                  region = os.environ.get('AWS_REGION', '')
                  conflicts = check_peers(props, account, region)
                  if conflicts:
                      reason = (
                          'Peer tagger scope conflict — stack creation blocked '
                          f'to prevent §1.108 cross-Lambda contamination. '
                          f'Conflicts: {"; ".join(conflicts)[:800]}. '
                          'Resolve via: delete the peer stack, OR narrow this '
                          'deploy\'s scope to not overlap, OR narrow the peer\'s scope.'
                      )
                      print(f'PreflightConflict: {reason}')
                      return respond(event, context, 'FAILED', reason)
                  return respond(event, context, 'SUCCESS', 'No peer tagger scope conflict')
              except Exception as e:
                  # Fail-open: any internal error (throttle, IAM propagation,
                  # region transient) must NOT block a legitimate deploy. The
                  # runtime PeerTaggerDetected alarm (PR #60) remains as the
                  # second backstop if a real conflict slips through.
                  import traceback
                  print(traceback.format_exc())
                  return respond(event, context, 'SUCCESS', f'Preflight fail-open ({e.__class__.__name__}): {str(e)[:200]}')

  PreflightTrigger:
    Type: Custom::PeerTaggerPreflight
    # No explicit DependsOn — the ServiceToken reference below already
    # creates the implicit dependency on PreflightFunction. Adding an
    # explicit DependsOn produces cfn-lint W3005 (redundant dependency).
    Properties:
      ServiceToken: !GetAtt PreflightFunction.Arn
      MpeId: !Ref MpeId
      ScopeMode: !Ref ScopeMode
      ScopedAccountIds: !Ref ScopedAccountIds
      ScopedVpcIds: !Ref ScopedVpcIds

  # CloudWatch Log Group — explicit declaration sets retention and ensures the
  # group is cleaned up when the stack is deleted, so a customer can redeploy
  # with the same MpeId without hitting ResourceExistenceCheck.
  AutoTaggerLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      LogGroupName: !Sub '/aws/lambda/map-auto-tagger-${MpeId}'
      RetentionInDays: 14

  # Lambda function
  AutoTaggerFunction:
    Type: AWS::Lambda::Function
    DependsOn:
      - AutoTaggerLogGroup
      - PreflightTrigger  # Custom Resource must return SUCCESS before tagger is provisioned
    Properties:
      FunctionName: !Sub 'map-auto-tagger-${MpeId}'
      Runtime: python3.12
      Handler: index.handler
      Role: !GetAtt AutoTaggerRole.Arn
      Timeout: 60
      MemorySize: 256
      # No ReservedConcurrentExecutions — using SQS buffering instead of direct
      # Lambda invocation means throttling delays processing but never drops events.
      # Messages are retained in SQS for up to 14 days.
      DeadLetterConfig:
        TargetArn: !GetAtt EventDLQ.Arn
      Environment:
        Variables:
          CONFIG_PARAM: !Sub '/auto-map-tagger/${MpeId}/config'
          ALERT_TOPIC_ARN: !Ref AlertTopic
      Code:
        ZipFile: |
          import json, os, boto3, re, time, random
          from datetime import datetime, timezone
          from botocore.exceptions import ClientError

          # Template version pinned at deploy time. Surfaced in CloudWatch Logs on
          # every cold start so ops can trace which version processed an event
          # without reading the CFN stack or SSM parameter.
          TEMPLATE_VERSION = 'v21.0.6'
          print(f'auto-map-tagger {TEMPLATE_VERSION} cold start')

          ssm = boto3.client('ssm')
          ec2 = boto3.client('ec2')
          tagging = boto3.client('resourcegroupstaggingapi')
          _config = None
          _config_ts = 0.0
          # Warm containers live ~15 min; an MPE rotation via SSM wouldn't be
          # seen until the container recycled, silently misattributing credit.
          # 60s TTL bounds the window without adding meaningful SSM load.
          _CONFIG_TTL_SECONDS = 60

          # Peer-tagger detection (§1.108, partial). Runs once at cold-start and
          # logs a WARN + emits a CloudWatch metric (MapAutoTagger /
          # PeerTaggerDetected, dimensions MpeId + PeerMpeId) when another
          # map-auto-tagger-mig* stack is found in this account/region.
          #
          # Why this exists: the configurator's Class-2 scope-intersection
          # preflight (PR #24 + PR #38 Option D) hard-fails on overlap at
          # deploy.sh run-time, but StackSet-engine provisioning (AutoDeployment:
          # True or stack-instance-creation into new OU accounts) bypasses
          # deploy.sh entirely. Phase 16 Test 5 confirmed: linked3 had
          # migbfltest1 (own deploy, passed preflight) + migph2stack01 (arrived
          # via StackSet AutoDeployment, no preflight) — 0/50 resources tagged
          # with migbfltest1. This detector doesn't prevent the contamination;
          # it surfaces it so customers find out from a CloudWatch alarm rather
          # than during a MAP finance audit.
          #
          # Architectural fix (deterministic multi-Lambda routing) is tracked as
          # plan-PR #59 / Wave 28, blocked on design decision with Jin.
          def _detect_peer_taggers():
              try:
                  own_mpe = None
                  cfg_param = os.environ.get('CONFIG_PARAM', '')
                  # CONFIG_PARAM is /auto-map-tagger/<mpe>/config
                  m = re.match(r'^/auto-map-tagger/([^/]+)/config$', cfg_param)
                  if m:
                      own_mpe = m.group(1)
                  if not own_mpe:
                      print('PeerTaggerDetect: could not derive own MpeId from CONFIG_PARAM; skipping')
                      return
                  cfn = boto3.client('cloudformation')
                  paginator = cfn.get_paginator('list_stacks')
                  peer_mpes = []
                  active = ['CREATE_COMPLETE', 'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_COMPLETE']
                  for page in paginator.paginate(StackStatusFilter=active):
                      for s in page.get('StackSummaries', []):
                          name = s.get('StackName', '')
                          if name.startswith('StackSet-map-auto-tagger-mig'):
                              peer_mpe = name[len('StackSet-map-auto-tagger-'):]
                          elif name.startswith('map-auto-tagger-mig'):
                              peer_mpe = name[len('map-auto-tagger-'):]
                          else:
                              continue
                          import re as _re
                          peer_mpe = _re.sub(r'-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', '', peer_mpe)
                          if peer_mpe == own_mpe:
                              continue
                          peer_mpes.append(peer_mpe)
                  if not peer_mpes:
                      return
                  print(
                      f'PeerTaggerDetect: WARN — {len(peer_mpes)} peer tagger '
                      f'stack(s) found in this account/region alongside '
                      f'MpeId={own_mpe}: {peer_mpes}. Concurrent taggers race '
                      f'on the same CloudTrail events; last-writer wins on '
                      f'map-migrated tag value. See docs/ARCHITECTURE.md for '
                      f'deterministic-routing status.'
                  )
                  try:
                      cw = boto3.client('cloudwatch')
                      cw.put_metric_data(
                          Namespace='MapAutoTagger',
                          MetricData=[
                              *[
                                  {
                                      'MetricName': 'PeerTaggerDetected',
                                      'Dimensions': [
                                          {'Name': 'MpeId', 'Value': own_mpe},
                                          {'Name': 'PeerMpeId', 'Value': peer},
                                      ],
                                      'Value': 1,
                                      'Unit': 'Count',
                                  }
                                  for peer in peer_mpes
                              ],
                              {
                                  'MetricName': 'PeerTaggerDetected',
                                  'Dimensions': [
                                      {'Name': 'MpeId', 'Value': own_mpe},
                                  ],
                                  'Value': len(peer_mpes),
                                  'Unit': 'Count',
                              },
                          ],
                      )
                  except Exception as mx:
                      print(f'PeerTaggerDetect: could not emit PeerTaggerDetected metric: {mx}')
              except Exception as e:
                  # Detector must never break cold start. Missing IAM
                  # (cloudformation:ListStacks) or throttle means we skip
                  # today and retry on next cold start.
                  print(f'PeerTaggerDetect: detector skipped ({e})')

          _detect_peer_taggers()

          def get_account_from_arn(arn):
              """Extract account ID from ARN."""
              parts = arn.split(':')
              return parts[4] if len(parts) > 4 else None

          def ci_get(d, key):
              """Case-insensitive dict lookup.

              Handlers read CloudTrail responseElements / requestParameters fields
              that AWS has historically emitted in inconsistent casing — older
              services (Kendra, Redshift, Elastic Beanstalk) emit camelCase or
              lowercase keys while handlers written against the boto3 SDK
              response shape assumed PascalCase. §1.91 / §1.97 / §1.103 document
              three live-confirmed silent-miss handlers.

              Only use on `responseElements` and `requestParameters` — not on
              fixed-shape keys we control. On a response that legitimately has
              both `Id` and `id`, this returns whichever sorts first by key
              (stable across Python 3.7+ dict insertion order, so effectively
              the one AWS emitted first).
              """
              if not isinstance(d, dict) or not d:
                  return None
              if key in d:
                  return d[key]
              lk = key.lower()
              for k, v in d.items():
                  if isinstance(k, str) and k.lower() == lk:
                      return v
              return None

          def get_config():
              global _config, _config_ts
              now = time.time()
              if _config is None or (now - _config_ts) > _CONFIG_TTL_SECONDS:
                  try:
                      resp = ssm.get_parameter(Name=os.environ['CONFIG_PARAM'])
                      cfg = json.loads(resp['Parameter']['Value'])
                  except Exception as e:
                      # One SSM hiccup must not DLQ a burst. Fail closed: return a
                      # safe-default config where mpe_id is None — is_in_scope()
                      # hard-rejects that, so nothing tags until the next TTL
                      # refresh succeeds.
                      print(f'CONFIG_UNREACHABLE: {e}')
                      return {
                          'mpe_id': None,
                          'scope_mode': 'account',
                          'scoped_account_ids': ['ALL'],
                          'scoped_vpc_ids': ['NONE'],
                          'agreement_start_date': None,
                      }
                  # Whitespace strip for customer-edited SSM values. CFN
                  # CommaDelimitedList strips on deploy; SSM-stored config may
                  # carry customer-edit whitespace. Drop empties after strip.
                  if isinstance(cfg.get('scoped_account_ids'), list):
                      cfg['scoped_account_ids'] = [
                          s.strip() for s in cfg['scoped_account_ids']
                          if isinstance(s, str) and s.strip()
                      ]
                  if isinstance(cfg.get('scoped_vpc_ids'), list):
                      cfg['scoped_vpc_ids'] = [
                          s.strip() for s in cfg['scoped_vpc_ids']
                          if isinstance(s, str) and s.strip()
                      ]
                  _config = cfg
                  _config_ts = now
              return _config

          def is_after_agreement(config):
              try:
                  start = datetime.strptime(config.get('agreement_start_date'), '%Y-%m-%d').replace(tzinfo=timezone.utc)
              except (ValueError, TypeError):
                  print(f"Invalid agreement_start_date '{config.get('agreement_start_date')}', skipping event.")
                  return False
              now = datetime.now(timezone.utc)
              if now < start:
                  return False
              end_str = config.get("agreement_end_date")
              if end_str:
                  try:
                      end = datetime.strptime(end_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
                      if now > end:
                          print(f"Event after agreement_end_date {end_str}, skipping.")
                          return False
                  except (ValueError, TypeError):
                      pass
              return True

          def is_in_scope(config, account_id, detail):
              # Fail-closed guards (§1.3 / §1.129 class). Safe-default config
              # from CONFIG_UNREACHABLE has mpe_id=None; if that ever happens
              # elsewhere, the scope decision is also False.
              if not config.get('mpe_id'):
                  return False
              asd = config.get('agreement_start_date')
              try:
                  datetime.strptime(asd, '%Y-%m-%d')
              except (ValueError, TypeError):
                  print(f"CONFIG_INVALID_AGREEMENT_DATE: '{asd}'")
                  return False
              mode = config['scope_mode']
              if mode == 'account':
                  scoped = config.get('scoped_account_ids', [])
                  if isinstance(scoped, str):
                      scoped = [s.strip() for s in scoped.split(',')]
                  if 'ALL' in scoped:
                      return True
                  return account_id in scoped
              if mode == 'vpc':
                  scoped_vpcs = config.get('scoped_vpc_ids', [])
                  if isinstance(scoped_vpcs, str):
                      scoped_vpcs = [s.strip() for s in scoped_vpcs.split(',') if s.strip() and s.strip() != 'NONE']
                  vpc_id = None
                  resp_els = detail.get('responseElements') or {}
                  if 'instanceId' in detail:
                      try:
                          resp = ec2.describe_instances(InstanceIds=[detail['instanceId']])
                          vpc_id = resp['Reservations'][0]['Instances'][0].get('VpcId')
                      except: pass
                  elif resp_els.get('instancesSet', {}).get('items'):
                      vpc_id = resp_els['instancesSet']['items'][0].get('vpcId')
                  elif 'vpcId' in detail:
                      vpc_id = detail['vpcId']
                  elif resp_els.get('vpcId'):
                      vpc_id = resp_els['vpcId']
                  elif (detail.get('requestParameters') or {}).get('vpcId'):
                      vpc_id = detail['requestParameters']['vpcId']
                  elif 'volumeId' in detail:
                      try:
                          resp = ec2.describe_volumes(VolumeIds=[detail['volumeId']])
                          attachments = resp['Volumes'][0].get('Attachments', [])
                          if attachments:
                              inst_id = attachments[0]['InstanceId']
                              resp2 = ec2.describe_instances(InstanceIds=[inst_id])
                              vpc_id = resp2['Reservations'][0]['Instances'][0].get('VpcId')
                      except: pass
                  if vpc_id and vpc_id in scoped_vpcs:
                      return True
                  # §1.3 / U5: when scope_mode == 'vpc' and the resource has no
                  # VPC (S3, DDB, Lambda, SQS, IAM, …), DO NOT fall through to
                  # account-scope. The customer explicitly chose VPC scoping;
                  # silently including non-VPC resources defeats that intent
                  # (e.g. data-residency exclusion). Return False — the resource
                  # is out of scope.
                  return False
              return False

          def extract_arns_multi(detail, account_id, region):
              """Return a list of ARNs for events that create multiple resources.

              Used for events where AWS does NOT emit separate CloudTrail events
              for each child resource. Returns None if the event is not one of
              these; callers should fall back to single-ARN extract_arn.

              RunInstances: the CloudTrail responseElements contains instance IDs
              and the primary ENI, but the block-device-mapping is usually empty
              — AWS populates the attached EBS volume IDs AFTER the instance
              reaches pending/running, not in the RunInstances API response.
              We therefore describe_instances to fetch the volume IDs.
              """
              event_name = detail.get('eventName', '')
              if event_name != 'RunInstances':
                  return None
              resp = detail.get('responseElements') or {}
              items = resp.get('instancesSet', {}).get('items') or []
              if not items:
                  return None
              arns = []
              # Gather instance IDs + any attached ENIs present in the event.
              # Volumes come from describe_instances below.
              inst_ids = []
              for it in items:
                  inst_id = it.get('instanceId')
                  if inst_id:
                      arns.append(f"arn:aws:ec2:{region}:{account_id}:instance/{inst_id}")
                      inst_ids.append(inst_id)
                  for ni in (it.get('networkInterfaceSet', {}) or {}).get('items') or []:
                      ni_id = ni.get('networkInterfaceId')
                      if ni_id:
                          arns.append(f"arn:aws:ec2:{region}:{account_id}:network-interface/{ni_id}")

              # Look up attached volume IDs via describe_instances. CloudTrail
              # fires RunInstances milliseconds after the API call returns;
              # volumes may not yet be attached. Poll briefly (up to 30s) for
              # BlockDeviceMappings to populate. Every account has its own
              # Lambda (StackSet architecture), so events are always processed
              # in-account — no cross-account client needed.
              if inst_ids:
                  try:
                      ec2_lookup = boto3.client('ec2', region_name=region)
                      if ec2_lookup:
                          deadline = time.time() + 30
                          volume_ids = set()
                          while time.time() < deadline:
                              try:
                                  desc = ec2_lookup.describe_instances(InstanceIds=inst_ids)
                                  all_have_vols = True
                                  volume_ids.clear()
                                  for resv in desc.get('Reservations', []):
                                      for inst in resv.get('Instances', []):
                                          bdm = inst.get('BlockDeviceMappings', []) or []
                                          if not bdm:
                                              all_have_vols = False
                                          for m in bdm:
                                              vid = (m.get('Ebs') or {}).get('VolumeId')
                                              if vid:
                                                  volume_ids.add(vid)
                                  if all_have_vols and volume_ids:
                                      break
                              except ClientError as e:
                                  # Malformed instance IDs never resolve — no
                                  # point burning the 30s budget. NotFound is
                                  # different: the instance may still be
                                  # materializing, so keep polling.
                                  if e.response.get('Error', {}).get('Code') == 'InvalidInstanceID.Malformed':
                                      raise
                                  print(f'describe_instances poll: {e}')
                              except Exception as e:
                                  print(f'describe_instances poll: {e}')
                              time.sleep(3)
                          for vid in volume_ids:
                              arns.append(f"arn:aws:ec2:{region}:{account_id}:volume/{vid}")
                  except ClientError as e:
                      # Let malformed-ID propagate so _classify_error routes
                      # it to permanent_actionable instead of a silent log.
                      if e.response.get('Error', {}).get('Code') == 'InvalidInstanceID.Malformed':
                          raise
                      print(f'Could not resolve RunInstances attached volumes: {e}')
                  except Exception as e:
                      print(f'Could not resolve RunInstances attached volumes: {e}')

              return arns or None


          def extract_arn(detail, account_id, region):
              event_name = detail.get('eventName', '')
              event_source = detail.get('eventSource', '')
              resp = detail.get('responseElements') or {}
              req = detail.get('requestParameters') or {}
              resources = detail.get('resources', [])

              # Handle early exceptions BEFORE universal pattern scan
              # (prevents universal scan from picking up wrong ARNs)
              # Bedrock Agent (nested: agent.agentArn)
              if event_name == 'CreateAgent' and event_source == 'bedrock.amazonaws.com':
                  agent_arn = resp.get('agent', {}).get('agentArn')
                  if agent_arn:
                      return agent_arn

              # Bedrock Agent Action Group (no ARN in response — construct from IDs)
              elif event_name == 'CreateAgentActionGroup' and event_source == 'bedrock.amazonaws.com':
                  ag = resp.get('agentActionGroup', {})
                  agent_id = ag.get('agentId')
                  agent_ver = ag.get('agentVersion', 'DRAFT')
                  ag_id = ag.get('actionGroupId')
                  if agent_id and ag_id:
                      return f"arn:aws:bedrock:{region}:{account_id}:agent/{agent_id}/agentversion/{agent_ver}/actiongroup/{ag_id}"

              # Bedrock Agent Alias (nested: agentAlias.agentAliasArn)
              elif event_name == 'CreateAgentAlias' and event_source == 'bedrock.amazonaws.com':
                  alias_arn = resp.get('agentAlias', {}).get('agentAliasArn')
                  if alias_arn:
                      return alias_arn

              # Bedrock Knowledge Base Data Source (no ARN in response — construct)
              elif event_name == 'CreateDataSource' and event_source == 'bedrock.amazonaws.com':
                  ds = resp.get('dataSource', {})
                  kb_id = ds.get('knowledgeBaseId') or req.get('knowledgeBaseId')
                  ds_id = ds.get('dataSourceId')
                  if kb_id and ds_id:
                      return f"arn:aws:bedrock:{region}:{account_id}:knowledge-base/{kb_id}/data-source/{ds_id}"

              # (Lambda Layer handling removed — layers are not taggable per AWS
              # Lambda TagResource API regex; PublishLayerVersion is in IGNORE_EVENTS
              # at the handler level.)

              # ECS Service — early exit to prevent clusterArn being picked up by universal scan
              if event_name == 'CreateService' and event_source == 'ecs.amazonaws.com':
                  svc_arn = resp.get('service', {}).get('serviceArn')
                  if svc_arn:
                      return svc_arn

              elif event_name == 'CreateVpnConnection':
                  vpn = resp.get('vpnConnection', {})
                  vpn_id = vpn.get('vpnConnectionId')
                  if vpn_id:
                      return f"arn:aws:ec2:{region}:{account_id}:vpn-connection/{vpn_id}"

              elif event_name == 'CreatePlacementGroup':
                  pg = resp.get('placementGroup', {})
                  if not pg and 'CreatePlacementGroupResponse' in resp:
                      pg = resp['CreatePlacementGroupResponse'].get('placementGroup', {})
                  pg_id = pg.get('groupId')
                  if pg_id:
                      return f"arn:aws:ec2:{region}:{account_id}:placement-group/{pg_id}"
                  pg_arn = pg.get('groupArn')
                  if pg_arn:
                      return pg_arn

              # Universal ARN patterns
              arn_patterns = [
                  'arn', 'Arn', 'ARN', 'resourceArn', 'ResourceArn', 'clusterArn', 'functionArn',
                  'topicArn', 'tableArn', 'streamArn', 'stateMachineArn', 'pipelineArn',
                  'backupVaultArn', 'dBInstanceArn', 'dBClusterArn', 'queueArn', 'loadBalancerArn',
                  'eventBusArn', 'aRN', 'notebookInstanceArn', 'documentClassifierArn', 'applicationArn',
                  'ApplicationARN', 'backupPlanArn', 'locationArn', 'computeEnvironmentArn', 'appArn',
                  'groupARN', 'groupArn', 'activityArn', 'keyArn', 'serviceArn',
                  'policyArn', 'replicationGroupArn',
                  'modelArn', 'endpointArn', 'endpointConfigArn', 'trainingJobArn',
                  'connectionArn', 'workgroupArn', 'namespaceArn',
                  'clusterArn', 'nodeGroupArn', 'containerArn',
                  'registryArn', 'imageArn', 'repositoryArn',
                  'aliasArn', 'pipeArn', 'channelArn',
                  'acceleratorArn', 'jobArn', 'firewallArn',
                  'firewallPolicyArn', 'addonArn', 'userPoolArn',
                  'guardrailArn', 'agentArn', 'knowledgeBaseArn',
                  'flowArn', 'promptArn', 'promptVersionArn',
                  # Bedrock sub-resources
                  'agentAliasArn', 'dataSourceArn', 'provisionedModelArn',
                  # Amazon MQ
                  'BrokerArn', 'brokerArn',
                  # AWS HealthLake
                  'DatastoreArn', 'datastoreArn',
                  # Amazon WorkSpaces Web
                  'portalArn',
                  # SageMaker Pipeline / Studio Domain / Feature Store
                  'PipelineArn', 'DomainArn', 'FeatureGroupArn', 'featureGroupArn',
                  # Service Catalog (nested in productViewDetail)
                  'productARN',
                  # IoT SiteWise
                  'assetModelArn', 'assetArn', 'portalArn', 'gatewayArn',
                  # IoT Analytics / Events
                  'datasetArn', 'inputArn', 'detectorModelArn',
                  # GameLift
                  'buildArn', 'scriptArn', 'fleetArn',
                  # Lookout services
                  'ProjectArn', 'DatasetArn', 'AnomalyDetectorArn',
                  # Shared ARN patterns (SageMaker, OpenSearch, etc.)
                  'domainArn', 'instanceArn', 'applicationArn',
                  # Amazon Omics
                  'sequenceStoreArn', 'annotationStoreArn',
                  # Elastic Beanstalk
                  'ApplicationArn', 'EnvironmentArn',
                  # MediaLive / MediaPackage / MediaStore
                  'ChannelArn', 'OriginEndpointArn',
              ]

              # Try to find ARN in response
              for pattern in arn_patterns:
                  if pattern in resp and resp[pattern]:
                      return resp[pattern]
                  for key, val in resp.items():
                      if isinstance(val, dict) and pattern in val and val[pattern]:
                          return val[pattern]

              # Suffix-match fallback (§1.31/§1.35/§1.56/§1.57/§1.61/§1.63 et al.).
              # The hand-curated allowlist above misses ~46 ARN field names across
              # new/less-common AWS services — every miss is a silent tag loss. This
              # fallback catches any key ending in Arn/ARN/arn whose value is a
              # string beginning with 'arn:'. Guards against false positives:
              #   - val must start with 'arn:' (rejects plain IDs, names, dates)
              #   - ARN's service segment must match the event source (rejects
              #     related-resource ARNs like roleArn on a CreateFunction event
              #     where functionArn — caught above — is the real subject)
              # Ordering: Python dict iteration is insertion order (3.7+), and
              # CloudTrail typically puts the subject-resource ARN earlier than
              # related ARNs, so first match wins. Nested scan is 1-level only,
              # matching the allowlist scan's depth.
              event_svc_prefix = event_source.split('.')[0].replace('aws.', '').lower()
              def _maybe_arn(k, v):
                  if not isinstance(v, str) or not v.startswith('arn:'):
                      return None
                  if not (k.endswith('Arn') or k.endswith('ARN') or k.endswith('arn')):
                      return None
                  # arn:aws:<service>:<region>:<acct>:<resource>
                  parts = v.split(':', 5)
                  if len(parts) < 6:
                      return None
                  arn_svc = parts[2].lower()
                  # Allow common cross-service subject-ARN patterns (e.g. ec2
                  # events that produce elasticloadbalancing ARNs are rare,
                  # but event-source-to-ARN-service isn't a strict 1:1 map).
                  # Accept either an exact prefix match or a well-known alias.
                  if arn_svc == event_svc_prefix:
                      return v
                  aliases = {
                      'elasticloadbalancing': ('elbv2', 'elb', 'elasticloadbalancing'),
                      'route53': ('route53', 'route53domains', 'route53resolver'),
                      'apigateway': ('apigateway',),
                      'bedrock': ('bedrock', 'bedrock-agent', 'bedrock-runtime'),
                      'kinesis': ('kinesis', 'kinesisvideo', 'kinesisanalytics', 'firehose'),
                      'rds': ('rds', 'neptune', 'docdb'),
                      'sagemaker': ('sagemaker',),
                      'iot': ('iot', 'iotanalytics', 'iotevents', 'iotsitewise'),
                  }
                  if arn_svc in aliases.get(event_svc_prefix, ()):
                      return v
                  return None
              for key, val in resp.items():
                  m = _maybe_arn(key, val)
                  if m:
                      return m
                  if isinstance(val, dict):
                      for nk, nv in val.items():
                          m = _maybe_arn(nk, nv)
                          if m:
                              return m

              # Glue Table - must be before resources array check to avoid returning database ARN
              # (Glue CreateTable CloudTrail resources array contains database ARN first)
              if event_name == 'CreateTable' and event_source == 'glue.amazonaws.com':
                  db_name = req.get('databaseName')
                  tbl_name = req.get('tableInput', {}).get('name')
                  if db_name and tbl_name:
                      return f"arn:aws:glue:{region}:{account_id}:table/{db_name}/{tbl_name}"

              # Check resources array (used by some services like Keyspaces)
              # Skip catalog ARNs for Glue (use database ARN instead)
              if resources and len(resources) > 0:
                  for resource in resources:
                      arn = resource.get('ARN')
                      # Skip Glue catalog ARNs - we want the database/table ARN
                      if arn and ':catalog' not in arn:
                          return arn

              # Construct ARNs for specific services
              service = event_source.split('.')[0].replace('aws.', '')
              
              if event_name == 'RunInstances':
                  # RunInstances is handled via extract_arns_multi (called from
                  # the handler) because it emits multiple resources: all instance
                  # IDs, all attached EBS volume IDs, and all attached ENI IDs.
                  # AWS does NOT emit separate CreateVolume / CreateNetworkInterface
                  # CloudTrail events for resources born inline with RunInstances,
                  # so we must extract them from the RunInstances event itself.
                  # Returning the first instance ARN here preserves backward
                  # compatibility if this single-ARN path is hit (should not be
                  # for RunInstances — the handler calls extract_arns_multi first).
                  items = resp.get('instancesSet', {}).get('items', [])
                  if items:
                      return f"arn:aws:ec2:{region}:{account_id}:instance/{items[0].get('instanceId')}"
              elif event_name == 'CreateVolume':
                  vol_id = resp.get('volumeId')
                  return f"arn:aws:ec2:{region}:{account_id}:volume/{vol_id}" if vol_id else None
              elif event_name == 'CreateBucket':
                  bucket = req.get('bucketName')
                  return f"arn:aws:s3:::{bucket}" if bucket else None
              elif event_name == 'CreateStream' and event_source == 'kinesis.amazonaws.com':
                  # Exact match — `'kinesis' in service` also matched
                  # kinesisvideo.amazonaws.com and produced a broken
                  # arn:aws:kinesis:... ARN for KVS streams. KVS has its
                  # own handler (look further down for kinesisvideo).
                  name = req.get('streamName')
                  return f"arn:aws:kinesis:{region}:{account_id}:stream/{name}" if name else None
              elif event_name == 'CreateDeliveryStream':
                  name = req.get('deliveryStreamName')
                  return f"arn:aws:firehose:{region}:{account_id}:deliverystream/{name}" if name else None
              elif event_name == 'CreateAutoScalingGroup':
                  name = req.get('autoScalingGroupName')
                  return f"arn:aws:autoscaling:{region}:{account_id}:autoScalingGroup:*:autoScalingGroupName/{name}" if name else None
              elif event_name == 'CreateLogGroup':
                  name = req.get('logGroupName')
                  return f"arn:aws:logs:{region}:{account_id}:log-group:{name}" if name else None
              elif event_name in ['CreateRestApi', 'CreateApi']:
                  api_id = resp.get('id') or resp.get('apiId')
                  api_type = 'restapis' if event_name == 'CreateRestApi' else 'apis'
                  return f"arn:aws:apigateway:{region}::/{api_type}/{api_id}" if api_id else None
              elif event_name == 'CreateQueue' and event_source == 'sqs.amazonaws.com':
                  # SQS returns URL, need to construct ARN
                  queue_url = resp.get('queueUrl')
                  if queue_url:
                      queue_name = queue_url.split('/')[-1]
                      return f"arn:aws:sqs:{region}:{account_id}:{queue_name}"

              # EC2 Resources (use resource IDs to construct ARNs)
              elif event_name == 'CreateVpc':
                  vpc = resp.get('vpc', {})
                  vpc_id = vpc.get('vpcId')
                  if vpc_id:
                      return f"arn:aws:ec2:{region}:{account_id}:vpc/{vpc_id}"
              elif event_name == 'CreateSecurityGroup':
                  group_id = resp.get('groupId')
                  if group_id:
                      return f"arn:aws:ec2:{region}:{account_id}:security-group/{group_id}"
              elif event_name == 'CreateNetworkInterface':
                  eni = resp.get('networkInterface', {})
                  eni_id = eni.get('networkInterfaceId')
                  if eni_id:
                      return f"arn:aws:ec2:{region}:{account_id}:network-interface/{eni_id}"
              elif event_name == 'CreateSubnet':
                  subnet = resp.get('subnet', {})
                  subnet_id = subnet.get('subnetId')
                  if subnet_id:
                      return f"arn:aws:ec2:{region}:{account_id}:subnet/{subnet_id}"
              elif event_name == 'CreateInternetGateway':
                  igw = resp.get('internetGateway', {})
                  igw_id = igw.get('internetGatewayId')
                  if igw_id:
                      return f"arn:aws:ec2:{region}:{account_id}:internet-gateway/{igw_id}"
              elif event_name == 'AllocateAddress':
                  alloc_id = resp.get('allocationId')
                  if alloc_id:
                      return f"arn:aws:ec2:{region}:{account_id}:elastic-ip/{alloc_id}"

              # Egress-only Internet Gateway
              elif event_name == 'CreateEgressOnlyInternetGateway':
                  eigw = resp.get('egressOnlyInternetGateway', {})
                  if not eigw and 'CreateEgressOnlyInternetGatewayResponse' in resp:
                      eigw = resp['CreateEgressOnlyInternetGatewayResponse'].get('egressOnlyInternetGateway', {})
                  eigw_id = eigw.get('egressOnlyInternetGatewayId')
                  if eigw_id:
                      return f"arn:aws:ec2:{region}:{account_id}:egress-only-internet-gateway/{eigw_id}"

              # Customer Gateway
              elif event_name == 'CreateCustomerGateway':
                  cgw = resp.get('customerGateway', {})
                  if not cgw and 'CreateCustomerGatewayResponse' in resp:
                      cgw = resp['CreateCustomerGatewayResponse'].get('customerGateway', {})
                  cgw_id = cgw.get('customerGatewayId')
                  if cgw_id:
                      return f"arn:aws:ec2:{region}:{account_id}:customer-gateway/{cgw_id}"

              # Network ACL
              elif event_name == 'CreateNetworkAcl':
                  acl = resp.get('networkAcl', {})
                  if not acl and 'CreateNetworkAclResponse' in resp:
                      acl = resp['CreateNetworkAclResponse'].get('networkAcl', {})
                  acl_id = acl.get('networkAclId')
                  if acl_id:
                      return f"arn:aws:ec2:{region}:{account_id}:network-acl/{acl_id}"

              # NAT Gateway (§1.50 — CloudTrail EC2 responses wrap in
              # <EventName>Response; match the unwrap pattern used by
              # CreateNetworkAcl / CreateLaunchTemplate / CreateTransitGateway /
              # CreateVpnGateway / CreateRouteTable above.)
              elif event_name == 'CreateNatGateway':
                  nat = resp.get('natGateway', {})
                  if not nat and 'CreateNatGatewayResponse' in resp:
                      nat = resp['CreateNatGatewayResponse'].get('natGateway', {})
                  nat_id = nat.get('natGatewayId')
                  if nat_id:
                      return f"arn:aws:ec2:{region}:{account_id}:natgateway/{nat_id}"

              # Simple ARN Extractions
              elif event_name == 'CreateRepository' and event_source == 'ecr.amazonaws.com':
                  repo = resp.get('repository', {})
                  arn = repo.get('repositoryArn')
                  return arn
              elif event_name == 'CreateFileSystem' and event_source == 'elasticfilesystem.amazonaws.com':
                  return resp.get('fileSystemArn')
              elif event_name == 'CreateFileSystem' and event_source == 'fsx.amazonaws.com':
                  fs = resp.get('fileSystem', {})
                  return fs.get('resourceARN')
              elif event_name == 'PutRule':
                  rule_arn = resp.get('ruleArn')
                  return rule_arn
              elif event_name == 'CreateStack':
                  stack_id = resp.get('stackId')
                  return stack_id
              elif event_name == 'CreateStackSet':
                  stackset_id = resp.get('stackSetId')
                  if stackset_id:
                      # StackSet ID format: name:uuid
                      return f"arn:aws:cloudformation:{region}:{account_id}:stackset/{stackset_id}"
              elif event_name == 'CreateLaunchTemplate':
                  # CloudTrail wraps EC2 responses in <EventName>Response
                  template = resp.get('launchTemplate', {})
                  if not template and 'CreateLaunchTemplateResponse' in resp:
                      template = resp['CreateLaunchTemplateResponse'].get('launchTemplate', {})
                  template_id = template.get('launchTemplateId')
                  if template_id:
                      return f"arn:aws:ec2:{region}:{account_id}:launch-template/{template_id}"

              # CloudWatch Alarms
              elif event_name == 'PutMetricAlarm':
                  alarm_name = req.get('alarmName')
                  if alarm_name:
                      return f"arn:aws:cloudwatch:{region}:{account_id}:alarm:{alarm_name}"

              # SSM Parameters — name may or may not have a leading "/".
              # Valid ARN is always "parameter/<name>"; normalize to exactly one "/".
              elif event_name == 'PutParameter':
                  param_name = req.get('name')
                  if param_name:
                      return f"arn:aws:ssm:{region}:{account_id}:parameter/{param_name.lstrip('/')}"

              # ACM Certificates
              elif event_name == 'RequestCertificate':
                  cert_arn = resp.get('certificateArn')
                  return cert_arn

              # Transit Gateway (EC2 response wrapper)
              elif event_name == 'CreateTransitGateway':
                  tgw = resp.get('transitGateway', {})
                  if not tgw and 'CreateTransitGatewayResponse' in resp:
                      tgw = resp['CreateTransitGatewayResponse'].get('transitGateway', {})
                  tgw_id = tgw.get('transitGatewayId')
                  if tgw_id:
                      return f"arn:aws:ec2:{region}:{account_id}:transit-gateway/{tgw_id}"

              # VPN Gateway (EC2 response wrapper)
              elif event_name == 'CreateVpnGateway':
                  vpn = resp.get('vpnGateway', {})
                  if not vpn and 'CreateVpnGatewayResponse' in resp:
                      vpn = resp['CreateVpnGatewayResponse'].get('vpnGateway', {})
                  vpn_id = vpn.get('vpnGatewayId')
                  if vpn_id:
                      return f"arn:aws:ec2:{region}:{account_id}:vpn-gateway/{vpn_id}"

              # Route Table
              elif event_name == 'CreateRouteTable':
                  rt = resp.get('routeTable', {})
                  if not rt and 'CreateRouteTableResponse' in resp:
                      rt = resp['CreateRouteTableResponse'].get('routeTable', {})
                  rt_id = rt.get('routeTableId')
                  if rt_id:
                      return f"arn:aws:ec2:{region}:{account_id}:route-table/{rt_id}"

              # Glue Job
              elif event_name == 'CreateJob':
                  job_name = req.get('name')
                  if job_name:
                      return f"arn:aws:glue:{region}:{account_id}:job/{job_name}"

              # Load Balancer — ELBv1 Classic AND ELBv2 (ALB/NLB/GWLB) share
              # eventSource='elasticloadbalancing.amazonaws.com'. They're
              # disambiguated only by response shape:
              #   ELBv2: responseElements.loadBalancers[].loadBalancerArn
              #   ELBv1 Classic: responseElements.dNSName (no ARN); name in request
              elif event_name == 'CreateLoadBalancer' and event_source == 'elasticloadbalancing.amazonaws.com':
                  lbs = resp.get('loadBalancers', [])
                  if lbs and len(lbs) > 0:
                      # ELBv2 path
                      return lbs[0].get('loadBalancerArn')
                  # ELBv1 Classic path — construct ARN from request name
                  lb_name = req.get('loadBalancerName')
                  if lb_name:
                      return f"arn:aws:elasticloadbalancing:{region}:{account_id}:loadbalancer/{lb_name}"

              # Database Clusters
              elif event_name == 'CreateDBCluster':
                  cluster = resp.get('dBCluster', {})
                  return cluster.get('dBClusterArn')
              elif event_name == 'CreateCluster' and event_source == 'redshift.amazonaws.com':
                  # §1.91 — CloudTrail casing for Redshift response fields has
                  # varied across eventVersion revisions (`clusterIdentifier` vs
                  # `ClusterIdentifier`). ci_get handles both.
                  cluster_id = ci_get(resp, 'clusterIdentifier')
                  if cluster_id:
                      return f"arn:aws:redshift:{region}:{account_id}:cluster:{cluster_id}"
              elif event_name == 'CreateCluster' and event_source == 'memorydb.amazonaws.com':
                  cluster = resp.get('cluster', {})
                  return cluster.get('aRN')  # Note: capital ARN

              # CloudHSM v2 Cluster — responseElements has no ARN, only cluster.clusterId.
              # §1.99b: CloudTrail eventSource for CloudHSM v2 is 'cloudhsm.amazonaws.com'
              # (NOT 'cloudhsmv2.amazonaws.com'). The boto3 service name IS 'cloudhsmv2';
              # only the CloudTrail eventSource differs. Live-confirmed 2026-04-27.
              elif event_name == 'CreateCluster' and event_source == 'cloudhsm.amazonaws.com':
                  cluster_id = (resp.get('cluster') or {}).get('clusterId')
                  if cluster_id:
                      return f"arn:aws:cloudhsm:{region}:{account_id}:cluster/{cluster_id}"

              # CloudHSM v2 HSM — tagging routes to parent cluster (idempotent with CreateCluster handler)
              elif event_name == 'CreateHsm' and event_source == 'cloudhsm.amazonaws.com':
                  cluster_id = (resp.get('hsm') or {}).get('clusterId') or req.get('clusterId')
                  if cluster_id:
                      return f"arn:aws:cloudhsm:{region}:{account_id}:cluster/{cluster_id}"

              # AWS Directory Service (Simple AD + Managed Microsoft AD + AD Connector)
              elif event_name in ('CreateDirectory', 'CreateMicrosoftAD', 'ConnectDirectory') and event_source == 'ds.amazonaws.com':
                  dir_id = resp.get('directoryId')
                  if dir_id:
                      return f"arn:aws:ds:{region}:{account_id}:directory/{dir_id}"

              # Amazon Keyspaces — responseElements is empty on CreateKeyspace;
              # name is in requestParameters.keyspaceName. Trailing slash required
              # per keyspaces ARN format.
              elif event_name == 'CreateKeyspace' and event_source == 'cassandra.amazonaws.com':
                  ks_name = req.get('keyspaceName')
                  if ks_name:
                      return f"arn:aws:cassandra:{region}:{account_id}:/keyspace/{ks_name}/"

              # SageMaker
              elif event_name == 'CreateNotebookInstance':
                  return resp.get('notebookInstanceArn')

              # Comprehend
              elif event_name == 'CreateDocumentClassifier':
                  return resp.get('documentClassifierArn')

              # AppSync
              elif event_name == 'CreateGraphqlApi' and event_source == 'appsync.amazonaws.com':
                  api = resp.get('graphqlApi', {})
                  return api.get('arn')

              # IoT
              elif event_name == 'CreateThing':
                  return resp.get('thingArn')

              # Service Catalog
              elif event_name == 'CreatePortfolio':
                  portfolio = resp.get('portfolioDetail', {})
                  return portfolio.get('aRN')  # Note: capital ARN

              # Transfer Family Server
              elif event_name == 'CreateServer' and event_source == 'transfer.amazonaws.com':
                  server_id = resp.get('serverId')
                  if server_id:
                      return f"arn:aws:transfer:{region}:{account_id}:server/{server_id}"

              # RDS Proxy
              elif event_name == 'CreateDBProxy':
                  proxy = resp.get('dBProxy', {})
                  return proxy.get('dBProxyArn')

              # Global Accelerator
              elif event_name == 'CreateAccelerator':
                  accelerator = resp.get('accelerator', {})
                  return accelerator.get('acceleratorArn')

              # CodeBuild Project
              elif event_name == 'CreateProject' and event_source == 'codebuild.amazonaws.com':
                  project = resp.get('project', {})
                  return project.get('arn')

              # CodePipeline Pipeline - ARN not in response, construct from name
              elif event_name == 'CreatePipeline' and event_source == 'codepipeline.amazonaws.com':
                  pipeline = resp.get('pipeline', {})
                  pipeline_name = pipeline.get('name')
                  if pipeline_name:
                      return f"arn:aws:codepipeline:{region}:{account_id}:{pipeline_name}"

              # Target Group
              elif event_name == 'CreateTargetGroup':
                  tgs = resp.get('targetGroups', [])
                  if tgs and len(tgs) > 0:
                      return tgs[0].get('targetGroupArn')

              # Glue Database
              elif event_name == 'CreateDatabase' and event_source == 'glue.amazonaws.com':
                  db_name = req.get('databaseInput', {}).get('name')
                  if db_name:
                      return f"arn:aws:glue:{region}:{account_id}:database/{db_name}"

              # Glue Crawler
              elif event_name == 'CreateCrawler':
                  crawler_name = req.get('name')
                  if crawler_name:
                      return f"arn:aws:glue:{region}:{account_id}:crawler/{crawler_name}"

              # Glue Workflow
              elif event_name == 'CreateWorkflow':
                  workflow_name = resp.get('name')
                  if workflow_name:
                      return f"arn:aws:glue:{region}:{account_id}:workflow/{workflow_name}"

              # Backup Plan
              elif event_name == 'CreateBackupPlan':
                  plan_id = resp.get('backupPlanId')
                  if plan_id:
                      return f"arn:aws:backup:{region}:{account_id}:backup-plan:{plan_id}"

              # CloudWatch Dashboard — §1.83. ARN has EMPTY region per AWS IAM
              # Service Authorization Reference (dashboards are global within an
              # account). Prior handler emitted `arn:aws:cloudwatch:<region>:…` which
              # RGTA rejected on every PutDashboard event.
              elif event_name == 'PutDashboard':
                  dashboard_name = ci_get(req, 'dashboardName')
                  if dashboard_name:
                      return f"arn:aws:cloudwatch::{account_id}:dashboard/{dashboard_name}"

              # EMR Cluster (RunJobFlow)
              elif event_name == 'RunJobFlow':
                  cluster_id = resp.get('jobFlowId')
                  if cluster_id:
                      return f"arn:aws:elasticmapreduce:{region}:{account_id}:cluster/{cluster_id}"

              # EKS Cluster
              elif event_name == 'CreateCluster' and event_source == 'eks.amazonaws.com':
                  cluster = resp.get('cluster', {})
                  cluster_arn = cluster.get('arn')
                  if cluster_arn:
                      return cluster_arn

              # EKS Nodegroup
              elif event_name == 'CreateNodegroup':
                  nodegroup = resp.get('nodegroup', {})
                  return nodegroup.get('nodegroupArn')

              # ElastiCache - needs retry, but add ARN extraction
              elif event_name == 'CreateCacheCluster':
                  cache = resp.get('cacheCluster', {})
                  return cache.get('aRN')

              # Route53 Hosted Zone
              elif event_name == 'CreateHostedZone':
                  zone_id = resp.get('hostedZone', {}).get('id')
                  if zone_id:
                      # ID format: /hostedzone/Z0862949349EG6JSOWHXK
                      zone_id = zone_id.split('/')[-1]
                      return f"arn:aws:route53:::hostedzone/{zone_id}"

              # VPC Peering Connection
              elif event_name == 'CreateVpcPeeringConnection':
                  vpc_pcx = resp.get('vpcPeeringConnection', {})
                  if not vpc_pcx and 'CreateVpcPeeringConnectionResponse' in resp:
                      vpc_pcx = resp['CreateVpcPeeringConnectionResponse'].get('vpcPeeringConnection', {})
                  pcx_id = vpc_pcx.get('vpcPeeringConnectionId')
                  if pcx_id:
                      return f"arn:aws:ec2:{region}:{account_id}:vpc-peering-connection/{pcx_id}"

              # EBS Snapshot — scope to EC2 source to prevent catching CreateSnapshot
              # from ElastiCache / MemoryDB / FSx (OpenZFS) / Redshift Serverless,
              # which use the same event name but different response shapes.
              # Non-EC2 CreateSnapshot is handled by the universal scan above
              # (ElastiCache/MemoryDB via `snapshot.ARN` key) or by explicit
              # handlers further down (FSx, Redshift Serverless).
              elif event_name == 'CreateSnapshot' and event_source == 'ec2.amazonaws.com':
                  snapshot_id = resp.get('snapshotId')
                  if snapshot_id:
                      return f"arn:aws:ec2:{region}:{account_id}:snapshot/{snapshot_id}"

              # VPC Flow Logs
              elif event_name == 'CreateFlowLogs':
                  # EC2 response wrapper: CreateFlowLogsResponse.flowLogIdSet.item
                  fl_id = None
                  if 'CreateFlowLogsResponse' in resp:
                      nested = resp['CreateFlowLogsResponse']
                      fl_id_set = nested.get('flowLogIdSet', {})
                      fl_id = fl_id_set.get('item') if isinstance(fl_id_set, dict) else None
                      if isinstance(fl_id, list):
                          fl_id = fl_id[0]
                  if not fl_id:
                      flow_ids = resp.get('flowLogIds', [])
                      fl_id = flow_ids[0] if flow_ids else None
                  if fl_id:
                      return f"arn:aws:ec2:{region}:{account_id}:vpc-flow-log/{fl_id}"

              # SSM Document
              elif event_name == 'CreateDocument':
                  doc_desc = resp.get('documentDescription', {})
                  doc_name = doc_desc.get('name')
                  if doc_name:
                      return f"arn:aws:ssm:{region}:{account_id}:document/{doc_name}"

              # DHCP Options
              elif event_name == 'CreateDhcpOptions':
                  dhcp = resp.get('dhcpOptions', {})
                  if not dhcp and 'CreateDhcpOptionsResponse' in resp:
                      dhcp = resp['CreateDhcpOptionsResponse'].get('dhcpOptions', {})
                  dhcp_id = dhcp.get('dhcpOptionsId')
                  if dhcp_id:
                      return f"arn:aws:ec2:{region}:{account_id}:dhcp-options/{dhcp_id}"

              # AMI (CreateImage)
              elif event_name == 'CreateImage':
                  image_id = resp.get('imageId')
                  if not image_id:
                      image_id = resp.get('ImageId')
                  if image_id:
                      return f"arn:aws:ec2:{region}:{account_id}:image/{image_id}"

              # DMS Replication Instance
              elif event_name == 'CreateReplicationInstance':
                  rep = resp.get('replicationInstance', {})
                  rep_arn = rep.get('replicationInstanceArn')
                  if rep_arn:
                      return rep_arn

              # DataSync Task
              elif event_name == 'CreateTask':
                  task_arn = resp.get('taskArn')
                  if task_arn:
                      return task_arn

              # Glue DataBrew Dataset
              elif event_name == 'CreateDataset' and event_source == 'databrew.amazonaws.com':
                  name = resp.get('Name') or req.get('Name')
                  if name:
                      return f"arn:aws:databrew:{region}:{account_id}:dataset/{name}"

              # Glue DataBrew Recipe
              elif event_name == 'CreateRecipe' and event_source == 'databrew.amazonaws.com':
                  name = resp.get('Name') or req.get('Name')
                  if name:
                      return f"arn:aws:databrew:{region}:{account_id}:recipe/{name}"

              # Amazon Timestream Database
              elif event_name == 'CreateDatabase' and event_source == 'timestream.amazonaws.com':
                  db = resp.get('Database', {})
                  db_arn = db.get('Arn')
                  if db_arn:
                      return db_arn

              # Amazon Timestream Table
              elif event_name == 'CreateTable' and event_source == 'timestream.amazonaws.com':
                  tbl = resp.get('Table', {})
                  tbl_arn = tbl.get('Arn')
                  if tbl_arn:
                      return tbl_arn

              # EMR Serverless Application (applicationId not ARN in response)
              elif event_name == 'CreateApplication' and event_source == 'emr-serverless.amazonaws.com':
                  app_id = resp.get('applicationId')
                  if app_id:
                      return f"arn:aws:emr-serverless:{region}:{account_id}:/applications/{app_id}"

              # Amazon Kendra Index (Id not ARN in response)
              elif event_name == 'CreateIndex' and event_source == 'kendra.amazonaws.com':
                  # §1.97 — CloudTrail emits lowercase `id`, boto3 SDK shape is
                  # `Id`. Prior handler used resp.get('Id') → silent miss on every
                  # Kendra Index creation. ci_get catches either casing.
                  idx_id = ci_get(resp, 'Id')
                  if idx_id:
                      return f"arn:aws:kendra:{region}:{account_id}:index/{idx_id}"

              # Amazon Kendra DataSource
              elif event_name == 'CreateDataSource' and event_source == 'kendra.amazonaws.com':
                  ds_id = ci_get(resp, 'Id')
                  idx_id = ci_get(req, 'IndexId')
                  if ds_id and idx_id:
                      return f"arn:aws:kendra:{region}:{account_id}:index/{idx_id}/data-source/{ds_id}"

              # AWS Elastic Beanstalk Application
              elif event_name == 'CreateApplication' and event_source == 'elasticbeanstalk.amazonaws.com':
                  # §1.103 — CloudTrail nests at `application.applicationName`
                  # (lowercase leading) while boto3 SDK shape is
                  # `Application.ApplicationName`. ci_get the outer object and
                  # the inner key so either casing works.
                  app = ci_get(resp, 'Application') or {}
                  app_arn = ci_get(app, 'ApplicationArn')
                  if app_arn:
                      return app_arn
                  app_name = ci_get(app, 'ApplicationName') or ci_get(req, 'ApplicationName')
                  if app_name:
                      return f"arn:aws:elasticbeanstalk:{region}:{account_id}:application/{app_name}"

              # AWS Elastic Beanstalk Environment
              elif event_name == 'CreateEnvironment' and event_source == 'elasticbeanstalk.amazonaws.com':
                  # Same casing-variance class as CreateApplication above.
                  env_arn = ci_get(resp, 'EnvironmentArn')
                  if env_arn:
                      return env_arn

              # Amazon GameLift Build
              elif event_name == 'CreateBuild' and event_source == 'gamelift.amazonaws.com':
                  build = resp.get('Build', {})
                  build_arn = build.get('BuildArn') or build.get('buildArn')
                  if build_arn:
                      return build_arn

              # Amazon GameLift Script
              elif event_name == 'CreateScript' and event_source == 'gamelift.amazonaws.com':
                  script = resp.get('Script', {})
                  script_arn = script.get('ScriptArn') or script.get('scriptArn')
                  if script_arn:
                      return script_arn

              # Amazon GameLift Fleet
              elif event_name == 'CreateFleet' and event_source == 'gamelift.amazonaws.com':
                  fleet_attrib = resp.get('FleetAttributes', {})
                  fleet_arn = fleet_attrib.get('FleetArn')
                  if fleet_arn:
                      return fleet_arn

              # RDS Snapshots (flat response — ARN is at top level)
              elif event_name == 'CreateDBSnapshot':
                  arn = resp.get('dBSnapshotArn')
                  if not arn:
                      arn = resp.get('dBSnapshot', {}).get('dBSnapshotArn')
                  if arn:
                      return arn

              elif event_name == 'CreateDBClusterSnapshot':
                  arn = resp.get('dBClusterSnapshotArn')
                  if not arn:
                      arn = resp.get('dBClusterSnapshot', {}).get('dBClusterSnapshotArn')
                  if arn:
                      return arn

              elif event_name == 'CreateDBInstanceReadReplica':
                  arn = resp.get('dBInstanceArn')
                  if not arn:
                      arn = resp.get('dBInstance', {}).get('dBInstanceArn')
                  if arn:
                      return arn

              # Bedrock Application Inference Profile
              elif event_name == 'CreateInferenceProfile':
                  inf_arn = resp.get('inferenceProfileArn')
                  if inf_arn:
                      return inf_arn

              # Kinesis Analytics v2 (KDA)
              elif event_name == 'CreateApplication' and event_source == 'kinesisanalytics.amazonaws.com':
                  app_arn = resp.get('applicationDetail', {}).get('applicationARN')
                  if not app_arn:
                      app_arn = resp.get('ApplicationARN')
                  if app_arn:
                      return app_arn

              # Redshift Serverless Namespace
              elif event_name == 'CreateNamespace':
                  ns = resp.get('namespace', {})
                  ns_arn = ns.get('namespaceArn')
                  if ns_arn:
                      return ns_arn

              # Redshift Serverless Workgroup (empty responseElements — construct from request)
              elif event_name == 'CreateWorkgroup':
                  wg = resp.get('workgroup', {})
                  wg_arn = wg.get('workgroupArn')
                  if not wg_arn:
                      wg_name = req.get('workgroupName')
                      if wg_name:
                          wg_arn = f"arn:aws:redshift-serverless:{region}:{account_id}:workgroup/{wg_name}"
                  if wg_arn:
                      return wg_arn

              # Carrier Gateway
              elif event_name == 'CreateCarrierGateway':
                  cagw = resp.get('carrierGateway', {})
                  if not cagw and 'CreateCarrierGatewayResponse' in resp:
                      cagw = resp['CreateCarrierGatewayResponse'].get('carrierGateway', {})
                  cagw_id = cagw.get('carrierGatewayId')
                  if cagw_id:
                      return f"arn:aws:ec2:{region}:{account_id}:carrier-gateway/{cagw_id}"

              # Glue Trigger
              elif event_name == 'CreateTrigger' and event_source == 'glue.amazonaws.com':
                  trigger_name = resp.get('name')
                  if trigger_name:
                      return f"arn:aws:glue:{region}:{account_id}:trigger/{trigger_name}"

              # VPC Endpoint
              elif event_name == 'CreateVpcEndpoint':
                  vpc_endpoint = resp.get('vpcEndpoint', {})
                  if not vpc_endpoint and 'CreateVpcEndpointResponse' in resp:
                      vpc_endpoint = resp['CreateVpcEndpointResponse'].get('vpcEndpoint', {})
                  vpc_endpoint_id = vpc_endpoint.get('vpcEndpointId')
                  if vpc_endpoint_id:
                      return f"arn:aws:ec2:{region}:{account_id}:vpc-endpoint/{vpc_endpoint_id}"

              # API Gateway v2 VPC Link
              elif event_name == 'CreateVpcLink' and event_source == 'apigateway.amazonaws.com':
                  vpc_link_id = resp.get('vpcLinkId')
                  if vpc_link_id:
                      return f"arn:aws:apigateway:{region}::/vpclinks/{vpc_link_id}"

              # App Mesh (3-level deep: mesh.metadata.arn)
              elif event_name == 'CreateMesh':
                  mesh_arn = resp.get('mesh', {}).get('metadata', {}).get('arn')
                  if mesh_arn:
                      return mesh_arn

              # API Gateway API Key — handler removed 2026-04-26 (plan-PR #56 D11).
              # CreateApiKey is now in IGNORE_EVENTS; its ARN shape is rejected
              # by all tagging APIs (RGTA + native) per
              # docs/MAP_TAGGING_GAP_ANALYSIS.md.

              # CloudWatch Insights Query Definition
              # CloudWatch Logs Insights QueryDefinition — §1.86. Not taggable
              # per AWS (the `arn:aws:logs:<region>:<acct>:query-definition:<id>`
              # shape is rejected by RGTA). Moved to IGNORE_EVENTS; dead
              # handler removed.

              # Athena WorkGroup
              elif event_name == 'CreateWorkGroup':
                  workgroup_name = req.get('name')
                  if workgroup_name:
                      return f"arn:aws:athena:{region}:{account_id}:workgroup/{workgroup_name}"

              # Service Discovery HTTP Namespace — §1.87. AWS returns an
              # operationId, not the namespace ID. Constructing the ARN would
              # require an async DescribeOperation poll (possibly minutes) that
              # doesn't fit the 180s SQS visibility window. Moved to
              # IGNORE_EVENTS rather than leaving a dead handler that generates
              # no_arn skips (and potential SNS noise if any follow-up promotes
              # no_arn to a permanent failure class).

              # AWS Security Hub — §1.84. EnableSecurityHub returns a null body,
              # so the universal ARN scan has nothing to extract. The hub is
              # always named `default` (the only supported case per AWS docs);
              # construct the ARN directly.
              elif event_name == 'EnableSecurityHub' and event_source == 'securityhub.amazonaws.com':
                  return f"arn:aws:securityhub:{region}:{account_id}:hub/default"

              # AWS Storage Gateway — §1.85. ActivateGateway returns the ARN on
              # `GatewayARN` (uppercase per the Storage Gateway API shape), which
              # the allowlist didn't include prior to PR #43a's suffix-match.
              # The suffix-match catches it, but Storage Gateway also requires
              # native `storagegateway:AddTagsToResource` dispatch (RGTA doesn't
              # cover it) — enforced in tag_resource below.
              elif event_name == 'ActivateGateway' and event_source == 'storagegateway.amazonaws.com':
                  gw_arn = ci_get(resp, 'GatewayARN') or ci_get(resp, 'gatewayArn')
                  if gw_arn:
                      return gw_arn

              # (CreateLoadBalancer handled above at the ELBv2 branch — both
              # ELBv1 Classic and ELBv2 share eventSource.)

              # ACM Private CA
              elif event_name == 'CreateCertificateAuthority':
                  ca_arn = resp.get('certificateAuthorityArn')
                  if ca_arn:
                      return ca_arn

              # RAM Resource Share
              elif event_name == 'CreateResourceShare':
                  share = resp.get('resourceShare', {})
                  share_arn = share.get('resourceShareArn')
                  if share_arn:
                      return share_arn

              # Storage Gateway Tape Pool
              elif event_name == 'CreateTapePool':
                  pool_arn = resp.get('poolARN')
                  if pool_arn:
                      return pool_arn

              # Cognito User Pool
              elif event_name == 'CreateUserPool' and event_source == 'cognito-idp.amazonaws.com':
                  pool = resp.get('userPool', {})
                  pool_id = pool.get('id')
                  if pool_id:
                      return f"arn:aws:cognito-idp:{region}:{account_id}:userpool/{pool_id}"

              # Cognito Identity Pool
              elif event_name == 'CreateIdentityPool':
                  pool_id = resp.get('identityPoolId')
                  if pool_id:
                      return f"arn:aws:cognito-identity:{region}:{account_id}:identitypool/{pool_id}"

              # Network Firewall Policy
              elif event_name == 'CreateFirewallPolicy':
                  nfw_resp = resp.get('firewallPolicyResponse', {})
                  nfw_arn = nfw_resp.get('firewallPolicyArn')
                  if nfw_arn:
                      return nfw_arn

              # Network Firewall
              elif event_name == 'CreateFirewall':
                  fw = resp.get('firewall', {})
                  fw_arn = fw.get('firewallArn')
                  if fw_arn:
                      return fw_arn

              # AWS WAFv2 WebACL
              # responseElements.summary.aRN — works for both REGIONAL and CLOUDFRONT scope.
              # (CLOUDFRONT-scope WebACLs only exist in us-east-1; ARN reflects scope via
              #  global/ vs regional/ path prefix.)
              elif event_name == 'CreateWebACL' and event_source == 'wafv2.amazonaws.com':
                  summary = resp.get('summary', {})
                  webacl_arn = summary.get('aRN') or summary.get('ARN') or summary.get('arn')
                  if webacl_arn:
                      return webacl_arn

              # AWS WAFv2 IPSet
              elif event_name == 'CreateIPSet' and event_source == 'wafv2.amazonaws.com':
                  summary = resp.get('summary', {})
                  ipset_arn = summary.get('aRN') or summary.get('ARN') or summary.get('arn')
                  if ipset_arn:
                      return ipset_arn

              # AWS CodeDeploy Application — response has only applicationId; build ARN.
              elif event_name == 'CreateApplication' and event_source == 'codedeploy.amazonaws.com':
                  app_name = req.get('applicationName')
                  if app_name:
                      return f"arn:aws:codedeploy:{region}:{account_id}:application:{app_name}"

              # AWS CodeDeploy DeploymentGroup — response has only deploymentGroupId; build ARN.
              elif event_name == 'CreateDeploymentGroup' and event_source == 'codedeploy.amazonaws.com':
                  app_name = req.get('applicationName')
                  dg_name = req.get('deploymentGroupName')
                  if app_name and dg_name:
                      return f"arn:aws:codedeploy:{region}:{account_id}:deploymentgroup:{app_name}/{dg_name}"

              # EKS Add-on
              elif event_name == 'CreateAddon':
                  addon = resp.get('addon', {})
                  addon_arn = addon.get('addonArn')
                  if addon_arn:
                      return addon_arn

              # (Lambda Layers and Aliases are not taggable — PublishLayerVersion is
              # in IGNORE_EVENTS. Aliases were already unhandled here.)

              # Transit Gateway VPC Attachment
              elif event_name == 'CreateTransitGatewayVpcAttachment':
                  tgwa = resp.get('transitGatewayVpcAttachment', {})
                  if not tgwa and 'CreateTransitGatewayVpcAttachmentResponse' in resp:
                      tgwa = resp['CreateTransitGatewayVpcAttachmentResponse'].get('transitGatewayVpcAttachment', {})
                  att_id = tgwa.get('transitGatewayAttachmentId')
                  if att_id:
                      return f"arn:aws:ec2:{region}:{account_id}:transit-gateway-attachment/{att_id}"

              # ElastiCache Snapshot
              elif event_name == 'CreateSnapshot' and event_source == 'elasticache.amazonaws.com':
                  snap = resp.get('snapshot', {})
                  snap_arn = snap.get('aRN') or snap.get('arn') or snap.get('ARN')
                  if not snap_arn:
                      snap_name = snap.get('snapshotName') or req.get('snapshotName')
                      if snap_name:
                          snap_arn = f"arn:aws:elasticache:{region}:{account_id}:snapshot:{snap_name}"
                  if snap_arn:
                      return snap_arn

              # WorkSpaces
              elif event_name == 'CreateWorkspaces':
                  pending = resp.get('pendingRequests', [])
                  if pending:
                      ws_id = pending[0].get('workspaceId')
                      if ws_id:
                          return f"arn:aws:workspaces:{region}:{account_id}:workspace/{ws_id}"

              # Route53 Health Check
              elif event_name == 'CreateHealthCheck':
                  health_check = resp.get('healthCheck', {})
                  health_check_id = health_check.get('id')
                  if health_check_id:
                      return f"arn:aws:route53:::healthcheck/{health_check_id}"

              # EC2 Key Pair
              elif event_name == 'CreateKeyPair':
                  key_pair_id = resp.get('keyPairId')
                  if key_pair_id:
                      return f"arn:aws:ec2:{region}:{account_id}:key-pair/{key_pair_id}"

              # DAX Cluster
              elif event_name == 'CreateCluster' and event_source == 'dax.amazonaws.com':
                  cluster = resp.get('cluster', {})
                  cluster_arn = cluster.get('clusterArn')
                  if cluster_arn:
                      return cluster_arn

              # DynamoDB RestoreTableFromBackup / RestoreTableToPointInTime
              elif event_name in ('RestoreTableFromBackup', 'RestoreTableToPointInTime') and event_source == 'dynamodb.amazonaws.com':
                  desc = resp.get('tableDescription', {})
                  table_arn = desc.get('tableArn')
                  if table_arn:
                      return table_arn

              # ElastiCache Serverless
              elif event_name == 'CreateServerlessCache' and event_source == 'elasticache.amazonaws.com':
                  sc = resp.get('serverlessCache', {})
                  sc_arn = sc.get('aRN') or sc.get('arn') or sc.get('ARN')
                  if sc_arn:
                      return sc_arn

              # MSK CreateClusterV2 (MSK Serverless / Provisioned v2)
              elif event_name == 'CreateClusterV2' and event_source == 'kafka.amazonaws.com':
                  cluster_arn = resp.get('clusterArn')
                  if cluster_arn:
                      return cluster_arn

              # Amazon OpenSearch Service (managed — NOT Serverless)
              elif event_name == 'CreateDomain' and event_source == 'es.amazonaws.com':
                  domain_status = resp.get('domainStatus', {})
                  domain_arn = domain_status.get('ARN') or domain_status.get('arn')
                  if domain_arn:
                      return domain_arn

              # SageMaker Domain (Studio) — case-insensitive per §1.91 class
              elif event_name == 'CreateDomain' and event_source == 'sagemaker.amazonaws.com':
                  domain_arn = ci_get(resp, 'domainArn')
                  if domain_arn:
                      return domain_arn

              # SageMaker Pipeline
              elif event_name == 'CreatePipeline' and event_source == 'sagemaker.amazonaws.com':
                  pipeline_arn = ci_get(resp, 'pipelineArn')
                  if pipeline_arn:
                      return pipeline_arn

              # SageMaker Feature Group
              elif event_name == 'CreateFeatureGroup' and event_source == 'sagemaker.amazonaws.com':
                  fg_arn = ci_get(resp, 'featureGroupArn')
                  if fg_arn:
                      return fg_arn

              # FSx Backup
              elif event_name == 'CreateBackup' and event_source == 'fsx.amazonaws.com':
                  backup = resp.get('backup', {})
                  backup_arn = backup.get('resourceARN')
                  if backup_arn:
                      return backup_arn

              # FSx CreateFileSystemFromBackup
              elif event_name == 'CreateFileSystemFromBackup' and event_source == 'fsx.amazonaws.com':
                  fs = resp.get('fileSystem', {})
                  fs_arn = fs.get('resourceARN')
                  if fs_arn:
                      return fs_arn

              # FSx OpenZFS Snapshot — response.snapshot.ResourceARN (capital R in ARN)
              elif event_name == 'CreateSnapshot' and event_source == 'fsx.amazonaws.com':
                  snapshot = resp.get('snapshot', {})
                  snap_arn = snapshot.get('ResourceARN') or snapshot.get('resourceARN')
                  if snap_arn:
                      return snap_arn

              # Redshift Serverless Snapshot
              elif event_name == 'CreateSnapshot' and event_source == 'redshift-serverless.amazonaws.com':
                  snapshot = resp.get('snapshot', {})
                  snap_arn = snapshot.get('snapshotArn')
                  if snap_arn:
                      return snap_arn

              # DMS Serverless (CreateReplicationConfig)
              elif event_name == 'CreateReplicationConfig' and event_source == 'dms.amazonaws.com':
                  rc = resp.get('replicationConfig', {})
                  rc_arn = rc.get('replicationConfigArn')
                  if rc_arn:
                      return rc_arn

              # Neptune (uses same event names as RDS but different source)
              elif event_name in ('CreateDBCluster', 'CreateDBInstance') and event_source == 'rds.amazonaws.com':
                  # Already handled by existing RDS handlers — this catches Neptune/DocDB
                  # which share the rds.amazonaws.com event source
                  pass

              # Transfer Family Connector
              elif event_name == 'CreateConnector' and event_source == 'transfer.amazonaws.com':
                  connector_id = resp.get('connectorId')
                  if connector_id:
                      return f"arn:aws:transfer:{region}:{account_id}:connector/{connector_id}"

              # Transfer Family User
              elif event_name == 'CreateUser' and event_source == 'transfer.amazonaws.com':
                  server_id = req.get('serverId')
                  user_name = resp.get('userName') or req.get('userName')
                  if server_id and user_name:
                      return f"arn:aws:transfer:{region}:{account_id}:user/{server_id}/{user_name}"

              # Direct Connect LAG
              elif event_name == 'CreateLag' and event_source == 'directconnect.amazonaws.com':
                  lag_id = resp.get('lagId')
                  if lag_id:
                      return f"arn:aws:directconnect:{region}:{account_id}:dxlag/{lag_id}"

              # Direct Connect Gateway
              elif event_name == 'CreateDirectConnectGateway' and event_source == 'directconnect.amazonaws.com':
                  gw = resp.get('directConnectGateway', {})
                  gw_id = gw.get('directConnectGatewayId')
                  if gw_id:
                      return f"arn:aws:directconnect::{account_id}:dx-gateway/{gw_id}"

              # Kinesis Video Streams (disambiguate from Kinesis Data Streams)
              elif event_name == 'CreateStream' and event_source == 'kinesisvideo.amazonaws.com':
                  stream_arn = resp.get('streamARN') or resp.get('streamArn')
                  if stream_arn:
                      return stream_arn

              # Elastic Disaster Recovery (DRS) Source Server
              elif event_name == 'CreateSourceServer' and event_source == 'drs.amazonaws.com':
                  source_arn = resp.get('arn') or resp.get('sourceServerArn')
                  if source_arn:
                      return source_arn

              # Deadline Cloud Farm
              elif event_name == 'CreateFarm' and event_source == 'deadline.amazonaws.com':
                  farm_id = resp.get('farmId')
                  if farm_id:
                      return f"arn:aws:deadline:{region}:{account_id}:farm/{farm_id}"

              # Deadline Cloud Queue
              elif event_name == 'CreateQueue' and event_source == 'deadline.amazonaws.com':
                  farm_id = req.get('farmId')
                  queue_id = resp.get('queueId')
                  if farm_id and queue_id:
                      return f"arn:aws:deadline:{region}:{account_id}:farm/{farm_id}/queue/{queue_id}"

              # Deadline Cloud Fleet
              elif event_name == 'CreateFleet' and event_source == 'deadline.amazonaws.com':
                  farm_id = req.get('farmId')
                  fleet_id = resp.get('fleetId')
                  if farm_id and fleet_id:
                      return f"arn:aws:deadline:{region}:{account_id}:farm/{farm_id}/fleet/{fleet_id}"

              # IoT Core Topic Rule
              elif event_name == 'CreateTopicRule' and event_source == 'iot.amazonaws.com':
                  rule_name = req.get('ruleName')
                  if rule_name:
                      return f"arn:aws:iot:{region}:{account_id}:rule/{rule_name}"

              # IoT SiteWise Asset
              elif event_name == 'CreateAsset' and event_source == 'iotsitewise.amazonaws.com':
                  asset_id = resp.get('assetId')
                  if asset_id:
                      return f"arn:aws:iotsitewise:{region}:{account_id}:asset/{asset_id}"

              # IoT SiteWise Asset Model
              elif event_name == 'CreateAssetModel' and event_source == 'iotsitewise.amazonaws.com':
                  model_id = resp.get('assetModelId')
                  if model_id:
                      return f"arn:aws:iotsitewise:{region}:{account_id}:asset-model/{model_id}"

              # IoT SiteWise Gateway
              elif event_name == 'CreateGateway' and event_source == 'iotsitewise.amazonaws.com':
                  gw_id = resp.get('gatewayId')
                  if gw_id:
                      return f"arn:aws:iotsitewise:{region}:{account_id}:gateway/{gw_id}"

              # IoT SiteWise Portal
              elif event_name == 'CreatePortal' and event_source == 'iotsitewise.amazonaws.com':
                  portal_id = resp.get('portalId')
                  if portal_id:
                      return f"arn:aws:iotsitewise:{region}:{account_id}:portal/{portal_id}"

              return None

          # Shared throttle-code set used by both native dispatch branches and
          # the RGTA fallthrough. RGTA throws 'ThrottledException' (with "ed")
          # — most other AWS services throw 'ThrottlingException' (with "ing").
          # Both variants must be enumerated, alongside the generic rate-limit
          # codes. Hoisted to module scope so _retry_throttles() and the RGTA
          # retry loop below share the same constant (v21.0.6, §1.81/§1.92).
          THROTTLE_CODES = {'ThrottlingException', 'ThrottledException', 'RequestLimitExceeded', 'TooManyRequestsException'}

          def _retry_throttles(fn):
              """Call fn() with exponential-backoff retry on throttle ClientError.

              Matches the pattern used by the RGTA fallthrough below — 4 attempts,
              1s → 2s → 4s → 8s (±25% jitter). Non-throttle ClientErrors propagate
              to the outer _process_event classifier on the first attempt. Absorbs
              short throttles inside the invocation so we don't burn one of the
              5 SQS redeliveries (180s VT each) on a recoverable rate limit
              (v21.0.6, §1.81/§1.92).
              """
              for attempt in range(4):
                  try:
                      return fn()
                  except ClientError as e:
                      code = e.response['Error']['Code']
                      if code in THROTTLE_CODES and attempt < 3:
                          sleep_time = (2 ** attempt) * 1.0 * (1 + random.uniform(-0.25, 0.25))
                          print(f"Native-tag throttled (code={code}), retry in {sleep_time:.1f}s (attempt {attempt+1}/4)")
                          time.sleep(sleep_time)
                          continue
                      raise

          def tag_resource(arn, tag_key, tag_value, tagging_client=None, ec2_client=None, target_region=None):
              _tagging = tagging_client or tagging
              _ec2 = ec2_client or ec2
              _region = target_region or boto3.session.Session().region_name

              def get_service_client(service):
                  return boto3.client(service, region_name=_region)

              # Handle special cases that don't support Resource Groups Tagging API.
              # Each native tag call is wrapped in _retry_throttles() so short
              # throttle bursts are absorbed in-invocation (up to 4 attempts,
              # ~15s total worst case) rather than burning SQS redelivery
              # budget. (§1.81/§1.92, v21.0.6)
              if ':s3:::' in arn:
                  s3 = get_service_client('s3')
                  bucket = arn.split(':::')[1]
                  existing = []
                  try:
                      existing = s3.get_bucket_tagging(Bucket=bucket)['TagSet']
                  except ClientError as e:
                      # `NoSuchTagSet` is the expected "bucket has no tags yet" response.
                      # Any other error (throttle, SCP deny, bucket-policy deny, transient
                      # 5xx) means we do NOT know the existing tag state — re-raise so
                      # SQS redrives rather than overwriting the customer's TagSet with
                      # just map-migrated. S3 TagSet replacement is destructive (full
                      # replace, not merge) — swallowing errors here deletes customer
                      # tags (Owner, CostCenter, etc.). See §1.2 / U3.
                      if e.response['Error']['Code'] != 'NoSuchTagSet':
                          raise
                  existing = [t for t in existing if t['Key'] != tag_key]
                  existing.append({'Key': tag_key, 'Value': tag_value})
                  _retry_throttles(lambda: s3.put_bucket_tagging(Bucket=bucket, Tagging={'TagSet': existing}))
                  return True
              elif ':bedrock:' in arn and any(r in arn for r in [':agent/',':flow/',':prompt/',':knowledge-base/']):
                  ba = boto3.client('bedrock-agent', region_name=arn.split(':')[3])
                  _retry_throttles(lambda: ba.tag_resource(resourceArn=arn, tags={tag_key: tag_value}))
                  return True
              elif ':quicksight:' in arn:
                  # QuickSight resources must be tagged in the region where they exist
                  qs_region = arn.split(':')[3] or region
                  qs = boto3.client('quicksight', region_name=qs_region)
                  account = arn.split(':')[4]
                  _retry_throttles(lambda: qs.tag_resource(ResourceArn=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':globalaccelerator:' in arn:
                  ga = boto3.client('globalaccelerator', region_name='us-west-2')
                  _retry_throttles(lambda: ga.tag_resource(ResourceArn=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':cloudfront:' in arn:
                  cf = boto3.client('cloudfront')
                  _retry_throttles(lambda: cf.tag_resource(Resource=arn, Tags={'Items': [{'Key': tag_key, 'Value': tag_value}]}))
                  return True
              elif ':route53:::hostedzone/' in arn:
                  r53 = boto3.client('route53')
                  zone_id = arn.split('/')[-1]
                  _retry_throttles(lambda: r53.change_tags_for_resource(ResourceType='hostedzone', ResourceId=zone_id, AddTags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':kinesis:' in arn and ':stream/' in arn:
                  kinesis = get_service_client('kinesis')
                  stream_name = arn.split('/')[-1]
                  _retry_throttles(lambda: kinesis.add_tags_to_stream(StreamName=stream_name, Tags={tag_key: tag_value}))
                  return True
              elif ':kinesisvideo:' in arn:
                  kv = get_service_client('kinesisvideo')
                  _retry_throttles(lambda: kv.tag_stream(StreamARN=arn, Tags={tag_key: tag_value}))
                  return True
              elif ':firehose:' in arn:
                  firehose = get_service_client('firehose')
                  stream_name = arn.split('/')[-1]
                  _retry_throttles(lambda: firehose.tag_delivery_stream(DeliveryStreamName=stream_name, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':apigateway:' in arn:
                  apigw = get_service_client('apigateway') if '/restapis/' in arn else get_service_client('apigatewayv2')
                  if '/restapis/' in arn:
                      _retry_throttles(lambda: apigw.tag_resource(resourceArn=arn, tags={tag_key: tag_value}))
                  else:
                      _retry_throttles(lambda: apigw.tag_resource(ResourceArn=arn, Tags={tag_key: tag_value}))
                  return True
              elif ':autoscaling:' in arn:
                  asg = get_service_client('autoscaling')
                  name = arn.split('/')[-1]
                  _retry_throttles(lambda: asg.create_or_update_tags(Tags=[{
                      'ResourceId': name,
                      'ResourceType': 'auto-scaling-group',
                      'Key': tag_key,
                      'Value': tag_value,
                      'PropagateAtLaunch': False
                  }]))
                  return True
              elif ':sqs:' in arn:
                  sqs = get_service_client('sqs')
                  queue_name = arn.split(':')[-1]
                  q_region = arn.split(':')[3]
                  q_account = arn.split(':')[4]
                  queue_url = f"https://sqs.{q_region}.amazonaws.com/{q_account}/{queue_name}"
                  _retry_throttles(lambda: sqs.tag_queue(QueueUrl=queue_url, Tags={tag_key: tag_value}))
                  return True
              elif ':memorydb:' in arn:
                  memorydb = get_service_client('memorydb')
                  _retry_throttles(lambda: memorydb.tag_resource(ResourceArn=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':dax:' in arn:
                  # §1.84. RGTA does not support DAX — native dax.tag_resource
                  # takes `ResourceName=<arn>` per the boto3 service model.
                  dax = get_service_client('dax')
                  _retry_throttles(lambda: dax.tag_resource(ResourceName=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':storagegateway:' in arn:
                  # §1.85. RGTA does not support Storage Gateway — native
                  # storagegateway.add_tags_to_resource takes the ARN directly.
                  sg = get_service_client('storagegateway')
                  _retry_throttles(lambda: sg.add_tags_to_resource(ResourceARN=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':iot:' in arn:
                  iot = get_service_client('iot')
                  _retry_throttles(lambda: iot.tag_resource(resourceArn=arn, tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':cassandra:' in arn:
                  # Keyspaces TagResource uses lowercase key/value (service inconsistency).
                  ks = get_service_client('keyspaces')
                  _retry_throttles(lambda: ks.tag_resource(resourceArn=arn, tags=[{'key': tag_key, 'value': tag_value}]))
                  return True
              elif ':cloudhsm:' in arn and ':cluster/' in arn:
                  # cloudhsmv2 TagResource takes the raw cluster_id as ResourceId.
                  chsm = get_service_client('cloudhsmv2')
                  cluster_id = arn.split('cluster/')[-1]
                  _retry_throttles(lambda: chsm.tag_resource(ResourceId=cluster_id, TagList=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':ds:' in arn and ':directory/' in arn:
                  # ds AddTagsToResource takes the raw directory_id as ResourceId.
                  ds = get_service_client('ds')
                  dir_id = arn.split('directory/')[-1]
                  _retry_throttles(lambda: ds.add_tags_to_resource(ResourceId=dir_id, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':dsql:' in arn:
                  _retry_throttles(lambda: get_service_client('dsql').tag_resource(resourceArn=arn, tags={tag_key: tag_value}))
                  return True
              elif ':vpc-lattice:' in arn:
                  _retry_throttles(lambda: get_service_client('vpc-lattice').tag_resource(resourceArn=arn, tags={tag_key: tag_value}))
                  return True
              elif ':bedrock-agentcore:' in arn:
                  r = arn.split(':')[3] or region
                  import urllib.parse, urllib.request, json as _json
                  from botocore.auth import SigV4Auth as _SigV4Auth
                  from botocore.awsrequest import AWSRequest as _AWSReq
                  def _agentcore_tag():
                      creds = boto3.Session().get_credentials().get_frozen_credentials()
                      encoded = urllib.parse.quote(arn, safe='')
                      url = f"https://bedrock-agentcore-control.{r}.amazonaws.com/tags/{encoded}"
                      body = _json.dumps({'tags': {tag_key: tag_value}}).encode()
                      aws_req = _AWSReq(method='POST', url=url, data=body, headers={'Content-Type': 'application/json'})
                      _SigV4Auth(creds, 'bedrock-agentcore', r).add_auth(aws_req)
                      http_req = urllib.request.Request(url, data=body, method='POST')
                      for h, v in aws_req.headers.items():
                          http_req.add_header(h, v)
                      try:
                          urllib.request.urlopen(http_req)
                      except urllib.error.HTTPError as he:
                          if he.code in (429, 503):
                              raise ClientError({'Error': {'Code': 'ThrottlingException', 'Message': str(he)}}, 'TagResource')
                          raise
                  _retry_throttles(_agentcore_tag)
                  return True
              elif ':payment-cryptography:' in arn:
                  r = arn.split(':')[3] or region
                  _retry_throttles(lambda: boto3.client('payment-cryptography', region_name=r).tag_resource(ResourceArn=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              elif ':networkmanager:' in arn:
                  _retry_throttles(lambda: boto3.client('networkmanager', region_name='us-east-1').tag_resource(ResourceArn=arn, Tags=[{'Key': tag_key, 'Value': tag_value}]))
                  return True
              else:
                  # Use universal Resource Groups Tagging API (with optional cross-account client).
                  # Retry up to 3 times with exponential backoff for throttling errors.
                  # Uses the module-level THROTTLE_CODES defined above so the native
                  # branches and this fallthrough share the same constant.
                  for attempt in range(4):
                      try:
                          response = _tagging.tag_resources(ResourceARNList=[arn], Tags={tag_key: tag_value})
                          if response.get('FailedResourcesMap'):
                              failures = response['FailedResourcesMap']
                              error_msg = failures.get(arn, {}).get('ErrorMessage', 'Unknown error')
                              raise Exception(f"Tagging API failed: {error_msg}")
                          return True
                      except ClientError as e:
                          code = e.response['Error']['Code']
                          if code in THROTTLE_CODES and attempt < 3:
                              sleep_time = (2 ** attempt) * 1.0 * (1 + random.uniform(-0.25, 0.25))
                              print(f"Throttled on attempt {attempt + 1}, retrying in {sleep_time:.1f}s")
                              time.sleep(sleep_time)
                              continue
                          # Fallback to EC2 tagging for VPC resources
                          if ':ec2:' in arn:
                              resource_id = arn.split('/')[-1]
                              _ec2.create_tags(Resources=[resource_id], Tags=[{'Key': tag_key, 'Value': tag_value}])
                              return True
                          raise e
                      except Exception as e:
                          if ':ec2:' in arn:
                              resource_id = arn.split('/')[-1]
                              _ec2.create_tags(Resources=[resource_id], Tags=[{'Key': tag_key, 'Value': tag_value}])
                              return True
                          raise e

          # Three-path error classifier (§1.115–§1.120):
          #
          #   TRANSIENT            → surface via batchItemFailures so SQS
          #                          redelivers up to maxReceiveCount=5
          #                          (~15 min budget); failing past that falls
          #                          into EventDLQ via the redelivery-exhaustion
          #                          path.
          #   PERMANENT_IGNORABLE  → silent ack + CW metric. The resource
          #                          genuinely no longer exists (deleted
          #                          between create and tag). Retrying forever
          #                          just wastes quota and floods alerts.
          #                          Terraform-rollback / test-infra churn is
          #                          the typical source.
          #   PERMANENT_ACTIONABLE → SNS publish + CW metric + send to EventDLQ
          #                          via batchItemFailures (redelivery-
          #                          exhaustion path). Treated as actionable by
          #                          humans.
          #
          # SCP AccessDenied at tag time is NOT classified here — Q3's
          # deploy-time preflight (simulate-principal-policy, PR #38) owns
          # preventing SCP-denied deploys. If a runtime SCP drift slips past
          # preflight, it falls into PERMANENT_ACTIONABLE with everything else
          # this classifier can't anticipate (B.7 class).
          _TRANSIENT_MARKERS = (
              'not in an available state',
              'not in available state',
              'is not currently available',
              'InvalidClusterState',
              'InvalidServerlessCacheStateFault',
              'InvalidDBClusterStateFault',
              'InvalidDBInstanceState',
              'CREATE_IN_PROGRESS',
              'UPDATE_IN_PROGRESS',
              'can not be updated',
              'rate of control plane requests',
              'ThrottlingException',
              'ThrottledException',
              'Throttling',
              'RequestLimitExceeded',
              'TooManyRequestsException',
              'LimitExceededException',
              # Directory Service (MS AD + Simple AD): directory stays in
              # "Creating" state for 5–10 min (Simple AD) or 25–45 min (MS AD).
              # Tagging during this window returns AWS error
              #   "The operation is not supported for directories in this
              #    state. Directory Status: Creating"
              # Without these markers the classifier routes to
              # permanent_actionable → SNS alert on every AD creation, even
              # though the condition resolves itself. Classifying as
              # transient retries through SQS redelivery (5 × 180s = 900s),
              # which succeeds for Simple AD within the retry budget. MS AD
              # provisioning exceeds 900s; its retries exhaust into
              # EventDLQ without false SNS alert noise, and the
              # ReconciliationFunction catches the tag on the next nightly
              # sweep. (§1.98)
              'Directory Status: Creating',
              'not supported for directories in this state',
              # S3 409: concurrent CreateBucket/PutBucketTagging or in-flight
              # lifecycle replication. Resolves on retry — AWS S3 error guide
              # explicitly classifies OperationAborted as retryable. (Q1)
              'OperationAborted',
              'conflicting conditional operation',
              # EC2 per-volume snapshot rate ceiling. Burst-create workloads
              # (backup sweeps, DR drills) hit this; retries succeed once the
              # token bucket refills.
              'SnapshotCreationPerVolumeRateExceeded',
              # AWS throttle variant: some services emit Throttling.User
              # (EC2, STS) distinct from plain 'Throttling'. Substring match
              # against 'Throttling' already catches it, but enumerate for
              # clarity alongside the other throttle codes.
              'Throttling.User',
              # ElastiCache provisioning race (§1.101). During rapid
              # CreateCacheCluster / CreateReplicationGroup bursts, the tag
              # API occasionally returns "<resource> is either not present or
              # not available" while the cluster is still settling. Retries
              # succeed once provisioning completes.
              'is either not present or not available',
              # SSM ConcurrentUpdateException (§1.44). PutParameter bursts
              # (e.g. customer automation rotating config in parallel) return
              # "concurrently modified" when two writers touch the same
              # parameter. Retry after SQS redelivery resolves it.
              'concurrently modified',
          )
          _PERMANENT_IGNORABLE_MARKERS = (
              # Resource genuinely deleted between create and tag event.
              # Retrying will not find it; alerting is noise.
              'ResourceNotFoundException',
              'Requested resource not found',
              'NoSuchBucket',
              'InvalidInstanceID.NotFound',
              'DBInstanceNotFound',
              'DBClusterNotFoundFault',
              'InvalidVolume.NotFound',
              # Bedrock system-defined inference profiles fire CreateInferenceProfile but the tag API returns this error. Application profiles (same event) tag normally; we silently ack the system ones.
              'System-defined Inference Profile is not taggable',
              'UPDATE_ROLLBACK_FAILED',
              'Tag related operations on the managed rule',
          )

          # Events to ignore — either not taggable, or would cause a feedback
          # loop (e.g. PutBucketTagging: tagging an S3 bucket fires this
          # event, which would re-trigger this Lambda, which would tag again
          # — infinite loop).
          _IGNORE_EVENTS = frozenset([
              'CreateGrant', 'CreateLogStream', 'CreateTags',
              'CreateOrUpdateTags', 'ModifyTags', 'DeleteTags',
              'PutBucketTagging', 'TagResource', 'TagResources',
              'UntagResource', 'UntagResources', 'AddTagsToResource',
              # CloudFormation change sets are intermediate — the stack
              # itself is tagged separately via CreateStack → TagResource.
              'CreateChangeSet',
              # Cognito user-pool clients are child resources of the
              # already-tagged user pool; they don't need their own tags.
              'CreateUserPoolClient',
              # Lambda Layers are not taggable — the Lambda TagResource API
              # regex rejects layer ARNs. Skip to avoid SNS alarm noise.
              'PublishLayerVersion', 'PublishLayerVersion20181031',
              # API Gateway API Keys — ARN shape rejected by all tagging
              # APIs. Documented in docs/MAP_TAGGING_GAP_ANALYSIS.md.
              'CreateApiKey',
              # Service Discovery HTTP Namespace — async operationId
              # resolution doesn't fit the SQS 180s VT. §1.87.
              'CreateHttpNamespace',
              # CloudWatch Logs Insights QueryDefinition — §1.86. ARN shape
              # rejected by RGTA + native.
              'PutQueryDefinition',
          ])

          def _classify_error(msg):
              if any(m in msg for m in _TRANSIENT_MARKERS):
                  return 'transient'
              if any(m in msg for m in _PERMANENT_IGNORABLE_MARKERS):
                  return 'permanent_ignorable'
              return 'permanent_actionable'

          def _emit_class_metric(error_class, mpe_id):
              try:
                  boto3.client('cloudwatch').put_metric_data(
                      Namespace='MapAutoTagger',
                      MetricData=[{
                          'MetricName': 'TagFailureByClass',
                          'Dimensions': [
                              {'Name': 'ErrorClass', 'Value': error_class},
                              {'Name': 'MpeId', 'Value': mpe_id},
                          ],
                          'Value': 1,
                          'Unit': 'Count',
                      }],
                  )
              except Exception as mx:
                  print(f'Could not emit TagFailureByClass metric ({error_class}): {mx}')

          def _publish_permanent_alert(arn, mpe_id, err_msg):
              alert_arn = os.environ.get('ALERT_TOPIC_ARN')
              if not alert_arn:
                  return
              try:
                  boto3.client('sns').publish(
                      TopicArn=alert_arn,
                      Subject='MAP Auto-Tagger: Permanent tagging failure',
                      Message=(
                          f'Failed to apply map-migrated tag (permanent-actionable class).\n\n'
                          f'Resource ARN: {arn}\n'
                          f'MPE ID: {mpe_id}\n'
                          f'Error: {err_msg}\n\n'
                          f'This error is not a known transient condition and is not a '
                          f'benign resource-deleted case. Investigate IAM policy, SCP '
                          f'drift, or tag-quota exhaustion. Event will land in EventDLQ.\n'
                      )
                  )
              except Exception as sns_e:
                  print(f'Could not publish SNS alert: {sns_e}')

          def _unwrap_sqs_record(record):
              """Extract the inner EventBridge event from one SQS record."""
              return json.loads(record['body'])

          def _process_event(eb_event, config):
              """Tag all ARNs implied by a single EventBridge event.

              Returns (status, error_class) where status is one of
              {'ok', 'skipped', 'transient', 'permanent_actionable'} and
              error_class is the classifier verdict for failures (or None).
              Callers treat 'ok' / 'skipped' / 'permanent_ignorable' as SQS
              acks and 'transient' / 'permanent_actionable' as retriable
              (batchItemFailures).
              """
              if not is_after_agreement(config):
                  print('Before agreement start date, skipping.')
                  return ('skipped', None)

              detail = eb_event.get('detail', {})
              account_id = detail.get('recipientAccountId', eb_event.get('account', ''))
              region = eb_event.get('region', detail.get('awsRegion', ''))

              if not is_in_scope(config, account_id, detail):
                  print(f'Resource not in MAP scope (account={account_id}), skipping.')
                  return ('skipped', None)

              if detail.get('errorCode') or detail.get('errorMessage'):
                  print(f'CloudTrail event has error, skipping: {detail.get("errorMessage")}')
                  return ('skipped', None)

              event_name = detail.get('eventName', '')
              if event_name in _IGNORE_EVENTS:
                  print(f'Skipping non-taggable event: {event_name}')
                  return ('skipped', None)

              # Events that emit multiple resources in a single CloudTrail event
              # (RunInstances → instances + attached EBS volumes + ENIs). AWS
              # does not publish separate CreateVolume / CreateNetworkInterface
              # events for children born inline with the parent.
              try:
                  arns = extract_arns_multi(detail, account_id, region)
              except ClientError as e:
                  # extract_arns_multi fails fast on InvalidInstanceID.Malformed
                  # so the classifier can route to permanent_actionable instead
                  # of burning the 30s describe_instances budget.
                  err_msg = str(e)
                  cls = _classify_error(err_msg)
                  _emit_class_metric(cls, config['mpe_id'])
                  if cls == 'transient':
                      return ('transient', err_msg)
                  if cls == 'permanent_ignorable':
                      return ('skipped', None)
                  _publish_permanent_alert(detail.get('eventName', '<unknown>'), config['mpe_id'], err_msg)
                  return ('permanent_actionable', err_msg)
              if not arns:
                  single = extract_arn(detail, account_id, region)
                  arns = [single] if single else []
              if not arns:
                  print(f'Could not extract ARN from event: {event_name}')
                  return ('skipped', None)

              tag_key = 'map-migrated'
              tag_value = config['mpe_id']

              # Per-ARN loop. A single record's ARNs share classification
              # fate: if any ARN classifies transient, the whole record goes
              # back to SQS (idempotent re-tag is safe); if any ARN is
              # permanent_actionable, the record goes to DLQ; otherwise ack.
              worst = 'ok'
              worst_err = None
              for arn in arns:
                  try:
                      tag_resource(arn, tag_key, tag_value, target_region=region)
                      print(f'Tagged {arn} with {tag_key}={tag_value}')
                      continue
                  except Exception as e:
                      err_msg = str(e)
                      cls = _classify_error(err_msg)
                      _emit_class_metric(cls, tag_value)

                      if cls == 'transient':
                          print(f'Transient [{arn}] {err_msg} — will retry via SQS')
                          # Transient trumps everything — retry wins.
                          return ('transient', err_msg)

                      if cls == 'permanent_ignorable':
                          print(f'Permanent-ignorable [{arn}] {err_msg} — resource no longer exists, skipping')
                          # Do not downgrade worst; keep scanning other ARNs.
                          continue

                      # permanent_actionable
                      print(f'Permanent-actionable [{arn}] {err_msg} — alerting + routing to DLQ')
                      _publish_permanent_alert(arn, tag_value, err_msg)
                      worst = 'permanent_actionable'
                      worst_err = err_msg
                      # Keep scanning remaining ARNs so siblings still tag
                      # (matches the pre-batch behavior's best-effort intent).

              return (worst, worst_err)

          def handler(event, context):
              """SQS batch handler.

              EventSourceMapping is configured with BatchSize=10 and
              FunctionResponseTypes=[ReportBatchItemFailures]. We iterate
              each SQS record, process its inner EventBridge event, and
              return a batchItemFailures list so only failing records get
              redelivered. Successful / skipped / permanent-ignorable
              records are acked by omission.
              """
              config = get_config()
              records = event.get('Records', [])

              # Backwards compatibility: if the Lambda is invoked directly
              # with a raw EventBridge event (no SQS envelope), process it
              # once. Used by ReconciliationFunction re-enqueue test paths
              # and local smoke tests.
              if not records:
                  status, _ = _process_event(event, config)
                  return {'status': status}

              failures = []
              for record in records:
                  msg_id = record.get('messageId')
                  try:
                      inner = _unwrap_sqs_record(record)
                      status, _ = _process_event(inner, config)
                      if status in ('transient', 'permanent_actionable'):
                          failures.append({'itemIdentifier': msg_id})
                  except Exception as e:
                      # Unexpected parse / unwrap error. Treat as transient
                      # so SQS redelivers — the DLQ will capture after
                      # maxReceiveCount if it's a poisoned message.
                      print(f'Unexpected error processing record {msg_id}: {e}')
                      failures.append({'itemIdentifier': msg_id})

              if failures:
                  print(f'Batch: {len(failures)}/{len(records)} records failed, reporting for redelivery')
              return {'batchItemFailures': failures}

  # EventBridge rule - matches ALL resource creation events
  AutoTagEventRule:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub 'map-auto-tagger-rule-${MpeId}'
      Description: Triggers MAP auto-tagger on resource creation events
      State: ENABLED
      EventPattern:
        detail-type:
          - AWS API Call via CloudTrail
        detail:
          eventName:
            - prefix: "Create"
            - prefix: "Run"
            - prefix: "Activate"
            - prefix: "Register"
            - prefix: "Put"
            - prefix: "Issue"
            - prefix: "Start"
            - prefix: "Request"    # ACM certificates
            - prefix: "Allocate"   # Elastic IPs, EBS snapshots
            - prefix: "Launch"     # EMR clusters, SageMaker jobs
            - prefix: "Import"     # EC2 AMIs, snapshots
            - prefix: "Publish"    # Lambda layers (PublishLayerVersion)
            - prefix: "Copy"       # RDS/EBS snapshots, AMIs
            - prefix: "Restore"    # RDS/Aurora/Redshift from snapshots
            - prefix: "Enable"     # SecurityHub
            - prefix: "Connect"    # AD Connector (ConnectDirectory)
      Targets:
        - Arn: !GetAtt EventQueue.Arn
          Id: MapAutoTaggerQueue

  # Permission for EventBridge to send to SQS
  EventQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref EventQueue
      PolicyDocument:
        Statement:
          - Effect: Allow
            Principal:
              Service: events.amazonaws.com
            Action: sqs:SendMessage
            Resource: !GetAtt EventQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !GetAtt AutoTagEventRule.Arn

  # SQS queue — buffers CloudTrail events for up to 14 days
  # Replaces direct EventBridge → Lambda invocation to avoid 24h retry limit
  EventQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub 'map-auto-tagger-events-${MpeId}'
      # VisibilityTimeout must be ≥ Lambda timeout (60s). Retry cadence needs to
      # cover slow-provisioning services: ElastiCache Serverless takes 3-8 min
      # to become taggable, Aurora clusters 5-10 min, MSK Serverless 3-5 min.
      # 5 retries × 180s = 900s total retry window, matching the E2E verify
      # poll budget. Previously 3 × 90s = 270s, which exhausted before
      # ElastiCache Serverless finished provisioning → DLQ + tagging miss.
      VisibilityTimeout: 180
      MessageRetentionPeriod: 1209600  # 14 days
      SqsManagedSseEnabled: true
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt EventDLQ.Arn
        maxReceiveCount: 5

  # Dead letter queue — receives events that fail after 5 processing attempts
  EventDLQ:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub 'map-auto-tagger-dlq-${MpeId}'
      MessageRetentionPeriod: 1209600  # 14 days
      SqsManagedSseEnabled: true

  # Lambda event source mapping — polls SQS and invokes Lambda.
  # BatchSize=10 with ReportBatchItemFailures: one invocation processes up to
  # 10 messages; only per-record failures are returned for redelivery (via
  # batchItemFailures). Raises drain rate ~10× vs prior BatchSize=1
  # (§1.123/§1.124 — measured 1.3 msg/s → expected ~10+ msg/s per Lambda).
  # MaximumBatchingWindowInSeconds=5 lets SQS fill small batches under steady
  # load without punishing latency on idle queues.
  EventQueueMapping:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      EventSourceArn: !GetAtt EventQueue.Arn
      FunctionName: !Ref AutoTaggerFunction
      BatchSize: 10
      MaximumBatchingWindowInSeconds: 5
      FunctionResponseTypes:
        - ReportBatchItemFailures
      Enabled: true

  # SNS topic for tagging alerts
  AlertTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: !Sub 'auto-map-tagger-alerts-${MpeId}'
      DisplayName: MAP 2.0 Auto-Tagger Alerts
      KmsMasterKeyId: alias/aws/sns

  # CloudWatch alarm - fires if Lambda errors spike
  TaggerErrorAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub 'map-auto-tagger-errors-${MpeId}'
      AlarmDescription: MAP auto-tagger Lambda is failing
      Namespace: AWS/Lambda
      MetricName: Errors
      Dimensions:
        - Name: FunctionName
          Value: !Ref AutoTaggerFunction
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 1
      Threshold: 3
      ComparisonOperator: GreaterThanOrEqualToThreshold
      AlarmActions:
        - !Ref AlertTopic

  # CloudWatch alarm - fires if events land in DLQ (failed after 5 SQS retries)
  DLQAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub 'map-auto-tagger-dlq-${MpeId}'
      AlarmDescription: >
        MAP auto-tagger events failed after 5 processing attempts and were moved
        to the dead letter queue. Check DLQ for stranded events and investigate
        Lambda errors. Events are retained for 14 days.
      Namespace: AWS/SQS
      MetricName: ApproximateNumberOfMessagesVisible
      Dimensions:
        - Name: QueueName
          Value: !GetAtt EventDLQ.QueueName
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      AlarmActions:
        - !Ref AlertTopic
      TreatMissingData: notBreaching

  # CloudWatch alarm — fires when the Lambda cold-start peer-tagger detector
  # (§1.108 Phase 16) finds another map-auto-tagger-mig* stack in this
  # account/region. Concurrent taggers race on the same CloudTrail events;
  # the map-migrated tag ends up with a non-deterministic MpeId value, which
  # can silently misattribute MAP credit. Alarm surface lets customers
  # discover this before a finance audit does.
  PeerTaggerDetectedAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub 'map-auto-tagger-peer-detected-${MpeId}'
      AlarmDescription: >
        Another map-auto-tagger stack was detected in this account at Lambda
        cold-start. Concurrent taggers race on CloudTrail events and the
        map-migrated tag value is last-writer-wins. Investigate which MPE is
        intended for which resources; if both are required, wait for the
        deterministic-routing feature (plan-PR #59) before relying on tag
        values for MAP credit attribution.
      Namespace: MapAutoTagger
      MetricName: PeerTaggerDetected
      Dimensions:
        - Name: MpeId
          Value: !Ref MpeId
      Statistic: Sum
      Period: 3600
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      AlarmActions:
        - !Ref AlertTopic
      TreatMissingData: notBreaching

  # CloudWatch alarm — slow, low-rate tagging failures that the per-minute
  # Errors alarm misses. Fires when ≥6 of the last 24 hourly buckets each
  # contain ≥1 permanent_actionable failure. This catches a trickle (one
  # failure/hour for 6+ hours) while ignoring one-off transients.
  TrickleFailureAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub 'map-auto-tagger-trickle-failures-${MpeId}'
      AlarmDescription: >
        Slow trickle of permanent-actionable tagging failures detected
        (≥6 of the last 24 hourly buckets had ≥1 failure). Investigate
        IAM drift, tag-quota exhaustion, or a new resource type the
        classifier isn't handling yet.
      Namespace: MapAutoTagger
      MetricName: TagFailureByClass
      Dimensions:
        - Name: ErrorClass
          Value: permanent_actionable
        - Name: MpeId
          Value: !Ref MpeId
      Statistic: Sum
      Period: 3600
      EvaluationPeriods: 24
      DatapointsToAlarm: 6
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      AlarmActions:
        - !Ref AlertTopic
      TreatMissingData: notBreaching

  # ── Reconciliation Lambda ────────────────────────────────────────────────
  # Daily safety-net that enumerates in-scope resources via RGTA GetResources,
  # diffs against map-migrated tag, and re-enqueues missing or wrong-MPE
  # resources through EventQueue so the live Lambda's classifier (PR #37)
  # handles the tag write via its normal path. Single tagging codepath.
  #
  # Design: docs/design-reconciliation.md (PR #36).
  # Closes the long-tail silent-failure classes the live path misses.

  ReconciliationLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      LogGroupName: !Sub '/aws/lambda/map-auto-tagger-reconciliation-${MpeId}'
      RetentionInDays: 14

  ReconciliationRole:
    Type: AWS::IAM::Role
    Properties:
      # IAM RoleName max 64 chars. Shortened "reconciliation" → "recon" so
      # `map-auto-tagger-recon-role-<MpeId>-<region>` fits (~60 chars for typical
      # 15-char MpeId + ap-northeast-2). The full-word name overran at 69 chars
      # and failed StackSet deploys (E2E run 24900531765, reconciliation Lambda initial ship).
      RoleName: !Sub 'map-auto-tagger-recon-role-${MpeId}-${AWS::Region}'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: map-auto-tagger-reconciliation-policy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: ReadScopeConfig
                Effect: Allow
                Action: ssm:GetParameter
                Resource: !Sub 'arn:aws:ssm:${AWS::Region}:${AWS::AccountId}:parameter/auto-map-tagger/${MpeId}/config'
              - Sid: EnumerateResources
                Effect: Allow
                Action:
                  # RGTA GetResources does not support resource-level IAM.
                  - tag:GetResources
                Resource: '*'
              - Sid: DescribeVpcMembership
                Effect: Allow
                Action:
                  # VPC-scope mode needs live VPC membership lookup for
                  # resources where RGTA omits VPC context.
                  - ec2:DescribeInstances
                  - ec2:DescribeVpcs
                  - ec2:DescribeVolumes
                  - ec2:DescribeNetworkInterfaces
                Resource: '*'
              - Sid: EnqueueToEventQueue
                Effect: Allow
                Action: sqs:SendMessage
                Resource: !GetAtt EventQueue.Arn
              - Sid: EmitReconciliationMetrics
                Effect: Allow
                Action: cloudwatch:PutMetricData
                Resource: '*'
                Condition:
                  StringEquals:
                    cloudwatch:namespace: MapAutoTagger
              - Sid: LambdaLogging
                Effect: Allow
                Action:
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: !Sub '${ReconciliationLogGroup.Arn}:*'

  ReconciliationFunction:
    Type: AWS::Lambda::Function
    DependsOn: ReconciliationLogGroup
    Properties:
      FunctionName: !Sub 'map-auto-tagger-reconciliation-${MpeId}'
      Runtime: python3.12
      Handler: index.handler
      Role: !GetAtt ReconciliationRole.Arn
      Timeout: 900
      MemorySize: 512
      Environment:
        Variables:
          CONFIG_PARAM: !Sub '/auto-map-tagger/${MpeId}/config'
          EVENT_QUEUE_URL: !Ref EventQueue
          MPE_ID: !Ref MpeId
      Code:
        ZipFile: |
          import json, os, time, boto3
          from datetime import datetime, timezone
          from botocore.exceptions import ClientError

          # Reconciliation Lambda.
          # Runs once per ReconciliationInterval (default daily). Enumerates
          # every taggable resource in this account via RGTA GetResources,
          # checks the map-migrated tag, and re-enqueues missing or wrong-MPE
          # resources through EventQueue so the live Lambda's Q1 classifier
          # (PR #37) handles the tag write. Reuses one tagging codepath.
          #
          # Wrong-MPE overwrite is safe-by-construction because Q3 Option D
          # preflight (PR #38) prevents overlapping-scope deploys. See
          # docs/design-reconciliation.md §3 for the safety argument.

          CONFIG_PARAM = os.environ['CONFIG_PARAM']
          EVENT_QUEUE_URL = os.environ['EVENT_QUEUE_URL']
          MPE_ID = os.environ['MPE_ID']
          TAG_KEY = 'map-migrated'
          # Run-time budget: Lambda hard ceiling is 900s. Emit the canary
          # metric at 13 min so operators see runs trending toward the cliff
          # before they actually hit it. Deferred checkpoint / pagination-
          # resume work per Q2-1 decision (see design doc §9).
          CANARY_THRESHOLD_SECS = 780

          ssm = boto3.client('ssm')
          sqs = boto3.client('sqs')
          cw = boto3.client('cloudwatch')
          tagging = boto3.client('resourcegroupstaggingapi')

          def _emit(name, value=1, **dims):
              try:
                  cw.put_metric_data(
                      Namespace='MapAutoTagger',
                      MetricData=[{
                          'MetricName': name,
                          'Dimensions': [{'Name': k, 'Value': str(v)} for k, v in dims.items()],
                          'Value': value,
                          'Unit': 'Count',
                      }],
                  )
              except Exception as e:
                  print(f'metric emit failed ({name}): {e}')

          def _load_config():
              resp = ssm.get_parameter(Name=CONFIG_PARAM)
              return json.loads(resp['Parameter']['Value'])

          def _in_agreement(cfg, creation_time):
              # Compare creation_time (datetime from RGTA) against agreement_start_date.
              # Out-of-agreement resources are NOT MAP-eligible and must NOT be tagged.
              try:
                  start = datetime.strptime(cfg['agreement_start_date'], '%Y-%m-%d').replace(tzinfo=timezone.utc)
              except (KeyError, ValueError) as e:
                  print(f'agreement_start_date invalid ({e}) — treating all resources as out-of-agreement')
                  _emit('ReconciliationConfigInvalid')
                  return False
              if creation_time is None:
                  # RGTA didn't provide CreationTime for this resource. Conservative
                  # default: skip it. Follow-up: per-service native fallback.
                  _emit('ReconciliationSkippedNoCreationTime')
                  return False
              if creation_time.tzinfo is None:
                  creation_time = creation_time.replace(tzinfo=timezone.utc)
              if creation_time < start:
                  return False
              end_str = cfg.get("agreement_end_date")
              if end_str:
                  try:
                      end = datetime.strptime(end_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
                      if creation_time > end:
                          return False
                  except (KeyError, ValueError):
                      pass
              return True

          def _current_map_tag_value(tag_list):
              for t in tag_list or []:
                  if t.get('Key') == TAG_KEY:
                      return t.get('Value')
              return None

          def _synth_event(arn, account_id, region):
              # Build a minimal CloudTrail-shaped envelope the live Lambda will
              # accept. The live Lambda's extract_arn supports resource-arn
              # passthrough; keep the detail focused.
              event_name = 'ReconciliationEnqueue'
              return {
                  'version': '0',
                  'source': 'map-auto-tagger.reconciliation',
                  'account': account_id,
                  'region': region,
                  'detail-type': 'AWS API Call via CloudTrail',
                  'detail': {
                      'eventName': event_name,
                      'recipientAccountId': account_id,
                      'awsRegion': region,
                      'responseElements': {'arn': arn},
                      'requestParameters': {},
                      'sourceIPAddress': 'map-auto-tagger.reconciliation',
                  },
              }

          def _parse_arn(arn):
              # arn:<partition>:<service>:<region>:<account>:<resource>
              parts = arn.split(':', 5)
              if len(parts) < 6:
                  return None, None
              return parts[4] or None, parts[3] or None

          def handler(event, context):
              start = time.time()
              cfg = _load_config()
              scanned = missing = wrong = already_ok = out_of_agree = 0
              canary_emitted = False
              # RGTA pagination. Default page size 100 (max).
              kwargs = {'ResourcesPerPage': 100, 'IncludeComplianceDetails': False}
              while True:
                  try:
                      page = tagging.get_resources(**kwargs)
                  except ClientError as e:
                      err = e.response.get('Error', {}).get('Code', '')
                      if err in ('ThrottledException', 'ThrottlingException'):
                          print(f'RGTA throttle: {e}, backing off 2s')
                          time.sleep(2)
                          continue
                      print(f'RGTA GetResources hard-failed: {e}')
                      _emit('ReconciliationRunAborted', reason='RgtaError')
                      raise
                  for r in page.get('ResourceTagMappingList', []):
                      scanned += 1
                      arn = r.get('ResourceARN', '')
                      if not arn:
                          continue
                      account_id, region = _parse_arn(arn)
                      if not account_id or not region:
                          continue
                      # Best-effort agreement-window filter. RGTA doesn't return
                      # CreationTime directly — we trust the agreement-check on
                      # the live Lambda side when the event enqueues. Reconciliation
                      # does not block on CreationTime; design doc §8.5 accepts
                      # 24h-lag semantics.
                      current = _current_map_tag_value(r.get('Tags'))
                      if current == MPE_ID:
                          already_ok += 1
                          continue
                      if current and current != MPE_ID:
                          wrong += 1
                          _emit('WrongMpeCorrected', ExpectedMpe=MPE_ID, FoundMpe=current)
                      else:
                          missing += 1
                          _emit('ReconciliationMissingTag')
                      try:
                          sqs.send_message(
                              QueueUrl=EVENT_QUEUE_URL,
                              MessageBody=json.dumps(_synth_event(arn, account_id, region)),
                          )
                      except ClientError as e:
                          print(f'SQS SendMessage failed for {arn}: {e}')
                          _emit('ReconciliationEnqueueFailed')
                  # Canary check between pages — cheap, avoids one-last-page overruns.
                  if not canary_emitted and (time.time() - start) > CANARY_THRESHOLD_SECS:
                      _emit('ReconciliationTimeoutCanary', MpeId=MPE_ID)
                      canary_emitted = True
                  token = page.get('PaginationToken')
                  if not token:
                      break
                  kwargs['PaginationToken'] = token
              _emit('ReconciliationResourcesScanned', scanned)
              print(json.dumps({
                  'scanned': scanned, 'already_ok': already_ok,
                  'missing': missing, 'wrong_mpe': wrong,
                  'out_of_agreement': out_of_agree,
                  'elapsed_secs': round(time.time() - start, 1),
              }))
              return {'status': 'completed', 'scanned': scanned, 'enqueued': missing + wrong}

  ReconciliationSchedule:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub 'map-auto-tagger-reconciliation-${MpeId}'
      Description: Daily safety-net reconciliation trigger
      ScheduleExpression: !Ref ReconciliationInterval
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReconciliationFunction.Arn
          Id: ReconciliationLambda

  ReconciliationSchedulePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref ReconciliationFunction
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt ReconciliationSchedule.Arn

Outputs:
  LambdaArn:
    Description: ARN of the MAP auto-tagger Lambda function
    Value: !GetAtt AutoTaggerFunction.Arn
  ConfigParameter:
    Description: SSM Parameter Store path for MAP tagger config
    Value: !Ref MapConfig
  TemplateVersion:
    Description: MAP 2.0 Auto-Tagger template version (pinned at deploy time)
    Value: v21.0.6
  AlertTopicArn:
    Description: SNS topic for tagger alerts - subscribe your email here
    Value: !Ref AlertTopic