diff --git a/modules/nextflow/src/main/groovy/nextflow/executor/Executor.groovy b/modules/nextflow/src/main/groovy/nextflow/executor/Executor.groovy
index faae075739..f71c7f3285 100644
--- a/modules/nextflow/src/main/groovy/nextflow/executor/Executor.groovy
+++ b/modules/nextflow/src/main/groovy/nextflow/executor/Executor.groovy
@@ -170,6 +170,19 @@ abstract class Executor {
         return false
     }
 
+    /**
+     * Task-aware variant of {@link #isContainerNative()}.
+     * Executors that support per-task driver selection (e.g. mixed docker + HPC)
+     * can override this to return different values per task.
+     * Defaults to the task-agnostic {@link #isContainerNative()}.
+     *
+     * @param task The task being evaluated
+     * @return {@code true} when the executor manages containers natively for this specific task
+     */
+    boolean isContainerNative(TaskRun task) {
+        return isContainerNative()
+    }
+
     /**
      * Determines which container engine settings in the nextflow config file
      * will be used by this executor e.g. {@code 'docker'}, {@code 'singularity'}, etc.
@@ -184,6 +197,19 @@ abstract class Executor {
         return null
     }
 
+    /**
+     * Task-aware variant of {@link #containerConfigEngine()}.
+     * Executors that support per-task driver selection can override this
+     * to return different container engine configs per task.
+     * Defaults to the task-agnostic {@link #containerConfigEngine()}.
+     *
+     * @param task The task being evaluated
+     * @return The container engine name for this specific task, or null
+     */
+    String containerConfigEngine(TaskRun task) {
+        return containerConfigEngine()
+    }
+
     /**
      * @return {@code true} whenever the secrets handling is managed by the executing platform itself
      */
diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy
index 821c6b3249..55990ca3f6 100644
--- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy
+++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy
@@ -767,13 +767,14 @@ class TaskRun implements Cloneable {
         // get the container engine expected to be used by this executor
         final sess = this.getProcessor().getSession()
         final exe = this.getProcessor().getExecutor()
-        final eng = exe.containerConfigEngine()
+        // use task-aware methods to allow per-task driver resolution
+        final eng = exe.containerConfigEngine(this)
         // when 'eng' is null the setting for the current engine marked as 'enabled' will be used
         final result
                 = sess.getContainerConfig(eng)
                 ?: new DockerConfig([:])
         // if a configuration is found is expected to enabled by default
-        if( exe.isContainerNative() ) {
+        if( exe.isContainerNative(this) ) {
             result.setEnabled(true)
         }
         return result
@@ -788,7 +789,7 @@ class TaskRun implements Cloneable {
     }
 
     boolean isContainerNative() {
-        return processor.executor?.isContainerNative() ?: false
+        return processor.executor?.isContainerNative(this) ?: false
     }
 
     boolean isArray() {
diff --git a/modules/nextflow/src/test/groovy/nextflow/processor/TaskRunTest.groovy b/modules/nextflow/src/test/groovy/nextflow/processor/TaskRunTest.groovy
index e3579642e1..764301e95c 100644
--- a/modules/nextflow/src/test/groovy/nextflow/processor/TaskRunTest.groovy
+++ b/modules/nextflow/src/test/groovy/nextflow/processor/TaskRunTest.groovy
@@ -575,7 +575,7 @@ class TaskRunTest extends Specification {
         isNative = task.isContainerNative()
         then:
         1 * task.processor.getExecutor() >> executor
-        1 * executor.isContainerNative() >> true
+        1 * executor.isContainerNative(task) >> true
         isNative
     }
 
@@ -843,8 +843,8 @@ class TaskRunTest extends Specification {
         when:
         def config = task.getContainerConfig()
         then:
-        1 * executor.containerConfigEngine() >> null
-        1 * executor.isContainerNative() >> false
+        1 * executor.containerConfigEngine(task) >> null
+        1 * executor.isContainerNative(task) >> false
         and:
         session.getContainerConfig(null) >> null
         and:
@@ -853,8 +853,8 @@ class TaskRunTest extends Specification {
         when:
         config = task.getContainerConfig()
         then:
-        1 * executor.containerConfigEngine() >> null
-        1 * executor.isContainerNative() >> false
+        1 * executor.containerConfigEngine(task) >> null
+        1 * executor.isContainerNative(task) >> false
         and:
         session.getContainerConfig(null) >> new PodmanConfig(registry:'xyz')
         and: