Support mixing of legacy vs typed workflows

Signed-off-by: Ben Sherman <bentshermann@gmail.com>

Author: bentsherman

modules/nextflow/src/main/groovy/nextflow/dataflow/ValueImpl.groovy

@@ -24,6 +24,8 @@ import groovyx.gpars.dataflow.DataflowVariable
 import nextflow.Global
 import nextflow.Session
 import nextflow.dag.NodeMarker
+import nextflow.dataflow.ops.CrossOpV2
+import nextflow.exception.ScriptRuntimeException
 import nextflow.extension.CH
 import nextflow.extension.DataflowHelper
 
@@ -50,6 +52,21 @@ class ValueImpl {
         return source
     }
 
+    ValueImpl cross(Object other) {
+        DataflowVariable left = this.getSource()
+        DataflowVariable right
+        if( other instanceof ValueImpl ) {
+            right = other.getSource()
+        }
+        else {
+            throw new ScriptRuntimeException("Operator `cross` expected a dataflow value, but received: ${other} [${other.class.simpleName}]")
+        }
+
+        final target = (DataflowVariable) new CrossOpV2(left, right).apply()
+        NodeMarker.addOperatorNode("cross", [left, right], [target])
+        return new ValueImpl(target)
+    }
+
     ChannelImpl flatMap(Function<?,Iterable> transform = null) {
         final target = CH.create()
         final onNext = { value ->

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

@@ -42,6 +42,8 @@ abstract class BaseScript extends Script implements ExecutionContext {
 
     private ScriptMeta meta
 
+    private boolean typingEnabled
+
     private WorkflowDef entryFlow
 
     private OutputDef publisher
@@ -64,6 +66,10 @@ abstract class BaseScript extends Script implements ExecutionContext {
         session
     }
 
+    boolean isTypingEnabled() {
+        return typingEnabled
+    }
+
     /**
      * Holds the configuration object which will used to execution the user tasks
      */
@@ -97,6 +103,13 @@ abstract class BaseScript extends Script implements ExecutionContext {
         binding.setVariable( 'secrets', SecretsLoader.secretContext() )
     }
 
+    /**
+     * Enable static typing for the script.
+     */
+    protected void enableTyping() {
+        this.typingEnabled = true
+    }
+
     /**
      * Define a params block.
      *

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

@@ -0,0 +1,109 @@
+/*
+ * 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
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import groovyx.gpars.dataflow.DataflowVariable
+import groovyx.gpars.dataflow.DataflowWriteChannel
+import nextflow.dataflow.ChannelImpl
+import nextflow.dataflow.ValueImpl
+import nextflow.util.RecordMap
+/**
+ * Utility functions for converting between v1 and v2 dataflow types.
+ *
+ * When a script enables the `nextflow.preview.types` flag, its workflows
+ * are "typed workflows", otherwise they are "legacy workflows":
+ *
+ * - Typed workflows use v2 dataflow types: ChannelImpl, ValueImpl
+ * - Legacy workflows use v1 dataflow types: DataflowBroadcast, DataflowVariable
+ *
+ * While a given script must be either typed or legacy, typed workflows
+ * can be composed with legacy workflows and vise versa. In order to support
+ * this, the calling workflow must be able to convert between v1 and v2 dataflow
+ * types based on whether the callee is typed or legacy.
+ *
+ * @author Ben Sherman <bentshermann@gmail.com>
+ */
+@Slf4j
+@CompileStatic
+class DataflowTypeHelper {
+
+    /**
+     * Normalize a source value to the appropriate dataflow type
+     * based on the target context.
+     *
+     * If a v2 dataflow type (ChannelImpl or ValueImpl) is being
+     * passed to a legacy process or workflow, normalize the source
+     * value by unwrapping it into a DataflowWriteChannel.
+     *
+     * If a v1 dataflow type (DataflowWriteChannel) is being passed
+     * to a typed process or workflow, normalize the source value by
+     * wrapping it as a ChannelImpl or ValueImpl.
+     *
+     * @param source
+     * @param typingEnabled
+     */
+    static Object normalize(Object source, boolean typingEnabled) {
+        return typingEnabled
+            ? normalizeV2(source)
+            : normalizeV1(source)
+    }
+
+    /**
+     * Normalize a source value by converting v1 dataflow types
+     * (DataflowWriteChannel, ChannelOut) to v2 dataflow types
+     * (ChannelImpl, ValueImpl).
+     *
+     * @param source
+     */
+    static Object normalizeV2(Object source) {
+        if( source instanceof ChannelOut )
+            return normalizeMultiChannelV2(source)
+        if( source instanceof DataflowVariable )
+            return new ValueImpl(source)
+        if( source instanceof DataflowWriteChannel )
+            return new ChannelImpl(source)
+        return source
+    }
+
+    private static Object normalizeMultiChannelV2(ChannelOut source) {
+        final names = source.getNames()
+        if( source.size() == 1 && (names.isEmpty() || names.first() == '$out') )
+            return normalizeV2(source[0])
+        final result = new HashMap<String,Object>()
+        for( int i = 0; i < source.size(); i++ )
+            result.put(names[i], normalizeV2(source[i]))
+        return new RecordMap(result)
+    }
+
+    /**
+     * Normalize a source value by converting v2 dataflow types
+     * (ChannelImpl, ValueImpl) to v1 dataflow types (DataflowWriteChannel,
+     * ChannelOut).
+     *
+     * @param source
+     */
+    static Object normalizeV1(Object source) {
+        if( source instanceof ChannelImpl )
+            return source.getSource()
+        if( source instanceof ValueImpl )
+            return source.getSource()
+        return source
+    }
+
+}

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

@@ -20,12 +20,10 @@ import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowBroadcast
 import groovyx.gpars.dataflow.DataflowReadChannel
-import groovyx.gpars.dataflow.DataflowVariable
+import groovyx.gpars.dataflow.DataflowWriteChannel
 import nextflow.Const
 import nextflow.Global
 import nextflow.Session
-import nextflow.dataflow.ChannelImpl
-import nextflow.dataflow.ValueImpl
 import nextflow.exception.ScriptRuntimeException
 import nextflow.extension.CH
 import nextflow.processor.TaskProcessor
@@ -144,9 +142,8 @@ class ProcessDef extends BindableDef implements IterableDef, ChainableDef {
             output = runV1(args, processConfig)
 
         // invoke process with typed inputs/outputs
-        // NOTE: .out property is not accessible with typed dataflow
         else if( processConfig instanceof ProcessConfigV2 )
-            return runV2(args, processConfig)
+            output = runV2(args, processConfig)
 
         // return process output
         return output
@@ -205,7 +202,8 @@ class ProcessDef extends BindableDef implements IterableDef, ChainableDef {
         return output
     }
 
-    private Object runV2(Object[] args, ProcessConfigV2 config) {
+    private ChannelOut runV2(Object[] args0, ProcessConfigV2 config) {
+        final args = ChannelOut.spread(args0)
         final declaredInputs = config.getInputs()
         final declaredOutputs = config.getOutputs()
 
@@ -217,16 +215,19 @@ class ProcessDef extends BindableDef implements IterableDef, ChainableDef {
         for( int i = 0; i < declaredInputs.size(); i++ )
             declaredInputs[i].setChannel(createSourceChannel(args[i]))
 
-        // set output
+        // set outputs
+        final singleton = declaredInputs.isSingleton()
+
         final feedbackChannels = getFeedbackChannels()
         if( feedbackChannels && feedbackChannels.size() != declaredOutputs.size() )
             throw new ScriptRuntimeException("Process `$processName` inputs and outputs do not have the same cardinality - Feedback loop is not supported"  )
 
-        Object output = null
-        if( declaredOutputs.size() > 0 ) {
-            final singleton = declaredInputs.isSingleton()
-            output = feedbackChannels ? feedbackChannels[0] : CH.create(singleton)
-            declaredOutputs[0].setChannel(output)
+        final channels = new LinkedHashMap<String,DataflowWriteChannel>()
+        for( int i = 0; i < declaredOutputs.size(); i++ ) {
+            final param = declaredOutputs[i]
+            final ch = feedbackChannels ? feedbackChannels[i] : CH.create(singleton)
+            param.setChannel(ch)
+            channels.put(param.getName(), ch)
         }
 
         for( final topic : declaredOutputs.getTopics() ) {
@@ -237,18 +238,12 @@ class ProcessDef extends BindableDef implements IterableDef, ChainableDef {
         // start processor
         createTaskProcessor().run()
 
-        return \
-            output instanceof DataflowVariable ? new ValueImpl(output) :
-            output instanceof DataflowBroadcast ? new ChannelImpl(output) :
-            null
+        return new ChannelOut(channels)
     }
 
     private DataflowReadChannel createSourceChannel(Object value) {
-        if( value instanceof ChannelImpl )
-            return CH.getReadChannel(value.getSource())
-
-        if( value instanceof ValueImpl )
-            return value.getSource()
+        if( value instanceof DataflowReadChannel || value instanceof DataflowBroadcast )
+            return CH.getReadChannel(value)
 
         final result = CH.value()
         result.bind(value)

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

@@ -20,6 +20,7 @@ import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowWriteChannel
+import org.codehaus.groovy.runtime.InvokerHelper
 import nextflow.NF
 import nextflow.dataflow.ChannelImpl
 import nextflow.dataflow.ValueImpl
@@ -104,17 +105,26 @@ class WorkflowBinding extends Binding  {
             final component = getComponent0(name)
             if( component ) {
                 checkScope0(component)
-                return component.invoke_o(args)
+                return invoke0(component, args)
             }
 
             // check it's an operator name
-            if( NF.hasOperator(name) )
+            if( !owner?.isTypingEnabled() && NF.hasOperator(name) )
                 return OpCall.create(name, args)
         }
 
         throw new MissingMethodException(name,this.getClass())
     }
 
+    private Object invoke0(ComponentDef component, Object args0) {
+        final componentTyped = component instanceof WorkflowDef && component.getOwner().isTypingEnabled()
+        final args = InvokerHelper.asArray(args0).stream()
+            .map(arg -> DataflowTypeHelper.normalize(arg, componentTyped))
+            .toArray()
+        final result = component.invoke_a(args)
+        return DataflowTypeHelper.normalize(result, owner?.isTypingEnabled())
+    }
+
     @Override
     void setProperty(String name, Object value) {
         // when a script variable name matches a BaseScript attribute name
@@ -149,8 +159,11 @@ class WorkflowBinding extends Binding  {
             super.getVariable(name)
         }
         catch( MissingPropertyException e ) {
+            if( owner?.isTypingEnabled() )
+                throw e
+
             if( !meta )
-                 throw e
+                throw e
 
             def component = getComponent0(name)
             if( component )

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

@@ -20,8 +20,6 @@ import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowWriteChannel
-import nextflow.dataflow.ChannelImpl
-import nextflow.dataflow.ValueImpl
 import nextflow.exception.MissingProcessException
 import nextflow.exception.MissingValueException
 import nextflow.exception.ScriptRuntimeException
@@ -138,26 +136,19 @@ class WorkflowDef extends BindableDef implements ChainableDef, IterableDef, Exec
         }
     }
 
-    protected Object collectOutputs(List<String> emissions) {
+    protected ChannelOut collectOutputs(List<String> emissions) {
         // make sure feedback channel cardinality matches
         if( feedbackChannels && feedbackChannels.size() != emissions.size() )
             throw new ScriptRuntimeException("Workflow `$name` inputs and outputs do not have the same cardinality - Feedback loop is not supported"  )
 
-        final channels = new LinkedHashMap<String,?>(emissions.size())
-        boolean typedDataflow = false
+        final channels = new LinkedHashMap<String, DataflowWriteChannel>(emissions.size())
         for( int i=0; i<emissions.size(); i++ ) {
             final targetName = emissions[i]
             if( !binding.hasVariable(targetName) )
                 throw new MissingValueException("Missing workflow output parameter: $targetName")
-            final obj = binding.getVariable(targetName)
+            final obj = DataflowTypeHelper.normalizeV1(binding.getVariable(targetName))
 
-            if( obj instanceof ChannelImpl || obj instanceof ValueImpl ) {
-                // TODO: feedbackChannels
-                typedDataflow = true
-                channels.put(targetName, obj)
-            }
-
-            else if( CH.isChannel(obj) ) {
+            if( CH.isChannel(obj) ) {
                 channels.put(targetName, target(i, obj))
             }
 
@@ -177,9 +168,7 @@ class WorkflowDef extends BindableDef implements ChainableDef, IterableDef, Exec
                 channels.put(targetName, value)
             }
         }
-        return typedDataflow
-            ? (channels.size() == 1 && channels.containsKey('$out') ? channels['$out'] : channels)
-            : new ChannelOut(channels as Map<String,DataflowWriteChannel>)
+        return new ChannelOut(channels)
     }
 
     protected DataflowWriteChannel target(int index, Object output) {
@@ -219,10 +208,8 @@ class WorkflowDef extends BindableDef implements ChainableDef, IterableDef, Exec
         }
         else {
             // otherwise collect the outputs from the workflow binding
-            final out = collectOutputs(declaredOutputs)
-            if( out instanceof ChannelOut )
-                this.output = out
-            return out
+            output = collectOutputs(declaredOutputs)
+            return output
         }
     }