Full execution provenance resolution

modules/nextflow/src/main/groovy/nextflow/Channel.groovy

@@ -522,7 +522,7 @@ class Channel  {
         def groupChannel = isFlat ? new DataflowQueue<>() : CH.create()
 
         new GroupTupleOp(groupOpts, mapChannel)
-                .setTarget(groupChannel)
+                .withTarget(groupChannel)
                 .apply()
 
         // -- flat the group resulting tuples

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -16,7 +16,6 @@
 
 package nextflow
 
-
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
@@ -53,6 +52,7 @@ import nextflow.processor.ErrorStrategy
 import nextflow.processor.TaskFault
 import nextflow.processor.TaskHandler
 import nextflow.processor.TaskProcessor
+import nextflow.prov.Tracker
 import nextflow.script.BaseScript
 import nextflow.script.ProcessConfig
 import nextflow.script.ProcessFactory
@@ -857,6 +857,8 @@ class Session implements ISession {
 
     DAG getDag() { this.dag }
 
+    Tracker getProvenance() { provenance }
+
     ExecutorService getExecService() { execService }
 
     /**

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

@@ -85,10 +85,8 @@ class BashWrapperBuilder {
             log.warn "Invalid value for `NXF_DEBUG` variable: $str -- See http://www.nextflow.io/docs/latest/config.html#environment-variables"
         }
         BASH = Collections.unmodifiableList( level > 0 ? ['/bin/bash','-uex'] : ['/bin/bash','-ue'] )
-
     }
 
-
     @Delegate
     ScriptFileCopyStrategy copyStrategy
 
@@ -480,6 +478,7 @@ class BashWrapperBuilder {
     protected String getTaskMetadata() {
         final lines = new StringBuilder()
         lines << '### ---\n'
+        lines << "### id: '${bean.taskId}'\n"
         lines << "### name: '${bean.name}'\n"
         if( bean.arrayIndexName ) {
             lines << '### array:\n'
@@ -493,12 +492,18 @@ class BashWrapperBuilder {
         if( containerConfig?.isEnabled() )
             lines << "### container: '${bean.containerImage}'\n"
 
-        if( outputFiles.size() > 0 ) {
+        if( outputFiles ) {
             lines << '### outputs:\n'
             for( final output : bean.outputFiles )
                 lines << "### - '${output}'\n"
         }
 
+        if( bean.upstreamTasks ) {
+            lines << '### upstream-tasks:\n'
+            for( final it : bean.upstreamTasks )
+                lines << "### - '${it}'\n"
+        }
+
         lines << '### ...\n'
     }
 

modules/nextflow/src/main/groovy/nextflow/extension/BranchOp.groovy

@@ -20,7 +20,9 @@ import groovy.transform.CompileStatic
 import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowWriteChannel
 import groovyx.gpars.dataflow.expression.DataflowExpression
+import groovyx.gpars.dataflow.operator.DataflowProcessor
 import nextflow.Channel
+import nextflow.extension.op.Op
 import nextflow.script.ChannelOut
 import nextflow.script.TokenBranchChoice
 import nextflow.script.TokenBranchDef
@@ -50,20 +52,21 @@ class BranchOp {
 
     ChannelOut getOutput() { this.output }
 
-    protected void doNext(it) {
+    protected void doNext(DataflowProcessor dp, Object it) {
         TokenBranchChoice ret = switchDef.closure.call(it)
         if( ret ) {
-            targets[ret.choice].bind(ret.value)
+            Op.bind(dp, targets[ret.choice], ret.value)
         }
     }
 
-    protected void doComplete(nope) {
+    protected void doComplete(DataflowProcessor dp) {
         for( DataflowWriteChannel ch : targets.values() ) {
             if( ch instanceof DataflowExpression ) {
-                if( !ch.isBound()) ch.bind(Channel.STOP)
+                if( !ch.isBound() )
+                    Op.bind(dp, ch, Channel.STOP)
             }
             else {
-                ch.bind(Channel.STOP)
+                Op.bind(dp, ch, Channel.STOP)
             }
         }
     }
@@ -72,7 +75,10 @@ class BranchOp {
         def events = new HashMap<String,Closure>(2)
         events.put('onNext', this.&doNext)
         events.put('onComplete', this.&doComplete)
-        DataflowHelper.subscribeImpl(source, events)
+        new SubscribeOp()
+            .withSource(source)
+            .withEvents(events)
+            .apply()
         return this
     }
 

modules/nextflow/src/main/groovy/nextflow/extension/BufferOp.groovy

@@ -16,6 +16,9 @@
 
 package nextflow.extension
 
+import static nextflow.extension.DataflowHelper.*
+import static nextflow.util.CheckHelper.*
+
 import groovy.transform.CompileDynamic
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
@@ -28,10 +31,11 @@ import groovyx.gpars.dataflow.operator.PoisonPill
 import nextflow.Channel
 import nextflow.Global
 import nextflow.Session
+import nextflow.extension.op.ContextGrouping
+import nextflow.extension.op.Op
 import org.codehaus.groovy.runtime.callsite.BooleanReturningMethodInvoker
-import static nextflow.extension.DataflowHelper.newOperator
-import static nextflow.util.CheckHelper.checkParams
 /**
+ * Implements the "buffer" operator
  *
  * @author Paolo Di Tommaso <[email protected]>
  */
@@ -160,20 +164,16 @@ class BufferOp {
         final listener = new DataflowEventAdapter() {
 
             @Override
-            Object controlMessageArrived(final DataflowProcessor processor, final DataflowReadChannel<Object> channel, final int index, final Object message) {
+            Object controlMessageArrived(final DataflowProcessor dp, final DataflowReadChannel<Object> channel, final int index, final Object message) {
                 if( message instanceof PoisonPill && remainder && buffer.size() ) {
-                    target.bind(buffer)
+                    Op.bind(dp, target, buffer)
                 }
                 return message
             }
 
             @Override
-            void afterRun(DataflowProcessor processor, List<Object> messages) {
-                if( !stopOnFirst )
-                    return
-                if( remainder && buffer)
-                    target.bind(buffer)
-                target.bind(Channel.STOP)
+            void afterStop(DataflowProcessor dp) {
+                Op.bind(dp, target, Channel.STOP)
             }
 
             @Override
@@ -187,24 +187,36 @@ class BufferOp {
         // -- open frame flag
         boolean isOpen = startingCriteria == null
 
-        // -- the operator collecting the elements
-        newOperator( source, target, listener ) {
+        // -- op code
+        final code = {
             if( isOpen ) {
                 buffer << it
             }
             else if( startingCriteria.call(it) ) {
                 isOpen = true
                 buffer << it
             }
-
+            final dp = getDelegate() as DataflowProcessor
             if( closeCriteria.call(it) ) {
-                ((DataflowProcessor) getDelegate()).bindOutput(buffer);
+                Op.bind(dp, target, buffer)
                 buffer = []
                 // when a *startingCriteria* is defined, close the open frame flag
                 isOpen = (startingCriteria == null)
             }
-
+            if( stopOnFirst ) {
+                if( remainder && buffer )
+                    Op.bind(dp, target, buffer)
+                dp.terminate()
+            }
         }
+
+        // -- the operator collecting the elements
+        new Op()
+            .withInput(source)
+            .withListener(listener)
+            .withContext(new ContextGrouping())
+            .withCode(code)
+            .apply()
     }
 
 }

modules/nextflow/src/main/groovy/nextflow/extension/ChainOp.groovy

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2013-2024, 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.extension
+
+import groovy.transform.CompileStatic
+import groovyx.gpars.dataflow.DataflowReadChannel
+import groovyx.gpars.dataflow.DataflowWriteChannel
+import groovyx.gpars.dataflow.operator.ChainWithClosure
+import groovyx.gpars.dataflow.operator.DataflowEventListener
+import nextflow.extension.op.Op
+/**
+ * Implements the chain operator
+ * 
+ * @author Paolo Di Tommaso <[email protected]>
+ */
+@CompileStatic
+class ChainOp {
+
+    private DataflowReadChannel source
+    private DataflowWriteChannel target
+    private List<DataflowEventListener> listeners = List.of()
+    private Closure action
+
+    static ChainOp create() {
+        new ChainOp()
+    }
+
+    ChainOp withSource(DataflowReadChannel source) {
+        assert source
+        this.source = source
+        return this
+    }
+
+    ChainOp withTarget(DataflowWriteChannel target) {
+        assert target
+        this.target = target
+        return this
+    }
+
+    ChainOp withListener(DataflowEventListener listener) {
+        assert listener != null
+        this.listeners = List.of(listener)
+        return this
+    }
+
+    ChainOp withListeners(List<DataflowEventListener> listeners) {
+        assert listeners != null
+        this.listeners = listeners
+        return this
+    }
+
+    ChainOp withAction(Closure action) {
+        this.action = action
+        return this
+    }
+
+    DataflowWriteChannel apply() {
+        assert source
+        assert target
+        assert action
+
+        new Op()
+            .withInput(source)
+            .withOutput(target)
+            .withListeners(listeners)
+            .withCode(new ChainWithClosure(action))
+            .apply()
+
+        return target
+    }
+}

modules/nextflow/src/main/groovy/nextflow/extension/ChannelEx.groovy

@@ -16,23 +16,23 @@
 
 package nextflow.extension
 
+import static nextflow.util.LoggerHelper.*
+
 import groovy.transform.CompileDynamic
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import groovyx.gpars.agent.Agent
+import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowWriteChannel
-import nextflow.Channel
-import nextflow.NF
 import nextflow.dag.NodeMarker
 import nextflow.exception.ScriptRuntimeException
+import nextflow.prov.Tracker
 import nextflow.script.ChainableDef
 import nextflow.script.ChannelOut
 import nextflow.script.ComponentDef
 import nextflow.script.CompositeDef
 import nextflow.script.ExecutionStack
 import org.codehaus.groovy.runtime.InvokerHelper
-import static nextflow.util.LoggerHelper.fmtType
-
 /**
  * Implements dataflow channel extension methods
  *
@@ -207,4 +207,11 @@ class ChannelEx {
         left.add(right)
     }
 
+    static Object unwrap(DataflowReadChannel self) {
+        final result = self.getVal()
+        return result instanceof Tracker.Msg
+            ? result.value
+            : result
+    }
+
 }