Add versioned task hasher strategy

Refactor TaskHasher into a configurable strategy pattern with
versioned implementations (V1, V2) selectable via the NXF_TASK_HASH_VER
environment variable. V2 (current behavior) is the default.

- TaskHasher: now an interface defining compute(), getTaskGlobalVars(),
  getTaskBinEntries()
- TaskHasherV1: original hashing (Maps by values, CacheFunnel after
  SerializableMarker)
- TaskHasherV2: current hashing (Maps by unordered entrySet, CacheFunnel
  before Map)
- TaskHasherFactory: Version enum + create() factory method
- HashBuilder: version-aware with branching for Map/CacheFunnel handling
- Session: caches resolved hash strategy version at startup

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

Author: pditommaso

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

@@ -59,6 +59,7 @@ import nextflow.plugin.Plugins
 import nextflow.processor.ErrorStrategy
 import nextflow.processor.TaskFault
 import nextflow.processor.TaskHandler
+import nextflow.processor.TaskHasherFactory
 import nextflow.processor.TaskProcessor
 import nextflow.script.BaseScript
 import nextflow.script.ProcessFactory
@@ -179,6 +180,11 @@ class Session implements ISession {
      */
     String runName
 
+    /**
+     * The task hash strategy version
+     */
+    TaskHasherFactory.Version hashStrategy
+
     /**
      * Enable stub run mode
      */
@@ -385,6 +391,9 @@ class Session implements ISession {
         this.runName = config.runName ?: NameGenerator.next()
         log.debug "Run name: $runName"
 
+        // -- hash strategy
+        this.hashStrategy = TaskHasherFactory.Version.DEFAULT()
+
         // -- dry run
         this.stubRun = config.stubRun
 

modules/nextflow/src/main/groovy/nextflow/processor/TaskHasher.groovy

@@ -18,240 +18,18 @@ package nextflow.processor
 import java.nio.file.Path
 
 import com.google.common.hash.HashCode
-import groovy.json.JsonOutput
 import groovy.transform.CompileStatic
-import groovy.transform.Memoized
-import groovy.util.logging.Slf4j
-import nextflow.Session
-import nextflow.exception.UnexpectedException
-import nextflow.util.CacheHelper
 /**
- * Implement task hash computation
+ * Define the interface for task hash computation
  *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
-@Slf4j
 @CompileStatic
-class TaskHasher {
+interface TaskHasher {
 
-    private TaskRun task
+    HashCode compute()
 
-    private TaskProcessor processor
+    Map<String,Object> getTaskGlobalVars()
 
-    private Session session
-
-    public TaskHasher(TaskRun task) {
-        this.task = task
-        this.processor = task.processor
-        this.session = task.processor.session
-    }
-
-    public HashCode compute() {
-
-        final keys = new ArrayList<Object>()
-
-        // add session UUID
-        keys << session.uniqueId
-
-        // add fully-qualified process name
-        keys << task.processor.name
-
-        // add source code of `script:` or `exec:` block
-        //
-        // - this allows task script to reference directives like `task.cpus`
-        //   without invalidating the cache
-        //
-        // - references to local variables, global variables, and `task.ext`
-        //   are included separately
-        keys << task.source
-
-        // add container fingerprint if present
-        if( task.isContainerEnabled() )
-            keys << task.getContainerFingerprint()
-
-        // add the name and value of each task input
-        for( final entry : task.inputs ) {
-            keys.add( entry.key.name )
-            keys.add( entry.value )
-        }
-
-        // add eval output commands
-        final outEvals = task.getOutputEvals()
-        if( outEvals ) {
-            keys.add("eval_outputs")
-            keys.add(computeEvalOutputCommands(outEvals))
-        }
-
-        // add variables referenced in the task script but not declared as input/output
-        def vars = getTaskGlobalVars()
-        if( vars ) {
-            log.trace "Task: ${task.processor.name} > Adding script vars hash code: ${vars}"
-            keys.add(vars.entrySet())
-        }
-
-        // add bin scripts referenced in the task script
-        final binEntries = getTaskBinEntries(task.source)
-        if( binEntries ) {
-            log.trace "Task: ${task.processor.name} > Adding scripts on project bin path: ${-> binEntries.join('; ')}"
-            keys.addAll(binEntries)
-        }
-
-        // add environment modules (`module` directive)
-        final modules = task.getConfig().getModule()
-        if( modules ) {
-            keys.addAll(modules)
-        }
-
-        // add conda packages (`conda` directive)
-        final conda = task.getCondaEnv()
-        if( conda ) {
-            keys.add(conda)
-        }
-
-        // add spack packages (`spack` and `arch` directives)
-        final spack = task.getSpackEnv()
-        final arch = task.getConfig().getArchitecture()
-
-        if( spack ) {
-            keys.add(spack)
-
-            if( arch ) {
-                keys.add(arch)
-            }
-        }
-
-        // add stub run marker if enabled
-        if( session.stubRun && task.config.getStubBlock() ) {
-            keys.add('stub-run')
-        }
-
-        // compute task hash
-        final mode = task.processor.getConfig().getHashMode()
-        final hash = computeHash(keys, mode)
-
-        // log task hash entries if enabled
-        if( session.dumpHashes ) {
-            session.dumpHashes == 'json'
-                ? dumpHashEntriesJson(task, keys, mode, hash)
-                : dumpHashEntriesLegacy(task, keys, mode, hash)
-        }
-
-        return hash
-    }
-
-    /**
-     * Compute a deterministic string representation of eval output commands for cache hashing.
-     * This method creates a consistent hash key based on the semantic names and command values
-     * of eval outputs, ensuring cache invalidation when eval outputs change.
-     *
-     * @param outEvals Map of eval parameter names to their command strings
-     * @return A concatenated string of "name=command" pairs, sorted for deterministic hashing
-     */
-    protected static String computeEvalOutputCommands(Map<String, String> outEvals) {
-        // Assert precondition that outEvals should not be null or empty when this method is called
-        assert outEvals != null && !outEvals.isEmpty(), "Eval outputs should not be null or empty"
-
-        final result = new StringBuilder()
-
-        // Sort entries by key for deterministic ordering. This ensures that the same set of
-        // eval outputs always produces the same hash regardless of map iteration order,
-        // which is critical for cache consistency across different JVM runs.
-        // Without sorting, HashMap iteration order can vary between executions, leading to
-        // different cache keys for identical eval output configurations and causing
-        // unnecessary cache misses and task re-execution
-        final sortedEntries = outEvals.entrySet().sort { a, b -> a.key.compareTo(b.key) }
-
-        // Build content using for loop to concatenate "name=command" pairs.
-        // This creates a symmetric pattern with input parameter hashing where both
-        // the parameter name and its value contribute to the cache key
-        for( final entry : sortedEntries ) {
-            // Add newline separator between entries for readability in debug scenarios
-            if( result.length() > 0 ) {
-                result.append('\n')
-            }
-            // Format: "semantic_name=bash_command" - both name and command value are
-            // included because changing either should invalidate the task cache
-            result.append(entry.key).append('=').append(entry.value)
-        }
-
-        return result.toString()
-    }
-
-    /**
-     * Get the mapping of global variables that were referenced by
-     * the task script, excluding references to `task.ext`.
-     */
-    Map<String,Object> getTaskGlobalVars() {
-        final result = task.getGlobalVars(task.processor.getOwnerScript().getBinding())
-        final directives = getTaskExtensionDirectiveVars()
-        result.putAll(directives)
-        return result
-    }
-
-    protected Map<String,Object> getTaskExtensionDirectiveVars() {
-        final variableNames = task.getVariableNames()
-        final result = new HashMap(variableNames.size())
-        final taskConfig = task.config
-        for( final key : variableNames ) {
-            if( !key.startsWith('task.ext.') )
-                continue
-            final value = taskConfig.eval(key.substring(5))
-            result.put(key, value)
-        }
-
-        return result
-    }
-
-    /**
-     * This method scans the task command string looking for invocations of scripts
-     * defined in the project bin folder.
-     *
-     * @param script The task command string
-     * @return The list of paths of scripts in the project bin folder referenced in the task command
-     */
-    @Memoized
-    List<Path> getTaskBinEntries(String script) {
-        List<Path> result = []
-        final tokenizer = new StringTokenizer(script, " \t\n\r\f()[]{};&|<>`")
-        while( tokenizer.hasMoreTokens() ) {
-            final token = tokenizer.nextToken()
-            final path = session.binEntries.get(token)
-            if( path )
-                result.add(path)
-        }
-        return result
-    }
-
-    private String safeTaskName(TaskRun task) {
-        return task != null ? task.lazyName() : task.processor.name
-    }
-
-    private HashCode computeHash(List keys, CacheHelper.HashMode mode) {
-        try {
-            return CacheHelper.hasher(keys, mode).hash()
-        }
-        catch (Throwable e) {
-            final msg = "Something went wrong while creating task hash for process '${task.processor.name}' -- Offending keys: ${ keys.collect { k -> "\n - type=${k.getClass().getName()} value=$k" } }"
-            throw new UnexpectedException(msg,e)
-        }
-    }
-
-    private void dumpHashEntriesJson(TaskRun task, List entries, CacheHelper.HashMode mode, hash) {
-        final collector = (item) -> [
-            hash: CacheHelper.hasher(item, mode).hash().toString(),
-            type: item?.getClass()?.getName(),
-            value: item?.toString()
-        ]
-        final json = JsonOutput.toJson(entries.collect(collector))
-        log.info "[${safeTaskName(task)}] cache hash: ${hash}; mode: ${mode}; entries: ${JsonOutput.prettyPrint(json)}"
-    }
-
-    private void dumpHashEntriesLegacy(TaskRun task, List entries, CacheHelper.HashMode mode, hash) {
-        final buffer = new StringBuilder()
-        buffer.append("[${safeTaskName(task)}] cache hash: ${hash}; mode: $mode; entries: \n")
-        for( final entry : entries ) {
-            buffer.append( "  ${CacheHelper.hasher(entry, mode).hash()} [${entry?.getClass()?.getName()}] $entry \n")
-        }
-        log.info(buffer.toString())
-    }
+    List<Path> getTaskBinEntries(String script)
 }

modules/nextflow/src/main/groovy/nextflow/processor/TaskHasherFactory.groovy

@@ -0,0 +1,49 @@
+/*
+ * 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.processor
+
+import groovy.transform.CompileStatic
+import nextflow.SysEnv
+/**
+ * Factory for creating versioned {@link TaskHasher} instances.
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+@CompileStatic
+class TaskHasherFactory {
+
+    enum Version {
+        V1,
+        V2
+
+        static Version DEFAULT() {
+            final val = SysEnv.get('NXF_TASK_HASH_VER')
+            return val ? valueOf(val.toUpperCase()) : V2
+        }
+    }
+
+    static TaskHasher create(TaskRun task) {
+        final version = task.processor.session.hashStrategy
+        switch( version ) {
+            case Version.V1:
+                return new TaskHasherV1(task)
+            case Version.V2:
+                return new TaskHasherV2(task)
+            default:
+                throw new IllegalArgumentException("Unknown task hasher version: ${version}")
+        }
+    }
+}