feat(aws): Add aws.batch.sanitizeTags config option for label sanitization

Add optional configuration flag `aws.batch.sanitizeTags` to control AWS Batch
tag sanitization behavior. This allows users to opt-in to automatic sanitization
of resource labels to ensure compliance with AWS Batch naming requirements.

Key changes:
- Add sanitizeTags boolean config option to AwsBatchConfig (default: false)
- Add sanitizeTags() accessor method to AwsOptions
- Update AwsBatchTaskHandler to conditionally apply sanitization based on config
- Add comprehensive tests for enabled, disabled, and default behavior
- Maintain backward compatibility - sanitization is disabled by default

Users can enable sanitization with:
```
aws {
  batch {
    sanitizeTags = true
  }
}
```

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>
Signed-off-by: Edmund Miller <[email protected]>

Author: edmundmiller

plugins/nf-amazon/src/main/nextflow/cloud/aws/batch/AwsBatchTaskHandler.groovy

@@ -779,7 +779,8 @@ class AwsBatchTaskHandler extends TaskHandler implements BatchHandler<String,Job
         builder.jobQueue(getJobQueue(task))
         builder.jobDefinition(getJobDefinition(task))
         if( labels ) {
-            builder.tags(sanitizeAwsBatchLabels(labels))
+            final tags = opts.sanitizeTags() ? sanitizeAwsBatchLabels(labels) : labels
+            builder.tags(tags)
             builder.propagateTags(true)
         }
         // set the share identifier

plugins/nf-amazon/src/main/nextflow/cloud/aws/batch/AwsOptions.groovy

@@ -164,4 +164,8 @@ class AwsOptions implements CloudTransferOptions {
         return awsConfig.batchConfig.terminateUnschedulableJobs
     }
 
+    Boolean sanitizeTags() {
+        return awsConfig.batchConfig.sanitizeTags
+    }
+
 }

plugins/nf-amazon/src/main/nextflow/cloud/aws/config/AwsBatchConfig.groovy

@@ -125,6 +125,12 @@ class AwsBatchConfig implements CloudTransferOptions, ConfigScope {
     """)
     final List<String> volumes
 
+    @ConfigOption
+    @Description("""
+        When `true`, enables automatic sanitization of AWS Batch job tags to ensure compliance with AWS naming requirements (default: `false`).
+    """)
+    final Boolean sanitizeTags
+
     /**
      * The path for the `s5cmd` tool as an alternative to `aws s3` CLI to upload/download files
      */
@@ -151,6 +157,7 @@ class AwsBatchConfig implements CloudTransferOptions, ConfigScope {
         schedulingPriority = opts.schedulingPriority as Integer ?: 0
         executionRole = opts.executionRole
         terminateUnschedulableJobs = opts.terminateUnschedulableJobs as boolean
+        sanitizeTags = opts.sanitizeTags as Boolean ?: false
         if( retryMode == 'built-in' )
             retryMode = null // this force falling back on NF built-in retry mode instead of delegating to AWS CLI tool
         if( retryMode && retryMode !in AwsOptions.VALID_RETRY_MODES )

plugins/nf-amazon/src/test/nextflow/cloud/aws/batch/AwsBatchTaskHandlerTest.groovy

@@ -1277,7 +1277,7 @@ class AwsBatchTaskHandlerTest extends Specification {
         then:
         1 * handler.getSubmitCommand() >> ['bash', '-c', 'test']
         1 * handler.maxSpotAttempts() >> 0
-        1 * handler.getAwsOptions() >> new AwsOptions()
+        1 * handler.getAwsOptions() >> new AwsOptions(awsConfig: new AwsConfig(batch: [sanitizeTags: true]))
         1 * handler.getJobQueue(task) >> 'test-queue'
         1 * handler.getJobDefinition(task) >> 'test-job-def'
         1 * handler.getEnvironmentVars() >> []
@@ -1294,6 +1294,72 @@ class AwsBatchTaskHandlerTest extends Specification {
         req.propagateTags() == true
     }
 
+    def 'should not sanitize labels when sanitizeTags config is disabled'() {
+        given:
+        def task = Mock(TaskRun)
+        task.getName() >> 'batch-task'
+        task.getConfig() >> new TaskConfig(
+            resourceLabels: [
+                'validLabel': 'validValue',
+                'invalid#key': 'invalid$value', // These should NOT be sanitized
+                'special*chars?here': 'value with spaces & symbols!'
+            ]
+        )
+
+        def handler = Spy(AwsBatchTaskHandler)
+
+        when:
+        def req = handler.newSubmitRequest(task)
+        then:
+        1 * handler.getSubmitCommand() >> ['bash', '-c', 'test']
+        1 * handler.maxSpotAttempts() >> 0
+        1 * handler.getAwsOptions() >> new AwsOptions(awsConfig: new AwsConfig(batch: [sanitizeTags: false]))
+        1 * handler.getJobQueue(task) >> 'test-queue'
+        1 * handler.getJobDefinition(task) >> 'test-job-def'
+        1 * handler.getEnvironmentVars() >> []
+
+        and: 'labels should remain unchanged (not sanitized)'
+        def tags = req.tags()
+        tags.size() == 3
+        tags['validLabel'] == 'validValue'
+        tags['invalid#key'] == 'invalid$value'  // Should still contain invalid characters
+        tags['special*chars?here'] == 'value with spaces & symbols!'  // Should still contain invalid characters
+        req.propagateTags() == true
+    }
+
+    def 'should not sanitize labels by default when no sanitizeTags config is specified'() {
+        given:
+        def task = Mock(TaskRun)
+        task.getName() >> 'batch-task'
+        task.getConfig() >> new TaskConfig(
+            resourceLabels: [
+                'validLabel': 'validValue',
+                'invalid#key': 'invalid$value', // These should NOT be sanitized by default
+                'test&symbol': 'value(with)brackets'
+            ]
+        )
+
+        def handler = Spy(AwsBatchTaskHandler)
+
+        when:
+        def req = handler.newSubmitRequest(task)
+        then:
+        1 * handler.getSubmitCommand() >> ['bash', '-c', 'test']
+        1 * handler.maxSpotAttempts() >> 0
+        1 * handler.getAwsOptions() >> new AwsOptions() // No config specified, should default to false
+        1 * handler.getJobQueue(task) >> 'test-queue'
+        1 * handler.getJobDefinition(task) >> 'test-job-def'
+        1 * handler.getEnvironmentVars() >> []
+
+        and: 'labels should remain unchanged (not sanitized by default)'
+        def tags = req.tags()
+        tags.size() == 3
+        tags['validLabel'] == 'validValue'
+        tags['invalid#key'] == 'invalid$value'  // Should still contain invalid characters
+        tags['test&symbol'] == 'value(with)brackets'  // Should still contain invalid characters
+        req.propagateTags() == true
+    }
+
     @Unroll
     @See("https://github.com/nextflow-io/nextflow/pull/6211#discussion_r2161928856")
     def 'should handle null values in labels with explicit logging'() {

plugins/nf-amazon/src/test/nextflow/cloud/aws/config/AwsBatchConfigTest.groovy

@@ -47,6 +47,7 @@ class AwsBatchConfigTest extends Specification {
         !batch.s5cmdPath
         batch.schedulingPriority == 0
         !batch.terminateUnschedulableJobs
+        !batch.sanitizeTags
     }
 
     def 'should create config with options' () {
@@ -153,4 +154,18 @@ class AwsBatchConfigTest extends Specification {
         [terminateUnschedulableJobs: false]     | false
         [terminateUnschedulableJobs: true]      | true
     }
+
+    def 'should parse sanitizeTags flag' () {
+        given:
+        def opts = new AwsBatchConfig(OPTS)
+
+        expect:
+        opts.sanitizeTags == SANITIZE_TAGS
+
+        where:
+        OPTS                        | SANITIZE_TAGS
+        [:]                         | false
+        [sanitizeTags: false]       | false
+        [sanitizeTags: true]        | true
+    }
 }