Report the task directives referenced by the task script

modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy

@@ -984,6 +984,22 @@ class TaskRun implements Cloneable {
             return context.getVariableNames()
     }
 
+    /**
+     * Report whether the rendered task command depends on the value of the given
+     * {@code task} directive e.g. {@code memory} for a script interpolating
+     * {@code "-Xmx${task.memory.toGiga()}g"}.
+     *
+     * The command is rendered *before* the task is scheduled, therefore an executor that
+     * adjusts the requested resources at schedule time needs to know whether the command
+     * carries a value it is about to change.
+     *
+     * @param directive The directive name e.g. {@code memory}
+     * @return {@code true} when the task script references the given directive
+     */
+    boolean isDirectiveReferenced(String directive) {
+        return getVariableNames().contains("task.${directive}".toString())
+    }
+
     /**
      * @param variableNames The collection of variables referenced in the task script
      * @param binding The script global binding
@@ -1047,4 +1063,3 @@ class TaskRun implements Cloneable {
         return config?.getStubBlock()?.getSource()
     }
 }
-

modules/nextflow/src/test/groovy/nextflow/script/parser/v2/ProcessDirectiveRefsTest.groovy

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2013-2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package nextflow.script.parser.v2
+
+import nextflow.Session
+import nextflow.processor.TaskConfig
+import nextflow.processor.TaskContext
+import nextflow.processor.TaskProcessor
+import nextflow.processor.TaskRun
+import nextflow.script.ScriptMeta
+import test.Dsl2Spec
+
+/**
+ * Verify that the {@code task} directives referenced by a process script are reported
+ * through the variable references already collected for the task hash.
+ *
+ * @author Ben Sherman <ben.sherman@seqera.io>
+ */
+class ProcessDirectiveRefsTest extends Dsl2Spec {
+
+    private Set<String> valNames(String text, String processName='foo') {
+        def parser = new ScriptLoaderV2(new Session())
+        parser.setModule(true)
+        parser.parse(text)
+        parser.runScript()
+        final process = ScriptMeta.get(parser.getScript()).getProcess(processName)
+        return process.@taskBody.getValNames()
+    }
+
+    def 'should collect any referenced task property' () {
+        expect:
+        valNames('''
+            process foo {
+                script:
+                """
+                samtools sort -@ ${task.cpus} -m ${task.memory.toMega()}M in.bam
+                """
+            }
+            ''') == ['task.cpus', 'task.memory'] as Set
+    }
+
+    def 'should collect the task ext and params references' () {
+        expect:
+        valNames('''
+            process foo {
+                script:
+                """
+                echo ${task.ext.args} ${params.foo}
+                """
+            }
+            ''') == ['task.ext.args', 'params.foo'] as Set
+    }
+
+    def 'should report the referenced directive through the task run' () {
+        given:
+        def parser = new ScriptLoaderV2(new Session())
+        parser.setModule(true)
+        parser.parse('''
+            process foo {
+                script:
+                """
+                java -Xmx${task.memory.toGiga()}g -jar app.jar
+                """
+            }
+            ''')
+        parser.runScript()
+        and:
+        def processor = Mock(TaskProcessor) {
+            getTaskBody() >> ScriptMeta.get(parser.getScript()).getProcess('foo').@taskBody
+            getDeclaredNames() >> []
+        }
+        def task = new TaskRun(processor: processor, context: new TaskContext(processor))
+
+        expect:
+        task.isDirectiveReferenced('memory')
+        !task.isDirectiveReferenced('cpus')
+    }
+
+    def 'should not add the task references to the task hash' () {
+        given:
+        // `task` is put in the task context by TaskConfig#setContext, therefore every
+        // `task.*` reference is skipped by #getGlobalVars as a task-local variable. This is
+        // what allows the directive references to travel on `valRefs` without invalidating
+        // the resume cache of every pipeline -- `task.ext.*` is the documented exception,
+        // re-added to the hash by TaskHasher#getTaskExtensionDirectiveVars.
+        def context = new TaskContext(holder: [:])
+        def config = new TaskConfig(memory: '8 GB').setContext(context)
+        and:
+        def task = Spy(TaskRun)
+        task.processor = Mock(TaskProcessor) { getName() >> 'foo' }
+        task.context = context
+        task.config = config
+        task.getVariableNames() >> (['task.memory', 'task.ext.args', 'x'] as Set)
+
+        when:
+        def vars = task.getGlobalVars(new Binding(x: 1))
+
+        then:
+        context.isLocalVar('task')
+        and:
+        vars == [x: 1]
+    }
+}

modules/nf-lang/src/main/java/nextflow/script/control/ScriptToGroovyHelper.java

@@ -46,17 +46,15 @@ public ScriptToGroovyHelper(SourceUnit sourceUnit) {
     }
 
     /**
-     * Get the list of variable references in a statement.
+     * Get the list of variable references in a statement:
      *
-     * This method is used to collect variable references that are not
-     * declared as process inputs, so that they are included in the task
-     * hash. This covers:
-     *
-     * - task ext properties (e.g. `task.ext.args`)
+     * - task properties (e.g. `task.ext.args`, `task.memory`)
      * - script params (e.g. `params.x`)
      *
-     * These properties are typically used like inputs, but are not
-     * explicitly declared, so they must be identified by their usage.
+     * Params and `task.ext` properties are collected so that they
+     * can be added to the task hash, since they are indirect inputs.
+     * Other task directives are collected so that the runtime can
+     * check for directive references.
      *
      * The resulting list expression should be provided as the fourth
      * argument of the BodyDef constructor.
@@ -89,7 +87,7 @@ public void visitPropertyExpression(PropertyExpression node) {
             }
 
             var name = asPropertyChain(node);
-            if( name.startsWith("task.ext.") || name.startsWith("params.") )
+            if( name.startsWith("task.") || name.startsWith("params.") )
                 variableRefs.add(name);
         }
 

plugins/nf-seqera/README.md

@@ -107,6 +107,18 @@ seqera {
 }
 ```
 
+> NOTE: When a process references `task.memory` in the script (e.g. `-Xmx${task.memory.toGiga()}g`), resource prediction is disabled for that process. Otherwise, the scheduler could reduce the memory allocation, and the task would try to allocate more memory than is available, and fail with an out-of-memory error.
+
+Set the `seqera/predictionModel` hint explicitly on the process to override this behaviour:
+
+```groovy
+process {
+    withName: FOO {
+        hints = ['seqera/predictionModel': 'qr/v1']
+    }
+}
+```
+
 ## Resources
 
 - [Seqera Platform Documentation](https://docs.seqera.io/)

plugins/nf-seqera/src/main/io/seqera/executor/SeqeraTaskHandler.groovy

@@ -131,10 +131,14 @@ class SeqeraTaskHandler extends TaskHandler implements FusionAwareTask {
             fusionConfig().snapshotsEnabled(),
             maxSpotAttempts(baseMachineOpts)
         )
-        // resolve optional per-task prediction model override from the seqera/predictionModel hint;
-        // when unset the task inherits the run-level model
+        // Resolve the optional per-task prediction model override from the seqera/predictionModel
+        // hint. An explicit hint always wins: the automatic check below is a safety net, and the
+        // user asking for a specific model on a process is a deliberate opt-out of it.
+        // When neither applies the value is left null, so the task inherits the run-level model
         final predictionModelHint = HintHelper.resolvePredictionModel(task.config.getHints())
-        final predictionModel = predictionModelHint ? PredictionModel.fromValue(predictionModelHint) : null
+        final predictionModel = predictionModelHint
+            ? PredictionModel.fromValue(predictionModelHint)
+            : (shouldDisablePrediction() ? PredictionModel.NONE : null)
         // build resource limit from process resourceLimits directive (upper bound for OOM retry scaling)
         final resourceLim = toResourceLimit()
         // validate container - Seqera executor requires all processes to specify a container image
@@ -169,6 +173,35 @@ class SeqeraTaskHandler extends TaskHandler implements FusionAwareTask {
         executor.getBatchSubmitter().submit(this, schedTask)
     }
 
+    /**
+     * Determine whether the resource prediction model must be disabled for this task.
+     *
+     * The prediction model can allocate less memory than the task requested. However the task script
+     * is rendered *before* the task is scheduled, therefore a script referencing {@code task.memory}
+     * carries the memory that was requested, not the one that has been allocated e.g. a JVM
+     * {@code -Xmx} setting exceeding the container memory and failing with an out-of-memory error.
+     *
+     * @return {@code true} when the task should be submitted with prediction model {@code none}
+     */
+    protected boolean shouldDisablePrediction() {
+        // Nothing to disable unless the run enables a prediction model. Checking this first
+        // also keeps the warning quiet for the runs where the resources are never adjusted,
+        // which would otherwise report a problem that cannot happen
+        final runModel = executor.getSeqeraConfig()?.predictionModel
+        if( !runModel || runModel == PredictionModel.NONE.getValue() )
+            return false
+        // Note only `memory` is checked. The scheduler can adjust the cpus as well, but a
+        // stale `task.cpus` costs an over-subscribed thread pool whereas a stale
+        // `task.memory` fails the task, and `task.cpus` is referenced by nearly every
+        // process -- disabling the prediction for those would defeat the feature
+        if( !task.isDirectiveReferenced('memory') )
+            return false
+        // warn once per process rather than once per task: the reference belongs to the
+        // process definition, so every one of its tasks would otherwise report it
+        log.warn1("Process `${task.processor?.name ?: task.lazyName()}` depends on the `task.memory` value -- resource prediction has been disabled for this process to prevent an under-allocation of the requested memory", firstOnly: true)
+        return true
+    }
+
     protected int maxSpotAttempts(MachineRequirementOpts opts) {
         final result = opts?.maxSpotAttempts
         if( result != null && result < 0 )

plugins/nf-seqera/src/test/io/seqera/executor/SeqeraTaskHandlerTest.groovy

@@ -1271,10 +1271,77 @@ class SeqeraTaskHandlerTest extends Specification {
         [:]                                  | null
     }
 
+    def 'submit disables the prediction model when the process references task.memory'() {
+        given:
+        Task captured = null
+        def handler = createSubmitHandler(
+            predictionModel: 'qr/v2',
+            memoryReferenced: true,
+            onSubmit: { captured = it },
+        )
+
+        when:
+        handler.submit()
+        then:
+        captured.getPredictionModel() == PredictionModel.NONE
+    }
+
+    def 'submit keeps the run-level prediction model when the process does not reference task.memory'() {
+        given:
+        Task captured = null
+        def handler = createSubmitHandler(
+            predictionModel: 'qr/v2',
+            memoryReferenced: false,
+            onSubmit: { captured = it },
+        )
+
+        when:
+        handler.submit()
+        then:
+        captured.getPredictionModel() == null
+    }
+
+    @Unroll
+    def 'submit does not disable the prediction model when the run-level model is #runModel'() {
+        given:
+        Task captured = null
+        def handler = createSubmitHandler(
+            predictionModel: runModel,
+            memoryReferenced: true,
+            onSubmit: { captured = it },
+        )
+
+        when:
+        handler.submit()
+        then:
+        captured.getPredictionModel() == null
+
+        where:
+        runModel << [null, '', 'none']
+    }
+
+    def 'submit lets an explicit predictionModel hint win over the task.memory check'() {
+        given:
+        Task captured = null
+        def handler = createSubmitHandler(
+            predictionModel: 'qr/v2',
+            memoryReferenced: true,
+            hints: ['seqera/predictionModel': 'qr/v1'],
+            onSubmit: { captured = it },
+        )
+
+        when:
+        handler.submit()
+        then:
+        captured.getPredictionModel() == PredictionModel.QR_V1
+    }
+
     private SeqeraTaskHandler createSubmitHandler(Map args) {
         final hints = args.hints as Map<String,Object> ?: [:]
         final baseMachineReq = args.baseMachineReq as MachineRequirementOpts
         final Closure onSubmit = args.onSubmit as Closure ?: {}
+        final memoryReferenced = args.memoryReferenced as boolean
+        final runPredictionModel = args.predictionModel as String
 
         def taskConfig = Mock(TaskConfig) {
             getCpus() >> 1
@@ -1289,6 +1356,7 @@ class SeqeraTaskHandlerTest extends Specification {
             getContainer() >> 'ubuntu:latest'
             getId() >> TaskId.of(1)
             getHash() >> HashCode.fromInt(1)
+            isDirectiveReferenced('memory') >> memoryReferenced
             lazyName() >> 'sample_task'
         }
         def executor = Mock(SeqeraExecutor) {
@@ -1298,6 +1366,7 @@ class SeqeraTaskHandlerTest extends Specification {
             }
             getSeqeraConfig() >> Mock(ExecutorOpts) {
                 getMachineRequirement() >> baseMachineReq
+                getPredictionModel() >> runPredictionModel
             }
             getRunResourceLabels() >> [:]
         }