Data lineage tracking (aka CID store)

build.gradle

@@ -237,7 +237,7 @@ task compile {
 
 def getRuntimeConfigs() {
     def names = subprojects
-            .findAll { prj -> prj.name in ['nextflow','nf-commons','nf-httpfs','nf-lang'] }
+            .findAll { prj -> prj.name in ['nextflow','nf-commons','nf-httpfs','nf-lang','nf-lineage'] }
             .collect { it.name }
 
     FileCollection result = null
@@ -263,7 +263,7 @@ task exportClasspath {
         def home = System.getProperty('user.home')
         def all = getRuntimeConfigs()
         def libs = all.collect { File file ->  /*println file.canonicalPath.replace(home, '$HOME');*/ file.canonicalPath;  }
-        ['nextflow','nf-commons','nf-httpfs','nf-lang'].each {libs << file("modules/$it/build/libs/${it}-${version}.jar").canonicalPath }
+        ['nextflow','nf-commons','nf-httpfs','nf-lang','nf-lineage'].each {libs << file("modules/$it/build/libs/${it}-${version}.jar").canonicalPath }
         file('.launch.classpath').text = libs.unique().join(':')
     }
 }
@@ -276,7 +276,7 @@ ext.nexusEmail = project.findProperty('nexusEmail')
 // `signing.keyId` property needs to be defined in the `gradle.properties` file
 ext.enableSignArchives = project.findProperty('signing.keyId')
 
-ext.coreProjects = projects( ':nextflow', ':nf-commons', ':nf-httpfs', ':nf-lang' )
+ext.coreProjects = projects( ':nextflow', ':nf-commons', ':nf-httpfs', ':nf-lang', ':nf-lineage' )
 
 configure(coreProjects) {
     group = 'io.nextflow'

modules/nextflow/build.gradle

@@ -51,7 +51,7 @@ dependencies {
     api 'io.seqera:lib-trace:0.1.0'
 
     testImplementation 'org.subethamail:subethasmtp:3.1.7'
-
+    testImplementation (project(':nf-lineage'))
     // test configuration
     testFixturesApi ("org.apache.groovy:groovy-test:4.0.26") { exclude group: 'org.apache.groovy' }
     testFixturesApi ("org.objenesis:objenesis:3.4")

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

@@ -1137,11 +1137,11 @@ class Session implements ISession {
         }
     }
 
-    void notifyFilePublish(Path destination, Path source=null) {
+    void notifyFilePublish(Path destination, Path source, Map annotations) {
         def copy = new ArrayList<TraceObserver>(observers)
         for( TraceObserver observer : copy  ) {
             try {
-                observer.onFilePublish(destination, source)
+                observer.onFilePublish(destination, source, annotations)
             }
             catch( Exception e ) {
                 log.error "Failed to invoke observer on file publish: $observer", e

modules/nextflow/src/main/groovy/nextflow/cli/CmdLineage.groovy

@@ -0,0 +1,291 @@
+/*
+ * 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.cli
+
+import java.nio.file.Paths
+
+import com.beust.jcommander.Parameter
+import com.beust.jcommander.Parameters
+import groovy.transform.CompileStatic
+import nextflow.config.ConfigBuilder
+import nextflow.config.ConfigMap
+import nextflow.exception.AbortOperationException
+import nextflow.plugin.Plugins
+import org.pf4j.ExtensionPoint
+
+/**
+ * CID command line interface
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+@CompileStatic
+@Parameters(commandDescription = "Explore workflows lineage metadata", commandNames = ['li'])
+class CmdLineage extends CmdBase implements UsageAware {
+
+    private static final String NAME = 'lineage'
+
+    interface LinCommand extends ExtensionPoint {
+        void log(ConfigMap config)
+        void describe(ConfigMap config, List<String> args)
+        void render(ConfigMap config, List<String> args)
+        void diff(ConfigMap config, List<String> args)
+        void find(ConfigMap config, List<String> args)
+    }
+
+    interface SubCmd {
+        String getName()
+        String getDescription()
+        void apply(List<String> args)
+        void usage()
+    }
+
+    private List<SubCmd> commands = new ArrayList<>()
+
+    private LinCommand operation
+
+    private ConfigMap config
+
+    CmdLineage() {
+        commands << new CmdLog()
+        commands << new CmdDescribe()
+        commands << new CmdRender()
+        commands << new CmdDiff()
+        commands << new CmdFind()
+    }
+
+    @Parameter(hidden = true)
+    List<String> args
+
+    @Override
+    String getName() {
+        return NAME
+    }
+
+    @Override
+    void run() {
+        if( !args ) {
+            usage(List.of())
+            return
+        }
+        // setup the plugins system and load the secrets provider
+        Plugins.init()
+        // load the config
+        this.config = new ConfigBuilder()
+            .setOptions(launcher.options)
+            .setBaseDir(Paths.get('.'))
+            .build()
+        // init plugins
+        Plugins.load(config)
+        // load the command operations
+        this.operation = Plugins.getExtension(LinCommand)
+        if( !operation )
+            throw new IllegalStateException("Unable to load lineage extensions.")
+        // consume the first argument
+        getCmd(args).apply(args.drop(1))
+    }
+
+    /**
+     * Print the command usage help
+     */
+    void usage() {
+        usage(args)
+    }
+
+    /**
+     * Print the command usage help
+     *
+     * @param args The arguments as entered by the user
+     */
+    void usage(List<String> args) {
+        if( !args ) {
+            List<String> result = []
+            result << this.getClass().getAnnotation(Parameters).commandDescription()
+            result << "Usage: nextflow $NAME <sub-command> [options]".toString()
+            result << ''
+            result << 'Commands:'
+            int len = 0
+            commands.forEach {len = it.name.size() > len ? it.name.size() : len }
+            commands.sort(){it.name}.each { result << "  ${it.name.padRight(len)}\t${it.description}".toString()  }
+            result << ''
+            println result.join('\n').toString()
+        }
+        else {
+            def sub = commands.find { it.name == args[0] }
+            if( sub )
+                sub.usage()
+            else {
+                throw new AbortOperationException("Unknown $NAME sub-command: ${args[0]}")
+            }
+        }
+    }
+
+    protected SubCmd getCmd(List<String> args) {
+
+        def cmd = commands.find { it.name == args[0] }
+        if( cmd ) {
+            return cmd
+        }
+
+        def matches = commands.collect{ it.name }.closest(args[0])
+        def msg = "Unknown cloud sub-command: ${args[0]}"
+        if( matches )
+            msg += " -- Did you mean one of these?\n" + matches.collect { "  $it"}.join('\n')
+        throw new AbortOperationException(msg)
+    }
+
+    class CmdLog implements SubCmd {
+
+        @Override
+        String getName() {
+            return 'log'
+        }
+
+        @Override
+        String getDescription() {
+            return 'Print the lineage execution log'
+        }
+
+        @Override
+        void apply(List<String> args) {
+            if (args.size() != 0) {
+                println("ERROR: Incorrect number of parameters")
+                usage()
+                return
+            }
+            operation.log(config)
+        }
+
+        @Override
+        void usage() {
+            println description
+            println "Usage: nextflow $NAME $name"
+        }
+    }
+
+    class CmdDescribe implements SubCmd{
+
+        @Override
+        String getName() {
+            return 'describe'
+        }
+
+        @Override
+        String getDescription() {
+            return 'Print the description of a Lineage ID (lid)'
+        }
+
+        void apply(List<String> args) {
+            if (args.size() != 1) {
+                println("ERROR: Incorrect number of parameters")
+                usage()
+                return
+            }
+
+            operation.describe(config, args)
+        }
+
+        @Override
+        void usage() {
+            println description
+            println "Usage: nextflow $NAME $name <lid> "
+        }
+    }
+
+    class CmdRender implements SubCmd {
+
+        @Override
+        String getName() { 'render' }
+
+        @Override
+        String getDescription() {
+            return 'Render the lineage graph for a workflow output'
+        }
+
+        void apply(List<String> args) {
+            if (args.size() < 1 || args.size() > 2) {
+                println("ERROR: Incorrect number of parameters")
+                usage()
+                return
+            }
+
+            operation.render(config, args)
+        }
+
+        @Override
+        void usage() {
+            println description
+            println "Usage: nextflow $NAME $name <workflow output lid> [<html output file>]"
+        }
+
+    }
+
+    class CmdDiff implements SubCmd {
+
+        @Override
+        String getName() { 'diff' }
+
+        @Override
+        String getDescription() {
+            return 'Show differences between two lineage descriptions'
+        }
+
+        void apply(List<String> args) {
+            if (args.size() != 2) {
+                println("ERROR: Incorrect number of parameters")
+                usage()
+                return
+            }
+            operation.diff(config, args)
+        }
+
+        @Override
+        void usage() {
+            println description
+            println "Usage: nextflow $NAME $name <lid-1> <lid-2>"
+        }
+
+    }
+
+    class CmdFind implements SubCmd {
+
+        @Override
+        String getName() { 'find' }
+
+        @Override
+        String getDescription() {
+            return 'Find lineage metadata descriptions matching with a query'
+        }
+
+        void apply(List<String> args) {
+            if (args.size() != 1) {
+                println("ERROR: Incorrect number of parameters")
+                usage()
+                return
+            }
+            operation.find(config, args)
+        }
+
+        @Override
+        void usage() {
+            println description
+            println "Usage: nextflow $NAME $name <query>"
+        }
+
+    }
+
+}

modules/nextflow/src/main/groovy/nextflow/cli/CmdRun.groovy

@@ -354,7 +354,8 @@ class CmdRun extends CmdBase implements HubOptions {
         runner.session.disableJobsCancellation = getDisableJobsCancellation()
 
         final isTowerEnabled = config.navigate('tower.enabled') as Boolean
-        if( isTowerEnabled || log.isTraceEnabled() )
+        final isDataEnabled = config.navigate("workflow.lineage.enabled") as Boolean
+        if( isTowerEnabled || isDataEnabled || log.isTraceEnabled() )
             runner.session.resolvedConfig = ConfigBuilder.resolveConfig(scriptFile.parent, this)
         // note config files are collected during the build process
         // this line should be after `ConfigBuilder#build`

modules/nextflow/src/main/groovy/nextflow/cli/Launcher.groovy

@@ -107,7 +107,8 @@ class Launcher {
                 new CmdSelfUpdate(),
                 new CmdPlugin(),
                 new CmdInspect(),
-                new CmdLint()
+                new CmdLint(),
+                new CmdLineage()
         ]
 
         if(SecretsLoader.isEnabled())
@@ -120,13 +121,20 @@ class Launcher {
 
         options = new CliOptions()
         jcommander = new JCommander(options)
-        allCommands.each { cmd ->
+        for( CmdBase cmd : allCommands ) {
             cmd.launcher = this;
-            jcommander.addCommand(cmd.name, cmd)
+            jcommander.addCommand(cmd.name, cmd, aliases(cmd))
         }
         jcommander.setProgramName( APP_NAME )
     }
 
+    private static final String[] EMPTY = new String[0]
+
+    private static String[] aliases(CmdBase cmd) {
+        final aliases = cmd.getClass().getAnnotation(Parameters)?.commandNames()
+        return aliases ?: EMPTY
+    }
+
     /**
      * Create the Jcommander 'interpreter' and parse the command line arguments
      */

modules/nextflow/src/main/groovy/nextflow/dag/MermaidHtmlRenderer.groovy

@@ -33,7 +33,7 @@ class MermaidHtmlRenderer implements DagRenderer {
         file.text = template.replace('REPLACE_WITH_NETWORK_DATA', network)
     }
 
-    private String readTemplate() {
+    static String readTemplate() {
         final writer = new StringWriter()
         final res = MermaidHtmlRenderer.class.getResourceAsStream('mermaid.dag.template.html')
         int ch