Data lineage programmatic API

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

@@ -16,6 +16,9 @@
 
 package nextflow
 
+import nextflow.extension.LinChannelEx
+import nextflow.plugin.Plugins
+
 import static nextflow.util.CheckHelper.*
 
 import java.nio.file.FileSystem
@@ -657,4 +660,42 @@ class Channel  {
         fromPath0Future = future.exceptionally(Channel.&handlerException)
     }
 
-}
+    static DataflowWriteChannel fromLineage(String uri) {
+        final result = CH.create()
+        if( NF.isDsl2() ) {
+            session.addIgniter { fromLineage0(result, uri) }
+        }
+        else {
+            fromLineage0(result, uri )
+        }
+        return result
+    }
+
+    private static void fromLineage0(DataflowWriteChannel channel, String uri) {
+        final operation = Plugins.getExtension(LinChannelEx)
+        if( !operation )
+            throw new IllegalStateException("Unable to load lineage extensions.")
+        def future = CompletableFuture.runAsync( { operation.viewLineage(session, channel, new URI(uri)) } as Runnable)
+        future.exceptionally(this.&handlerException)
+    }
+
+    static DataflowWriteChannel queryLineage(String queryString) {
+        final result = CH.create()
+        if( NF.isDsl2() ) {
+            session.addIgniter { queryLineage0(result, queryString) }
+        }
+        else {
+            queryLineage0(result, queryString )
+        }
+        return result
+    }
+
+    private static void queryLineage0(DataflowWriteChannel channel, String query) {
+        final operation = Plugins.getExtension(LinChannelEx)
+        if( !operation )
+            throw new IllegalStateException("Unable to load lineage extensions.")
+        def future = CompletableFuture.runAsync( { operation.queryLineage(session, channel, query) } as Runnable)
+        future.exceptionally(this.&handlerException)
+    }
+
+}

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

@@ -381,7 +381,7 @@ class Session implements ISession {
         this.dag = new DAG()
 
         // -- init output dir
-        this.outputDir = FileHelper.toCanonicalPath(config.outputDir ?: 'results')
+        this.outputDir = FileHelper.toCanonicalPath(config.outputDir ?: config.navigate('params.outdir') ?: 'results')
 
         // -- init work dir
         this.workDir = FileHelper.toCanonicalPath(config.workDir ?: 'work')

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

@@ -0,0 +1,10 @@
+package nextflow.extension
+
+import groovyx.gpars.dataflow.DataflowWriteChannel
+import nextflow.Session
+
+interface LinChannelEx {
+    void viewLineage(Session session, DataflowWriteChannel channel, URI uri)
+
+    void queryLineage(Session session, DataflowWriteChannel channel, String query)
+}

modules/nextflow/src/main/groovy/nextflow/file/PathVisitor.groovy

@@ -66,7 +66,7 @@ class PathVisitor {
             applyRegexPattern0(filePattern)
 
         else if( filePattern != null )
-            applyGlobPattern0(filePattern as Path)
+            applyPathPattern0(filePattern as Path)
 
         else
             throw new IllegalArgumentException("Missing file pattern argument")
@@ -103,6 +103,27 @@ class PathVisitor {
             target.bind(STOP)
     }
 
+    private void applyPathPattern0(Path filePattern) {
+        if( isQuery(filePattern) )
+            applyQueryablePath0(filePattern as QueryablePath)
+        else
+            applyGlobPattern0(filePattern)
+    }
+
+    private static boolean isQuery(Path filePattern) {
+        log.debug("Checking if query: $filePattern.class ")
+        return filePattern instanceof QueryablePath && (filePattern as QueryablePath).hasQuery()
+    }
+
+    private boolean applyQueryablePath0(QueryablePath path) {
+        final paths = path.resolveQuery()
+        if( !paths )
+            throw new FileNotFoundException("No files found for ${path}")
+
+        paths.forEach { emit0(it) }
+        close0()
+    }
+
     private void applyGlobPattern0(Path filePattern) {
 
         final glob = opts?.containsKey('glob') ? opts.glob as boolean : true

modules/nextflow/src/main/groovy/nextflow/file/QueryablePath.groovy

@@ -0,0 +1,13 @@
+package nextflow.file
+
+import java.nio.file.Path
+
+/**
+ * Interface to indicate a Path could contain a query that is resolved to several real paths.
+ *
+ * @author Jorge Ejarque <jorge.ejarque@seqera.io>
+ */
+interface QueryablePath {
+    boolean hasQuery();
+    List<Path> resolveQuery();
+}

modules/nf-lineage/src/main/nextflow/lineage/LinChanneExImpl.groovy

@@ -0,0 +1,81 @@
+/*
+ * 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.lineage
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import groovyx.gpars.dataflow.DataflowWriteChannel
+import nextflow.Channel
+import nextflow.Session
+import nextflow.extension.LinChannelEx
+import nextflow.lineage.fs.LinPath
+import nextflow.lineage.fs.LinPathFactory
+import nextflow.lineage.serde.LinSerializable
+
+/**
+ * Lineage channel extensions
+ *
+ * @author Jorge Ejarque <jorge.ejarque@seqera.io>
+ */
+@CompileStatic
+@Slf4j
+class LinChanneExImpl implements LinChannelEx{
+
+    void viewLineage(Session session, DataflowWriteChannel channel, URI uri) {
+        final store = getStore(session)
+        emitResults(channel, LinUtils.query(store, uri))
+        channel.bind(Channel.STOP)
+    }
+
+    void queryLineage(Session session, DataflowWriteChannel channel, String query) {
+        final store = getStore(session)
+        emitSearchResults(channel, store.search(query))
+        channel.bind(Channel.STOP)
+    }
+
+
+    protected LinStore getStore(Session session){
+        final store = LinStoreFactory.getOrCreate(session)
+        if( !store ) {
+            throw new Exception("Lineage store not found - Check Nextflow configuration")
+        }
+        return store
+    }
+
+    private static void emitResults(DataflowWriteChannel channel, Collection results){
+        if( !results ) {
+            return
+        }
+        // Remove nested collections of a single element
+        if( results.size() == 1 ) {
+            final entry = results[0]
+            if( entry instanceof Collection ) {
+                emitResults(channel, entry)
+            } else {
+                channel.bind(LinUtils.encodeSearchOutputs(entry))
+            }
+        } else
+            results.forEach { channel.bind(LinUtils.encodeSearchOutputs(it)) }
+    }
+
+    private void emitSearchResults(DataflowWriteChannel channel, Map<String, LinSerializable> results) {
+        if( !results ) {
+            return
+        }
+        results.keySet().forEach { channel.bind(LinPathFactory.create(LinPath.LID_PROT + it)) }
+    }
+}

modules/nf-lineage/src/main/nextflow/lineage/LinObserver.groovy

@@ -87,6 +87,7 @@ class LinObserver implements TraceObserver {
     private Session session
     private WorkflowOutput workflowOutput
     private Map<String,String> outputsStoreDirLid = new HashMap<String,String>(10)
+    private Set<String> publishedFiles = new HashSet<String>()
     private PathNormalizer normalizer
 
     LinObserver(Session session, LinStore store){
@@ -124,6 +125,10 @@ class LinObserver implements TraceObserver {
     @Override
     void onFlowComplete(){
         if (this.workflowOutput){
+            //Add publishedFiles
+            for (String path: publishedFiles){
+                workflowOutput.output.add(new Parameter(Path.simpleName, null, path))
+            }
             workflowOutput.createdAt = OffsetDateTime.now()
             final key = executionHash + '#output'
             this.store.save(key, workflowOutput)
@@ -360,6 +365,7 @@ class LinObserver implements TraceObserver {
                 LinUtils.toDate(attrs?.lastModifiedTime()),
                 convertAnnotations(annotations))
             store.save(key, value)
+            publishedFiles.add(asUriString(key))
         } catch (Throwable e) {
             log.warn("Unexpected error storing published file '${destination.toUriString()}' for workflow '${executionHash}'", e)
         }
@@ -411,8 +417,9 @@ class LinObserver implements TraceObserver {
     private Object convertPathsToLidReferences(Object value){
         if( value instanceof Path ) {
             try {
-                final key = getWorkflowOutputKey(value)
-                return asUriString(key)
+                final key = asUriString(getWorkflowOutputKey(value))
+                publishedFiles.remove(key)
+                return key
             } catch (Throwable e){
                 //Workflow output key not found
                 return value

modules/nf-lineage/src/main/nextflow/lineage/LinUtils.groovy

@@ -316,7 +316,7 @@ class LinUtils {
      * @param output Output to encode
      * @return Output encoded as a JSON string
      */
-    static String encodeSearchOutputs(Object output, boolean prettyPrint) {
+    static String encodeSearchOutputs(Object output, boolean prettyPrint = false) {
         if (output instanceof LinSerializable) {
             return new LinEncoder().withPrettyPrint(prettyPrint).encode(output)
         } else {