Fix serialization of dataflow params

modules/nextflow/src/main/groovy/nextflow/script/ParamsDsl.groovy

@@ -48,7 +48,9 @@ class ParamsDsl {
     Map<String,Param> getDeclarations() { declarations }
 
     void apply(Session session) {
-        final params = ParamsHelper.resolveParams(declarations.values(), session.cliParams ?: [:], session.configParams ?: [:])
+        final cliParams = session.cliParams ?: [:]
+        final configParams = session.configParams ?: [:]
+        final params = ParamsHelper.resolveParams(declarations.values(), cliParams, configParams)
 
         // propagate resolved params to all scripts for legacy compatibility
         if( !session.binding.getScriptPath() )
@@ -59,6 +61,8 @@ class ParamsDsl {
             final script = ScriptMeta.getScriptByPath(scriptPath)
             script.binding.setParams(params, true)
         }
+
+        session.getParams().setPlainValues(ParamsHelper.resolvePlainParams(declarations.values(), cliParams, configParams))
     }
 
 }

modules/nextflow/src/main/groovy/nextflow/script/ParamsHelper.groovy

@@ -70,12 +70,44 @@ class ParamsHelper {
                 throw new ScriptRuntimeException("Parameter `${name}` was specified on the command line or params file but is not declared in the script or config")
         }
 
-        final given = cliParams.subMap(names) + configParams.subMap(names)
+        final given = givenParams(names, cliParams, configParams)
         return resolveParams(declarations, given, '') { Param decl, Object value ->
             resolveParam(decl, value, cliParams.containsKey(decl.name))
         }
     }
 
+    /**
+     * Resolve declared params from the command line and config to plain
+     * values, which can be serialized before the dataflow network has
+     * started (e.g. for lineage or Seqera Platform).
+     *
+     * Each param is resolved as in {@link #resolveParams(Collection,Map,Map)},
+     * except that a {@code Channel<E>} param is resolved to its samplesheet
+     * and a {@code Value<V>} param to its value of type {@code V}, instead
+     * of a dataflow value. The params are assumed to be valid, i.e. already
+     * resolved by {@link #resolveParams(Collection,Map,Map)}.
+     *
+     * @param declarations
+     * @param cliParams
+     * @param configParams
+     */
+    static Map<String,Object> resolvePlainParams(Collection<Param> declarations, Map cliParams, Map configParams) {
+        final given = givenParams(declarations*.name as Set<String>, cliParams, configParams)
+        final result = new LinkedHashMap<String,Object>(declarations.size())
+        for( final decl : declarations ) {
+            final name = decl.name
+            final value = given.containsKey(name)
+                ? resolveParam(decl, given.get(name), cliParams.containsKey(name), true)
+                : resolveDefault(decl, true)
+            result.put(name, value)
+        }
+        return result
+    }
+
+    private static Map<String,?> givenParams(Set<String> names, Map cliParams, Map configParams) {
+        return cliParams.subMap(names) + configParams.subMap(names)
+    }
+
     /**
      * Resolve declared params against the given values. A param
      * with no given value is given its default value.
@@ -154,21 +186,26 @@ class ParamsHelper {
      * @param value
      * @param fromCli whether the value came from the command line (and is
      *                therefore a string that may need to be parsed)
+     * @param plain whether to give the plain value of a {@code Channel<E>}
+     *              or {@code Value<V>} param instead of a dataflow value
+     *              (see {@link #resolvePlainParams})
      */
-    static Object resolveParam(Param decl, Object value, boolean fromCli) {
+    static Object resolveParam(Param decl, Object value, boolean fromCli, boolean plain=false) {
         if( value == null )
             return null
 
         final rawType = TypeHelper.getRawType(decl.type)
 
         if( rawType == Channel )
-            return ChannelNamespace.fromList(loadChannelInput(decl, value))
+            return plain ? value : ChannelNamespace.fromList(loadChannelInput(decl, value))
 
-        if( rawType == Value )
-            return ChannelNamespace.value(resolveParam(elementDecl(decl), value, fromCli))
+        if( rawType == Value ) {
+            final result = resolveParam(elementDecl(decl), value, fromCli, plain)
+            return plain ? result : ChannelNamespace.value(result)
+        }
 
         if( TypeHelper.isRecordType(decl.type) && value instanceof Map )
-            return resolveRecord(decl, (Map)value, fromCli)
+            return resolveRecord(decl, (Map)value, fromCli, plain)
 
         final result = fromCli
             ? resolveFromCli(decl, value)
@@ -177,7 +214,7 @@ class ParamsHelper {
         return result
     }
 
-    private static RecordMap resolveRecord(Param decl, Map value, boolean fromCli) {
+    private static RecordMap resolveRecord(Param decl, Map value, boolean fromCli, boolean plain) {
         final type = (Class)decl.type
         final result = new LinkedHashMap<String,Object>(value)
         for( final field : type.getDeclaredFields() ) {
@@ -192,7 +229,7 @@ class ParamsHelper {
                 continue
             }
             final fieldDecl = new Param("${decl.name}.${name}", field.getGenericType(), optional, null)
-            result.put(name, resolveParam(fieldDecl, fieldValue, fromCli))
+            result.put(name, resolveParam(fieldDecl, fieldValue, fromCli, plain))
         }
         return new RecordMap(result)
     }
@@ -450,10 +487,11 @@ class ParamsHelper {
      * the pipeline is called.
      *
      * @param decl
+     * @param plain see {@link #resolveParam}
      */
-    static Object resolveDefault(Param decl) {
+    static Object resolveDefault(Param decl, boolean plain=false) {
         if( decl.defaultValue != null )
-            return resolveParam(decl, decl.defaultValue, false)
+            return resolveParam(decl, decl.defaultValue, false, plain)
         final type = TypeHelper.getRawType(decl.type)
         return type.isAnnotationPresent(PipelineParams)
             ? new RecordMap([:])

modules/nextflow/src/main/groovy/nextflow/script/ScriptBinding.groovy

@@ -216,6 +216,8 @@ class ScriptBinding extends WorkflowBinding {
 
         private List<String> scriptAssignment = []
 
+        private Map<String,Object> plainValues
+
         @Delegate
         private Map<String,Object> target
 
@@ -239,6 +241,26 @@ class ScriptBinding extends WorkflowBinding {
             return new ParamsMap(this, overrides)
         }
 
+        /**
+         * Set the plain values of the params declared in the params
+         * block (see {@link ParamsHelper#resolvePlainParams}).
+         *
+         * @param values
+         */
+        void setPlainValues(Map<String,Object> values) {
+            plainValues = values
+        }
+
+        /**
+         * Get the params with each dataflow value of a declared param
+         * (e.g. a {@code Channel<E>} param, or a field of a record param)
+         * replaced by the plain value it was resolved from, so that the
+         * params can be serialized before the dataflow network has started.
+         */
+        Map<String,Object> toPlainMap() {
+            return plainValues ? target + plainValues : this
+        }
+
         @Override
         Object get(Object key) {
             if( !target.containsKey(key) ) {

modules/nextflow/src/test/groovy/nextflow/script/ParamsDslTest.groovy

@@ -19,7 +19,11 @@ package nextflow.script
 import java.nio.file.Files
 import java.nio.file.Path
 
+import groovyx.gpars.dataflow.DataflowBroadcast
+import groovyx.gpars.dataflow.DataflowVariable
 import nextflow.Session
+import nextflow.dataflow.ChannelImpl
+import nextflow.dataflow.ValueImpl
 import nextflow.exception.AbortOperationException
 import nextflow.exception.ScriptRuntimeException
 import nextflow.file.FileHelper
@@ -298,6 +302,170 @@ class ParamsDslTest extends Specification {
         samplesheet?.delete()
     }
 
+    def 'should give dataflow params as plain values'() {
+        given:
+        def samplesheet = Files.createTempFile('test', '.csv')
+        samplesheet.text = 'id,count\na,1\nb,2\n'
+        def cliParams = [samples: samplesheet.toString(), limit: '5']
+        def configParams = [outdir: 'results'] + cliParams
+
+        when:
+        def params = runScript(
+            '''\
+            nextflow.enable.types = true
+
+            params {
+                samples: Channel<Sample>
+                limit: Value<Integer>
+                factor: Value<Integer> = 3
+                label: String = 'demo'
+            }
+
+            record Sample {
+                id: String
+                count: Integer
+            }
+
+            workflow { params }
+            ''',
+            config: [params: configParams],
+            params: cliParams,
+            configParams: configParams
+        )
+        then:
+        params.samples instanceof ChannelImpl
+        ((ChannelImpl)params.samples).getSource() instanceof DataflowBroadcast
+        params.limit instanceof ValueImpl
+        ((ValueImpl)params.limit).getSource() instanceof DataflowVariable
+        and:
+        params.toPlainMap() == [outdir: 'results', samples: samplesheet.toString(), limit: 5, factor: 3, label: 'demo']
+
+        cleanup:
+        samplesheet?.delete()
+    }
+
+    def 'should give a channel param set to a path in the config as that path'() {
+        given:
+        def samplesheet = Files.createTempFile('test', '.csv')
+        samplesheet.text = 'id\na\n'
+        def configParams = [samples: samplesheet]
+
+        when:
+        def params = runScript(
+            '''\
+            nextflow.enable.types = true
+
+            params {
+                samples: Channel<Sample>
+            }
+
+            record Sample {
+                id: String
+            }
+
+            workflow { params }
+            ''',
+            config: [params: configParams],
+            configParams: configParams
+        )
+        then:
+        params.samples instanceof ChannelImpl
+        params.toPlainMap().samples.is(samplesheet)
+
+        cleanup:
+        samplesheet?.delete()
+    }
+
+    def 'should keep the non-dataflow fields of a record param'() {
+        given:
+        def samplesheet = Files.createTempFile('test', '.csv')
+        samplesheet.text = 'id\na\n'
+        def reference = Files.createTempFile('test', '.fa')
+        def cliParams = [inputs: [samples: samplesheet.toString(), reference: reference.toString()]]
+
+        when:
+        def params = runScript(
+            '''\
+            nextflow.enable.types = true
+
+            params {
+                inputs: Inputs
+            }
+
+            record Inputs {
+                samples: Channel<Sample>
+                reference: Path
+            }
+
+            record Sample {
+                id: String
+            }
+
+            workflow { params }
+            ''',
+            config: [params: cliParams],
+            params: cliParams,
+            configParams: cliParams
+        )
+        then:
+        def plain = params.toPlainMap()
+        params.inputs.samples instanceof ChannelImpl
+        plain.inputs.samples == samplesheet.toString()
+        plain.inputs.reference == params.inputs.reference
+
+        cleanup:
+        samplesheet?.delete()
+        reference?.delete()
+    }
+
+    def 'should give non-dataflow params unchanged as plain values'() {
+        given:
+        def inputFile = Files.createTempFile('test', '.csv')
+        def cliParams = [input: inputFile.toString(), chunk_size: '3', sample: [id: 'a', greeting: 'hola']]
+        def configParams = [outdir: 'results'] + cliParams
+
+        when:
+        def params = runScript(
+            '''\
+            params {
+                input: Path
+                chunk_size: Integer = 1
+                save_intermeds: Boolean
+                sample: Sample
+            }
+
+            record Sample {
+                id: String
+                greeting: String
+            }
+
+            workflow { params }
+            ''',
+            config: [params: configParams],
+            params: cliParams,
+            configParams: configParams
+        )
+        then:
+        def plain = params.toPlainMap()
+        plain == params
+
+        cleanup:
+        inputFile?.delete()
+    }
+
+    def 'should give the params as plain values without a params block'() {
+        when:
+        def params = runScript(
+            '''\
+            params.input = 'samples.csv'
+
+            workflow { params }
+            '''
+        )
+        then:
+        params.toPlainMap().is(params)
+    }
+
     def 'should validate record param from nested map'() {
         when: 'a script is invoked as `nextflow run module.nf --sample.id a --sample.greeting hola`'
         def result = runScript(