Merge branch 'master' into feat/seqera-auto-labels-filter [ci fast]

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

Author: pditommaso

docs/reference/cli.md

@@ -815,7 +815,7 @@ The `launch` command launches a pipeline run in Seqera Platform. To log in and c
 : A mnemonic name to assign to the run.
 
 `-main-script`
-: The script file to be executed when launching a project directory or repository.
+: The script file to execute when launching a project directory or repository. Should be a path relative to the project root, e.g. `-main-script subproject/main.nf`.
 
 `-params-file`
 : A JSON or YAML file to load parameters from.
@@ -918,6 +918,11 @@ The `lint` command parses and analyzes the given Nextflow scripts and config fil
 : File pattern to exclude from linting (default: `.git, .lineage, .nextflow, .nf-test, nf-test.config, work`).
 : Can be specified multiple times.
 
+`-files-from`
+: :::{versionadded} 26.04.0
+  :::
+: Read list of paths to lint from a text file. Use `-` to read from standard input.
+
 `-format`
 : Format scripts and config files that have no errors.
 
@@ -1598,7 +1603,8 @@ The `run` command is used to execute a local pipeline script or remote pipeline
 : Library extension path.
 
 `-main-script` (`main.nf`)
-: The script file to be executed when launching a project directory or repository.
+: The script file to execute when launching a project directory or repository. Should be a path relative to the project root, e.g. `-main-script subproject/main.nf`.
+: The project `nextflow.config` is always read from the project root regardless of the main script location.
 
 `-name`
 : Assign a mnemonic name to the a pipeline run.

docs/reference/config.md

@@ -1418,9 +1418,6 @@ The following settings are available:
 `seqera.executor.autoLabels`
 : When `true`, automatically adds workflow metadata labels to the session with the `nextflow.io/` prefix (default: `false`). The following labels are added: `projectName`, `userName`, `runName`, `sessionId`, `resume`, `revision`, `commitId`, `repository`, `manifestName`, `runtimeVersion`. A `seqera.io/runId` label is also added, computed as a SipHash of the session ID and run name.
 
-`seqera.executor.labels`
-: Custom labels to apply to AWS resources for cost tracking and resource organization. Labels are propagated to ECS tasks, capacity providers, and EC2 instances. When used together with `autoLabels`, user-defined labels take precedence over auto-generated labels.
-
 `seqera.executor.machineRequirement.arch`
 : The CPU architecture for task execution, e.g. `'x86_64'` or `'arm64'`.
 

docs/reference/env-vars.md

@@ -123,6 +123,11 @@ The following environment variables control the configuration of the Nextflow ru
 : The file storage path against which relative file paths are resolved.
 : For example, with `NXF_FILE_ROOT=/some/root/path`, the use of `file('hello')` will be resolved to the absolute path `/some/root/path/hello`. A remote root path can be specified using the usual protocol prefix, e.g. `NXF_FILE_ROOT=s3://my-bucket/data`. Files defined using an absolute path are not affected by this setting.
 
+`NXF_FUSION_TRACE`
+: :::{versionadded} 26.04.0
+  :::
+: When set to `true`, collect task resource metrics (CPU, memory, I/O) from the Fusion trace file (`.fusion/trace.json`) produced in the task work directory, replacing the metrics collected by the default bash command-trace wrapper. Requires Fusion to be enabled. GPU metrics from Fusion are always collected regardless of this setting.
+
 `NXF_HOME`
 : Nextflow home directory (default: `$HOME/.nextflow`).
 

modules/nextflow/src/main/groovy/nextflow/cli/CmdLint.groovy

@@ -65,6 +65,12 @@ class CmdLint extends CmdBase {
     )
     List<String> excludePatterns = ['.git', '.lineage', '.nextflow', '.nf-test', 'nf-test.config', 'work']
 
+    @Parameter(
+        names = ['-files-from'],
+        description = 'Read list of paths to lint from a text file (one per line, use - for stdin)'
+    )
+    String filesFrom
+
     @Parameter(
         names = ['-o', '-output'],
         description = 'Output mode for reporting errors: full, extended, concise, json, markdown',
@@ -119,7 +125,10 @@ class CmdLint extends CmdBase {
 
     @Override
     void run() {
-        if( !args )
+        // read input files from positional args and -files-from option
+        final inputs = getInputs(args, filesFrom)
+
+        if( !inputs )
             throw new AbortOperationException("Error: No input files were specified")
 
         if( spaces && tabs )
@@ -134,21 +143,21 @@ class CmdLint extends CmdBase {
 
         scriptParser = new ScriptParser(baseDir, classLoader)
         configParser = new ConfigParser()
-        errorListener = outputMode == 'json'
-            ? new JsonErrorListener()
-            : outputMode == 'markdown'
-            ? new MarkdownErrorListener()
-            : new StandardErrorListener(outputMode, launcher.options.ansiLog)
+        errorListener = switch( outputMode ) {
+            case 'json' -> new JsonErrorListener()
+            case 'markdown' -> new MarkdownErrorListener()
+            default -> new StandardErrorListener(outputMode, launcher.options.ansiLog, launcher.options.quiet)
+        }
         formattingOptions = new FormattingOptions(spaces, !tabs, harhsilAlignment, false, sortDeclarations)
 
         errorListener.beforeAll()
 
         // collect files to lint
         final List<File> files = []
 
-        for( final arg : args ) {
+        for( final input : inputs ) {
             PathUtils.visitFiles(
-                Path.of(arg),
+                Path.of(input),
                 (path) -> !PathUtils.isExcluded(path, excludePatterns),
                 (path) -> files.add(path.toFile()))
         }
@@ -178,6 +187,24 @@ class CmdLint extends CmdBase {
             throw new AbortOperationException()
     }
 
+    private static List<String> getInputs(List<String> args, String filesFrom) {
+        final List<String> result = []
+        result.addAll(args)
+
+        if( filesFrom ) {
+            final lines = filesFrom == '-'
+                ? System.in.readLines()
+                : Path.of(filesFrom).readLines()
+            for( final line : lines ) {
+                final trimmed = line.trim()
+                if( trimmed )
+                    result.add(trimmed)
+            }
+        }
+
+        return result
+    }
+
     private void parse(File file) {
         final name = file.getName()
         if( name.endsWith('.nf') )

modules/nextflow/src/main/groovy/nextflow/cli/CmdRun.groovy

@@ -616,9 +616,7 @@ class CmdRun extends CmdBase implements HubOptions {
         /*
          * try to look for a pipeline in the repository
          */
-        try( final manager = new AssetManager(pipelineName, this) ) {
-            if( revision )
-                manager.setRevision(revision)
+        try( final manager = new AssetManager(pipelineName, revision, mainScript, this) ) {
             final repo = manager.getProjectWithRevision()
             final remoteSource = !manager.isLocalScmSource()
 

modules/nextflow/src/main/groovy/nextflow/executor/BashWrapperBuilder.groovy

@@ -26,6 +26,7 @@ import java.nio.file.Path
 import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
+import nextflow.Global
 import nextflow.SysEnv
 import nextflow.container.ContainerBuilder
 import nextflow.container.ContainerHelper
@@ -582,7 +583,9 @@ class BashWrapperBuilder {
     }
 
     protected boolean isTraceRequired() {
-        statsEnabled || fixOwnership()
+        if( fusionEnabled && Global.isFusionTraceEnabled() )
+            return fixOwnership()
+        return statsEnabled || fixOwnership()
     }
 
     protected String shellPath() {

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

@@ -73,6 +73,8 @@ class TaskBean implements Serializable, Cloneable {
 
     boolean statsEnabled
 
+    boolean fusionEnabled
+
     List<String> outputEnvNames
 
     Map<String,String> outputEvals
@@ -164,6 +166,7 @@ class TaskBean implements Serializable, Cloneable {
         this.outputEnvNames = task.getOutputEnvNames()
         this.outputEvals = task.getOutputEvals()
         this.statsEnabled = task.getProcessor().getSession().statsEnabled
+        this.fusionEnabled = task.getProcessor().isFusionEnabled()
 
         this.inputFiles = task.getInputFilesMap()
         this.outputFiles = task.getOutputFilesNames()

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

@@ -17,13 +17,16 @@
 package nextflow.processor
 
 import static nextflow.processor.TaskStatus.*
+import static nextflow.trace.TraceRecord.toLong
+import static nextflow.trace.TraceRecord.toFloat
 
 import java.nio.file.NoSuchFileException
 import java.util.concurrent.atomic.AtomicBoolean
 
 import groovy.transform.CompileDynamic
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import nextflow.Global
 import nextflow.trace.TraceRecord
 import nextflow.util.TestOnly
 /**
@@ -241,15 +244,21 @@ abstract class TaskHandler {
                 }
             }
 
-            final file = task.workDir?.resolve(TaskRun.CMD_TRACE)
-            try {
-                if(file) record.parseTraceFile(file)
-            }
-            catch( NoSuchFileException e ) {
-                // ignore it
-            }
-            catch( IOException e ) {
-                log.debug "[WARN] Cannot read trace file: $file -- Cause: ${e.message}"
+            // When Fusion trace is enabled, task metrics are collected by the Fusion client
+            // and reported in .fusion/trace.json — skip parsing the .command.trace file
+            // generated by the bash trace wrapper, as it won't exist
+            final fusionTraceEnabled = task.processor.executor.isFusionEnabled() && Global.isFusionTraceEnabled()
+            if( !fusionTraceEnabled ) {
+                final file = task.workDir?.resolve(TaskRun.CMD_TRACE)
+                try {
+                    if(file) record.parseTraceFile(file)
+                }
+                catch( NoSuchFileException e ) {
+                    log.trace "Unable to find trace file: $file"
+                }
+                catch( IOException e ) {
+                    log.debug "[WARN] Cannot read trace file: $file -- Cause: ${e.message}"
+                }
             }
 
             parseFusionTrace(record)
@@ -261,22 +270,82 @@ abstract class TaskHandler {
     protected void parseFusionTrace(TraceRecord record) {
         if( !task.processor.executor.isFusionEnabled() )
             return
+        // Resolve the Fusion trace file path (.fusion/trace.json) in the task work directory.
+        // This file is produced by the Fusion client and contains process, cgroup, and GPU metrics.
         final fusionTrace = task.workDir?.resolve(TaskRun.FUSION_TRACE)
         try {
-            if( fusionTrace ) {
-                final gpu = TraceRecord.parseFusionTraceFile(fusionTrace)
-                if( gpu )
-                    record.gpuMetrics = gpu
-            }
+            if( !fusionTrace )
+                return
+            // Parse the full Fusion trace JSON — sections: 'proc', 'cgroup', 'gpu'
+            final json = TraceRecord.parseFusionTraceFile(fusionTrace)
+            // GPU metrics are always extracted when available, regardless of the
+            // NXF_FUSION_TRACE setting — this preserves backward compatibility
+            final gpu = (Map<String,Object>) json.get('gpu')
+            if( gpu )
+                record.gpuMetrics = gpu
+            // When NXF_FUSION_TRACE is disabled, stop here — only GPU metrics are collected.
+            // When enabled, also map the 'proc' and 'cgroup' sections into TraceRecord fields,
+            // replacing the metrics that would normally come from the bash trace wrapper
+            if( !Global.isFusionTraceEnabled() )
+                return
+            applyFusionMetrics(record, json)
         }
         catch( NoSuchFileException e ) {
-            // ignore it
+            log.trace "Unable to find Fusion trace file: $fusionTrace"
         }
         catch( Exception e ) {
             log.debug "[WARN] Cannot read Fusion trace file: $fusionTrace -- Cause: ${e.message}"
         }
     }
 
+    protected void applyFusionMetrics(TraceRecord record, Map<String,Object> json) {
+        final proc = (Map<String,Object>) json.get('proc')
+        final cgroup = (Map<String,Object>) json.get('cgroup')
+
+        if( proc ) {
+            // CPU and timing metrics
+            record.store.put('realtime', toLong(proc.get('realtime')))
+            record.store.put('%cpu', toFloat(proc.get('pct_cpu')) / 10.0f as float)
+            record.store.put('cpu_model', proc.get('cpu_name')?.toString())
+            // I/O metrics
+            record.store.put('rchar', toLong(proc.get('rchar')))
+            record.store.put('wchar', toLong(proc.get('wchar')))
+            record.store.put('syscr', toLong(proc.get('syscr')))
+            record.store.put('syscw', toLong(proc.get('syscw')))
+            record.store.put('read_bytes', toLong(proc.get('read_bytes')))
+            record.store.put('write_bytes', toLong(proc.get('write_bytes')))
+            // context switches
+            record.store.put('vol_ctxt', toLong(proc.get('vol_ctxt')))
+            record.store.put('inv_ctxt', toLong(proc.get('inv_ctxt')))
+        }
+
+        // Prefer cgroup memory metrics (more accurate for containerized tasks)
+        if( cgroup ) {
+            record.store.put('vmem', toLong(cgroup.get('memory_current')))
+            record.store.put('rss', toLong(cgroup.get('memory_rss')))
+            record.store.put('peak_vmem', toLong(cgroup.get('memory_peak')))
+            record.store.put('peak_rss', toLong(cgroup.get('memory_peak_rss')))
+            // Compute %mem as peak RSS against the cgroup memory limit — i.e. how close the
+            // task got to its allocated memory. This overrides proc.pct_mem, which is RSS against
+            // total host memory (ps %MEM semantics) and underestimates utilization for containerized
+            // tasks whose limit is smaller than the host. Using peak (not current) RSS because Fusion
+            // overwrites memory_rss post-exit, making the last sample unrepresentative.
+            final memLimit = toLong(cgroup.get('memory_limit'))
+            final memPeakRss = toLong(cgroup.get('memory_peak_rss'))
+            if( memLimit > 0 ) {
+                record.store.put('%mem', (memPeakRss / (float)memLimit * 100.0f) as float)
+            }
+        }
+        else if( proc ) {
+            // Fallback to proc memory metrics when cgroup is not available
+            record.store.put('vmem', toLong(proc.get('vmem')))
+            record.store.put('rss', toLong(proc.get('rss')))
+            record.store.put('peak_vmem', toLong(proc.get('peak_vmem')))
+            record.store.put('peak_rss', toLong(proc.get('peak_rss')))
+            record.store.put('%mem', toFloat(proc.get('pct_mem')) / 10.0f as float)
+        }
+    }
+
     /**
      * Determine if a process can be forked i.e. can launch
      * a parallel task execution. This is only enforced when

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

@@ -331,6 +331,8 @@ class TaskProcessor {
      */
     Executor getExecutor() { executor }
 
+    boolean isFusionEnabled() { executor?.isFusionEnabled() ?: false }
+
     /**
      * @return The {@code DataflowOperator} underlying this process
      */

modules/nextflow/src/main/groovy/nextflow/scm/AssetManager.groovy

@@ -114,12 +114,12 @@ class AssetManager implements Closeable {
         build(pipelineName, config)
     }
 
-    AssetManager( String pipelineName, String revision, HubOptions cliOpts = null ) {
+    AssetManager( String pipelineName, String revision, String mainScript = null, HubOptions cliOpts = null ) {
         assert pipelineName
-        // build the object
+        // read the default config file (if available)
         def config = ProviderConfig.getDefault()
         // build the object
-        build(pipelineName, config, cliOpts, revision)
+        build(pipelineName, config, cliOpts, revision, mainScript)
     }
 
     /**
@@ -131,23 +131,24 @@ class AssetManager implements Closeable {
      * @return The {@link AssetManager} object itself
      */
     @PackageScope
-    AssetManager build( String pipelineName, Map config = null, HubOptions cliOpts = null, String revision = null ) {
+    AssetManager build( String pipelineName, Map config = null, HubOptions cliOpts = null, String revision = null, String mainScript = null ) {
 
         this.providerConfigs = ProviderConfig.createFromMap(config)
 
         this.project = resolveName(pipelineName)
+        if( mainScript )
+            this.mainScript = mainScript
+
+        if( !isValidProjectName(project) )
+            throw new IllegalArgumentException("Not a valid project name: ${project}")
 
-        if( !isValidProjectName(this.project) ) {
-            throw new IllegalArgumentException("Not a valid project name: ${this.project}")
-        }
         // Initialize strategy based on environment and repository state
         initStrategy(revision)
         this.hub = checkHubProvider(cliOpts)
         this.provider = createHubProvider(hub)
 
-        if( revision ){
+        if( revision )
             setRevision(revision)
-        }
 
         strategy.setProvider(this.provider)
 
@@ -414,10 +415,13 @@ class AssetManager implements Closeable {
 
         def parts = name.split('/') as List<String>
         def last = parts[-1]
-        if( last.endsWith('.nf') || last.endsWith('.nxf') ) {
+        if( last.endsWith('.nf') ) {
             if( parts.size()==1 )
                 throw new AbortOperationException("Not a valid project name: $name")
 
+            if( mainScript )
+                throw new AbortOperationException("Not a valid project name: $name -- Project name must be a directory when main script is provided")
+
             if( parts.size()==2 ) {
                 mainScript = last
                 parts = [ parts.first() ]
@@ -462,6 +466,9 @@ class AssetManager implements Closeable {
         if( !isUrl )
             return null
 
+        if( repository.endsWith('.nf') )
+            throw new AbortOperationException("Repository URL must not end with a script file extension (.nf) -- use `-main-script` to specify the relative script path")
+
         try {
             def url = new GitUrl(repository)