Fix bugs with workflow output and lineage (#6254)

---------

Signed-off-by: Ben Sherman <[email protected]>
Signed-off-by: jorgee <[email protected]>
Co-authored-by: jorgee <[email protected]>

Author: bentsherman

docs/workflow.md

@@ -572,6 +572,10 @@ output {
 
 Each `>>` specifies a *source file* and *publish target*. The source file should be a file or collection of files, and the publish target should be a directory or file name. If the publish target ends with a slash, it is treated as the directory in which source files are published. Otherwise, it is treated as the target filename of a source file. Only files that are published with the `>>` operator are saved to the output directory.
 
+:::{note}
+Files that do not originate from the work directory are not published.
+:::
+
 ### Index files
 
 Each output can create an index file of the values that were published. An index file preserves the structure of channel values, including metadata, which is simpler than encoding this information with directories and file names. The index file can be a CSV (`.csv`), JSON (`.json`), or YAML (`.yml`, `.yaml`) file. The channel values should be files, lists, or maps.
@@ -628,6 +632,10 @@ This example will produce the following index file:
 "3"|"sample 3"|"results/fastq/3a.fastq"|"results/fastq/3b.fastq"
 ```
 
+:::{note}
+Files that do not originate from the work directory are not published, but are included in the index file.
+:::
+
 See [Output directives](#output-directives) for the list of available index directives.
 
 ### Output directives

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

@@ -269,6 +269,8 @@ class PublishOp {
                     return normalizePath(el, targetResolver)
                 if( el instanceof Collection<Path> )
                     return normalizePaths(el, targetResolver)
+                if( el instanceof Map )
+                    return normalizePaths(el, targetResolver)
                 return el
             }
         }
@@ -281,6 +283,8 @@ class PublishOp {
                         return Map.entry(k, normalizePath(v, targetResolver))
                     if( v instanceof Collection<Path> )
                         return Map.entry(k, normalizePaths(v, targetResolver))
+                    if( v instanceof Map )
+                        return Map.entry(k, normalizePaths(v, targetResolver))
                     return Map.entry(k, v)
                 }
         }

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

@@ -135,6 +135,44 @@ class OutputDslTest extends Specification {
         root?.deleteDir()
     }
 
+    def 'should preserve non-task output files in workflow output'() {
+        given:
+        def root = Files.createTempDirectory('test')
+        def outputDir = root.resolve('results')
+        def workDir = root.resolve('work')
+        def inputDir = root.resolve('inputs'); Files.createDirectories(inputDir)
+        def file1 = inputDir.resolve('file1.txt'); file1.text = 'Hello'
+        def file2 = inputDir.resolve('file2.txt'); file2.text = 'world'
+        def record = [id: '1', file1: file1, file2: file2]
+        and:
+        def session = Mock(Session) {
+            getOutputs() >> [:]
+            getConfig() >> [:]
+            getOutputDir() >> outputDir
+            getWorkDir() >> workDir
+        }
+        Global.session = session
+        and:
+        assignOutput(session, 'foo', [ record ])
+        and:
+        def dsl = new OutputDsl()
+        and:
+        SysEnv.push(NXF_FILE_ROOT: root.toString())
+
+        when:
+        dsl.declare('foo') {
+        }
+        dsl.apply(session)
+        await(dsl)
+        then:
+        0 * session.notifyFilePublish(_)
+        1 * session.notifyWorkflowOutput(new WorkflowOutputEvent('foo', [ record ], null))
+
+        cleanup:
+        SysEnv.pop()
+        root?.deleteDir()
+    }
+
     def 'should set publish options in output declaration' () {
         when:
         def dsl1 = new OutputDsl.DeclareDsl()

modules/nf-commons/src/main/nextflow/serde/gson/GsonEncoder.groovy

@@ -17,6 +17,7 @@
 package nextflow.serde.gson
 
 import java.lang.reflect.Type
+import java.nio.file.Path
 import java.time.Instant
 import java.time.OffsetDateTime
 
@@ -80,6 +81,7 @@ abstract class GsonEncoder<T> implements Encoder<T, String> {
         builder.registerTypeAdapter(Instant.class, new InstantAdapter())
         builder.registerTypeAdapter(OffsetDateTime.class, new OffsetDateTimeAdapter())
         builder.registerTypeAdapter(GStringImpl.class, new GStringSerializer())
+        builder.registerTypeHierarchyAdapter(Path.class, new PathAdapter())
         if( factory )
             builder.registerTypeAdapterFactory(factory)
         if( prettyPrint )

modules/nf-commons/src/main/nextflow/serde/gson/PathAdapter.groovy

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2013-2025, 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.serde.gson
+
+import com.google.gson.TypeAdapter
+import com.google.gson.stream.JsonReader
+import com.google.gson.stream.JsonWriter
+import nextflow.file.FileHelper
+
+import java.nio.file.Path
+
+class PathAdapter extends TypeAdapter<Path> {
+    @Override
+    void write(JsonWriter writer, Path value) throws IOException {
+        writer.value(value?.toUriString())
+    }
+
+    @Override
+    Path read(JsonReader reader) throws IOException {
+        if (reader.peek() == JsonToken.NULL) {
+            reader.nextNull()
+            return null
+        }
+        return FileHelper.asPath(reader.nextString())
+    }
+}

modules/nf-lineage/src/test/nextflow/lineage/serde/LinEncoderTest.groovy

@@ -16,6 +16,7 @@
 
 package nextflow.lineage.serde
 
+import java.nio.file.Path
 import java.time.OffsetDateTime
 
 import nextflow.lineage.model.v1beta1.Checksum
@@ -83,22 +84,25 @@ class LinEncoderTest extends Specification{
         result.params.get(0).name == "param1"
     }
 
-    def 'should encode and decode WorkflowResults'(){
+    def 'should encode and decode WorkflowOutputs'(){
         given:
         def encoder = new LinEncoder()
         and:
         def time = OffsetDateTime.now()
-        def wfResults = new WorkflowOutput(time, "lid://1234", [new Parameter("String", "a", "A"), new Parameter("String", "b", "B")])
+        def wfResults = new WorkflowOutput(time, "lid://1234", [
+            new Parameter("Collection", "a", [id: 'id', file: 'sample.txt' as Path]),
+            new Parameter("String", "b", "B")
+        ])
+
         when:
         def encoded = encoder.encode(wfResults)
         def object = encoder.decode(encoded)
-
         then:
         object instanceof WorkflowOutput
         def result = object as WorkflowOutput
         result.createdAt == time
         result.workflowRun == "lid://1234"
-        result.output == [new Parameter("String", "a", "A"), new Parameter("String", "b", "B")]
+        result.output == [new Parameter("Collection", "a", [id: 'id', file: 'sample.txt']), new Parameter("String", "b", "B")]
     }
 
     def 'should encode and decode TaskRun'() {
@@ -133,7 +137,7 @@ class LinEncoderTest extends Specification{
         result.binEntries.get(0).checksum.value == "78910"
     }
 
-    def 'should encode and decode TaskResults'(){
+    def 'should encode and decode TaskOutputs'(){
         given:
         def encoder = new LinEncoder()
         and: