Repository: populse/capsul (language: python)

Issue:
Test things
Most of the tests are currently broken due to the heavy works on the new API and implementation.
A few tests have been repaired, then broken again several times, because as the general tests do not pass, we don't even run them and look at the few ones that should pass.
We should disable all tests which are not ready to pass, and activate the ones which should, then run them (or at least look at appveyor messages and fix accordingly), so that we know when we actually break things that should work, and re-enable tests as we fix them.
Test things
Most of the tests are currently broken due to the heavy works on the new API and implementation.
A few tests have been repaired, then broken again several times, because as the general tests do not pass, we don't even run them and look at the few ones that should pass.
We should disable all tests which are not ready to pass, and activate the ones which should, then run them (or at least look at appveyor messages and fix accordingly), so that we know when we actually break things that should work, and re-enable tests as we fix them.
Run Capsul v3 pipelines on three engines: builtin, remote station (ssh) and cluster (Torque/PBS)

Run Capsul v3 pipelines on three engines: builtin, remote station (ssh) and cluster (Torque/PBS)



Discussion / hints:
I'm starting with disabling all failing tests.
I would need some help to adapt `pipeline.test.test_custom_nodes` to new API. This is a rather big test case and it would save me time to discuss with @denisri (its author) about what it is supposed to do.
I will have a look at it, and we may talk (maybe this afternoon but I'm trying to finish something else this morning)
There is no rush. I wrote that not to forget. We will also have to work together on issue #228 I just created. We may have to plan one or several hands on meeting with some precise topics to work on.
The thing is that most of the custom nodes were designed to work with parameters values on-the-fly, on client side. For instance to separate a list of files in a train sublist and a test sublist for cross-validation pipelines. But now values are not propagated through pipeline nodes any longer, thus they don't get their inputs and don't produce their outputs correctly, I think - at least not in "real time" when debugging a pipeline in a graphical editor. So I don't know how to proceed with them...
How do we activate parameters values propagation through links ? Is it only using the completion system ?
There is almost no relation between the links and the path generation system. We can consider completion system as independent.
There is a value propagation triggered mainly when values are set on the pipeline itself (redefinition of `__setattr__`), but also internally when some internal values are changed (for instance for instance when temporary file names are set or, during execution, when a process change its output values). Propagation is done explicitly using `Pipeline`'s `dispatch_value` method. If something is not working it may indicate a bug in `dispatch_value` or a misuse of it internally. But it may also mean that the system is different (no parameter modification inside a pipeline is allowed) and we should rethink the example. Up to now, I had always been able to adapt tests without much change. We should put this test case on the list of things we should do together in order to see if a simple test adaptation is enough or if some changes in the API rules and internal machinery are required.
Well, then there should be a problem with values propagation. I run in an `ipython3 --gui=qt` shell the following code:
```
from capsul.api import executable
from capsul.qt_gui.widgets import PipelineDeveloperView
import os
pipeline = executable('capsul.pipeline.test.test_custom_nodes.Pipeline1')
pipeline.main_inputs = [os.path.join('/tmp', 'file%d' % i)
        for i in range(4)]
pipeline.test = pipeline.main_inputs[2]
pipeline.subject = 'subject2'
pipeline.output_directory = os.path.join('/tmp', 'out_dir')
view1 = PipelineDeveloperView(pipeline, allow_open_controller=True,
                                       show_sub_pipelines=True,
                                       enable_edition=True)
view1.show()
```
The input values, for instace for the pipeline parameter `main_inputs` (value: `['/tmp/file0', '/tmp/file1', '/tmp/file2', '/tmp/file3']`, are propagated to some nodes, like `train2` or `test`, but not to the `LOO` node:
```
print(pipeline.nodes['LOO'].inputs)
[]
```
If I use `pipeline.dispatch_value(pipeline, 'main_inputs', pipeline.main_inputs)`, it doesn't change anything (is it the right way to use it ?)
The thing is that this function is using the `get_linked_items()` method, and when I use it directly:
```
list(pipeline.get_linked_items(pipeline, 'main_inputs', in_sub_pipelines=False, activated_only=False, process_only=False))
```
I obtain only one node/plug (`train2`/`in1`) and not the `LOO` node, which thus never receives its input values.

Oh OK, I see in the method `get_linked_items` that, despite the `process_only=False` option, it denies the existence of non-process nodes, except for switches: there is a `if isinstance(...)` sequence which tests for `Pipeline`, `Process` and `Swicth` instances, but does nothing for other node types.
Switches are handled specifically because they are used to propagate their input values to outputs, thus add output items in `get_linked_items()`. Other kinds of custom nodes may act similarly but a bit differently. I think custom nodes should be able to provide their "internal propagation" function(s) so that the system keeps extensible. This seems quite simple by moving the code for switches out of `get_linked_items()` and implementing it in a virtual `Node` method, which will have a specialization in `Switch`.
For `dispatch_value()` there is another problem: it calls `get_linked_items()` to get linked plugs, then assigns the _same_ value to all linked plugs. However some custom nodes do not only propagate values to other plugs, but may modify values (like sorting or filtering lists), which make different kinds of "linked items" but with _different_ values. So the system currently implemented does not allow custom nodes to work.
I think custom nodes are a bit complicated to manage in a dynamic environment (where all parameters are not known in advance). Filtering nodes should be processes unless all parameter can be computed before execution. By using processes  for custom nodes, we would keep the system simple and compatible with dynamic values. An to avoid the overhead of calling a process (serialize parameters, start a new Python, etc.) for a simple filter, we could provide parameter filtering processes that would embed the main process and perform the parameters filtering in the `execution` method.
Currently custom nodes are designed for static pipelines, yes. Making them dynamic would be nice but this would need huge modifications in capsul and workflows.
A filtering node is typically used before an iteration node, and today iterations are managed entirely statically. Making them dynamic would require to create new jobs / sub-workflows and operate parameters completion (using Capsul) during workflow execution. I think we have agreed not to dig into this for now, at least not before we reach a point where capsul v3 is able to replace all functionalities of capsul v2.
Otherwise making custom nodes `Process` instances is conceivable for _dynamic pipelines_ (at least when we don't have an iteration node depending on them), but as you said there are overheads, the biggest of which you did not even mention: in a cluster system a process is a cluster job, thus goes in an execution queue, waits for a slot to execute, runs on a separate machine with possibly separate filesystems, etc. However we already have tools to somewhat avoid some of the overheads, especially using Soma-Workflow's `EngineExecutionJob`s, which run directly in the engine, not in actual jobs.
But doing what you suggest will somewhat end up with moving all links and completion to the workflow system instead of the pipeline system. This is actually possibly a desirable evolution, but I don't feel we should deal with it right now...

Now after saying all that, I have to add that custom nodes have been made to ease development of Capsul pipelines, they have been developed quite recently, and as we don't really have Capsul-specific pipelines yet, these nodes have not been used for real applications yet. We could decide to drop them if they are really too difficult to integrate to Capsul v3, but I think that some utility tools like them would be useful for pipeline developers, and if they cannot fit in the capsul v3 features, I strongly fear that pipeline developers will face heavy difficulties to implement pipelines which actually need similar features. A cross-validation pipeline is a very classical architecture and I guess that Capsul should allow to build such pipelines.
I had only dynamic output values in mind but not dynamic workflow structure which I agree is too complex. I was wrong because I ignored cases where they are not independent such as cross validation. But I think this discussion point out some fundamental questions that we have to answer: 

- How far do we want to go in the pipeline design ? Should we keep it very simple with a minimum number of features or make it a bit less simple but with more useful features  ? We never talked about cross-validation in the use cases I think. This is a good example to answer this question.
- Do we want to support other workflow engines than Soma-Workflow ? If yes, it will be difficult to rely on Soma-Workflow specific features such as `EngineExecutionJob`. Most systems do not have an equivalent feature.
I'm starting with disabling all failing tests.
I would need some help to adapt `pipeline.test.test_custom_nodes` to new API. This is a rather big test case and it would save me time to discuss with @denisri (its author) about what it is supposed to do.
I will have a look at it, and we may talk (maybe this afternoon but I'm trying to finish something else this morning)
There is no rush. I wrote that not to forget. We will also have to work together on issue #228 I just created. We may have to plan one or several hands on meeting with some precise topics to work on.
The thing is that most of the custom nodes were designed to work with parameters values on-the-fly, on client side. For instance to separate a list of files in a train sublist and a test sublist for cross-validation pipelines. But now values are not propagated through pipeline nodes any longer, thus they don't get their inputs and don't produce their outputs correctly, I think - at least not in "real time" when debugging a pipeline in a graphical editor. So I don't know how to proceed with them...
How do we activate parameters values propagation through links ? Is it only using the completion system ?
There is almost no relation between the links and the path generation system. We can consider completion system as independent.
There is a value propagation triggered mainly when values are set on the pipeline itself (redefinition of `__setattr__`), but also internally when some internal values are changed (for instance for instance when temporary file names are set or, during execution, when a process change its output values). Propagation is done explicitly using `Pipeline`'s `dispatch_value` method. If something is not working it may indicate a bug in `dispatch_value` or a misuse of it internally. But it may also mean that the system is different (no parameter modification inside a pipeline is allowed) and we should rethink the example. Up to now, I had always been able to adapt tests without much change. We should put this test case on the list of things we should do together in order to see if a simple test adaptation is enough or if some changes in the API rules and internal machinery are required.
Well, then there should be a problem with values propagation. I run in an `ipython3 --gui=qt` shell the following code:
```
from capsul.api import executable
from capsul.qt_gui.widgets import PipelineDeveloperView
import os
pipeline = executable('capsul.pipeline.test.test_custom_nodes.Pipeline1')
pipeline.main_inputs = [os.path.join('/tmp', 'file%d' % i)
        for i in range(4)]
pipeline.test = pipeline.main_inputs[2]
pipeline.subject = 'subject2'
pipeline.output_directory = os.path.join('/tmp', 'out_dir')
view1 = PipelineDeveloperView(pipeline, allow_open_controller=True,
                                       show_sub_pipelines=True,
                                       enable_edition=True)
view1.show()
```
The input values, for instace for the pipeline parameter `main_inputs` (value: `['/tmp/file0', '/tmp/file1', '/tmp/file2', '/tmp/file3']`, are propagated to some nodes, like `train2` or `test`, but not to the `LOO` node:
```
print(pipeline.nodes['LOO'].inputs)
[]
```
If I use `pipeline.dispatch_value(pipeline, 'main_inputs', pipeline.main_inputs)`, it doesn't change anything (is it the right way to use it ?)
The thing is that this function is using the `get_linked_items()` method, and when I use it directly:
```
list(pipeline.get_linked_items(pipeline, 'main_inputs', in_sub_pipelines=False, activated_only=False, process_only=False))
```
I obtain only one node/plug (`train2`/`in1`) and not the `LOO` node, which thus never receives its input values.

Oh OK, I see in the method `get_linked_items` that, despite the `process_only=False` option, it denies the existence of non-process nodes, except for switches: there is a `if isinstance(...)` sequence which tests for `Pipeline`, `Process` and `Swicth` instances, but does nothing for other node types.
Switches are handled specifically because they are used to propagate their input values to outputs, thus add output items in `get_linked_items()`. Other kinds of custom nodes may act similarly but a bit differently. I think custom nodes should be able to provide their "internal propagation" function(s) so that the system keeps extensible. This seems quite simple by moving the code for switches out of `get_linked_items()` and implementing it in a virtual `Node` method, which will have a specialization in `Switch`.
For `dispatch_value()` there is another problem: it calls `get_linked_items()` to get linked plugs, then assigns the _same_ value to all linked plugs. However some custom nodes do not only propagate values to other plugs, but may modify values (like sorting or filtering lists), which make different kinds of "linked items" but with _different_ values. So the system currently implemented does not allow custom nodes to work.
I think custom nodes are a bit complicated to manage in a dynamic environment (where all parameters are not known in advance). Filtering nodes should be processes unless all parameter can be computed before execution. By using processes  for custom nodes, we would keep the system simple and compatible with dynamic values. An to avoid the overhead of calling a process (serialize parameters, start a new Python, etc.) for a simple filter, we could provide parameter filtering processes that would embed the main process and perform the parameters filtering in the `execution` method.
Currently custom nodes are designed for static pipelines, yes. Making them dynamic would be nice but this would need huge modifications in capsul and workflows.
A filtering node is typically used before an iteration node, and today iterations are managed entirely statically. Making them dynamic would require to create new jobs / sub-workflows and operate parameters completion (using Capsul) during workflow execution. I think we have agreed not to dig into this for now, at least not before we reach a point where capsul v3 is able to replace all functionalities of capsul v2.
Otherwise making custom nodes `Process` instances is conceivable for _dynamic pipelines_ (at least when we don't have an iteration node depending on them), but as you said there are overheads, the biggest of which you did not even mention: in a cluster system a process is a cluster job, thus goes in an execution queue, waits for a slot to execute, runs on a separate machine with possibly separate filesystems, etc. However we already have tools to somewhat avoid some of the overheads, especially using Soma-Workflow's `EngineExecutionJob`s, which run directly in the engine, not in actual jobs.
But doing what you suggest will somewhat end up with moving all links and completion to the workflow system instead of the pipeline system. This is actually possibly a desirable evolution, but I don't feel we should deal with it right now...

Now after saying all that, I have to add that custom nodes have been made to ease development of Capsul pipelines, they have been developed quite recently, and as we don't really have Capsul-specific pipelines yet, these nodes have not been used for real applications yet. We could decide to drop them if they are really too difficult to integrate to Capsul v3, but I think that some utility tools like them would be useful for pipeline developers, and if they cannot fit in the capsul v3 features, I strongly fear that pipeline developers will face heavy difficulties to implement pipelines which actually need similar features. A cross-validation pipeline is a very classical architecture and I guess that Capsul should allow to build such pipelines.
I had only dynamic output values in mind but not dynamic workflow structure which I agree is too complex. I was wrong because I ignored cases where they are not independent such as cross validation. But I think this discussion point out some fundamental questions that we have to answer: 

- How far do we want to go in the pipeline design ? Should we keep it very simple with a minimum number of features or make it a bit less simple but with more useful features  ? We never talked about cross-validation in the use cases I think. This is a good example to answer this question.
- Do we want to support other workflow engines than Soma-Workflow ? If yes, it will be difficult to rely on Soma-Workflow specific features such as `EngineExecutionJob`. Most systems do not have an equivalent feature.



Fix patch:
diff --git a/capsul/api.py b/capsul/api.py
index 1436e7096..1ff020ad0 100644
--- a/capsul/api.py
+++ b/capsul/api.py
@@ -30,3 +30,4 @@
 from .pipeline.pipeline_nodes import Node
 from .pipeline.pipeline_nodes import Switch
 from .application import Capsul, executable
+from .execution_context import CapsulWorkflow
diff --git a/capsul/application.py b/capsul/application.py
index 1c5626450..53b11251d 100644
--- a/capsul/application.py
+++ b/capsul/application.py
@@ -2,10 +2,11 @@
 import dataclasses
 import importlib
 import importlib.resources
+import inspect
 import json
+import os
 from pathlib import Path
 import types
-import inspect
 import sys
 
 from soma.controller import field, Controller
@@ -67,10 +68,31 @@ class Capsul(Singleton):
 
     '''    
 
-    def __singleton_init__(self, app_name='capsul', site_file=None):
+    def __singleton_init__(self, app_name='capsul', user_file=undefined, site_file=undefined):
+        if user_file is undefined:
+            user_file = os.environ.get('CAPSUL_USER_CONFIG')
+            if user_file is None:
+                user_file = os.path.expanduser('~/.config/capsul/capsul_user.json')
+                if not os.path.exists(user_file):
+                    user_file = None
+            elif not user_file:
+                user_file = None
+        if site_file is undefined:
+            site_file = os.environ.get('CAPSUL_SITE_CONFIG')
+            if site_file is None:
+                for site_file in (os.path.expandvars('$CASA_CONF/capsul_site.json'),
+                                  '/etc/capsul/capsul_site.json'):
+                    if os.path.exists(site_file):
+                        break
+                else:
+                    site_file = None
+            elif not site_file:
+                site_file = None
         if isinstance(site_file, Path):
             site_file=str(site_file)
-        c = ApplicationConfiguration(app_name=app_name, site_file=site_file)
+        if isinstance(user_file, Path):
+            user_file=str(user_file)
+        c = ApplicationConfiguration(app_name=app_name, user_file=user_file, site_file=site_file)
         self.config = c.merged_config
 
     @staticmethod
diff --git a/capsul/config/configuration.py b/capsul/config/configuration.py
index cee1a7cbe..2e57858b0 100644
--- a/capsul/config/configuration.py
+++ b/capsul/config/configuration.py
@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-
 
+import importlib
 import json
+import multiprocessing
 import os
 import sys
-import importlib
 import tempfile
 
 from soma.controller import (Controller, field,
@@ -13,10 +14,15 @@
 from ..dataset import Dataset
 
 
-default_workers_type = 'builtin'
-default_database_url = f'sqlite://{tempfile.gettempdir()}{os.sep}capsul_engine_database.sqlite'
-# default_workers_type = 'celery'
-# default_database_url = f'redis+socket://{tempfile.gettempdir()}{os.sep}capsul_engine_database.redis'
+default_engine_database = {
+    'type': 'redis+socket',
+    'path': f'{tempfile.gettempdir()}{os.sep}capsul_engine_database.rdb',
+}
+
+default_engine_start_workers = {
+    'type': 'builtin',
+    'count': max(1, int(multiprocessing.cpu_count()/4)),
+}
 
 def full_module_name(module_name):
     '''
@@ -148,8 +154,11 @@ class EngineConfiguration(Controller):
     dataset: OpenKeyDictController[Dataset]
     config_modules: list[str]
     python_modules: list[str]
-    workers_type: str = default_workers_type
-    database_url: str = default_database_url
+
+    database: field(type_=dict, default_factory= lambda: default_engine_database)
+    
+    start_workers: field(type_=dict, default_factory = lambda: default_engine_start_workers)
+
 
     def add_module(self, module_name, allow_existing=False):
         ''' Loads a modle and adds it in the engine configuration.
@@ -229,13 +238,25 @@ def __init__(self):
     def load(self, filename):
         ''' Load configuration from a JSON or YAML file
         '''
-        with open(filename) as f:
-            if filename.endswith('.yaml'):
-                # YAML support is optional, yaml module may not
-                # be installed
-                import yaml
+
+        if filename.endswith('.py'):
+            context = {}
+            exec(filename, globals=context, locals=context)
+            conf = None
+            for n in ('config', 'configuration', 'conf'):
+                conf = context.get(n)
+                if conf:
+                    break
+            if not conf:
+                raise RuntimeError(f'No valid configuration found in "{filename}"')
+        elif filename.endswith('.yaml'):
+            # YAML support is optional, yaml module may not
+            # be installed
+            import yaml
+            with open(filename) as f:
                 conf = yaml.safe_load(f)
-            else:
+        else:
+            with open(filename) as f:
                 conf = json.load(f)
         self.import_dict(conf)
 
diff --git a/capsul/database/__init__.py b/capsul/database/__init__.py
index f836abc44..264525940 100644
--- a/capsul/database/__init__.py
+++ b/capsul/database/__init__.py
@@ -3,7 +3,7 @@
 import dateutil.parser
 import importlib
 from pprint import pprint
-from urllib.parse import urlsplit, urlunsplit
+import re
 import sys
 import time
 
@@ -17,32 +17,104 @@
 
 database_classes = {
     'sqlite': 'capsul.database.populse_db:Populse_dbExecutionDatabase',
+    'redis': 'capsul.database.redis:RedisExecutionDatabase',
     'redis+socket': 'capsul.database.redis:RedisExecutionDatabase',
 }
 
-def execution_database(database_url):
-    url = urlsplit(database_url)
-    class_string = database_classes.get(url.scheme)
+class URL:
+    pattern = re.compile(
+        r'^(?P<scheme>[^:]+)://'
+        r'(?:(?P<login>[^:]+)(?::(?P<password>[^@]+))?@)?'
+        r'(?:(?P<host>[\w\.]+)(?::(?P<port>\d+))?)?'
+        r'(?P<path>[^;?#]*)'
+        r'(?:;(?P<parameters>[^?#]+))?'
+        r'(?:\?(?P<query>[^#]+))?'
+        r'(?:#(?P<fragment>.+))?$'
+        )
+
+    def __init__(self, string):
+        m = self.pattern.match(string)
+        if not m:
+            raise ValueError(f'Invalid URL: {string}')
+        for k, v in m.groupdict().items():
+            setattr(self, k, v)
+    
+    def __str__(self):
+        if self.login:
+            if self.password:
+                login = f'{self.login}:{self.password}@'
+            else:
+                login = f'{self.login}@'
+        else:
+            login = ''
+        if self.host:
+            if self.port:
+                host = f'{self.host}:{self.port}'
+            else:
+                host = self.host
+        else:
+            host = ''
+        if self.path:
+            path = self.path
+        else:
+            path = ''
+        if self.parameters:
+            parameters = f';{parameters}'
+        else:
+            parameters = ''
+        if self.query:
+            query = f'?{query}'
+        else:
+            query = ''
+        if self.fragment:
+            fragment = f'#{fragment}'
+        else:
+            fragment = ''
+        return f'{self.scheme}://{login}{host}{path}{parameters}{query}{fragment}'
+
+
+def engine_database(config):
+    class_string = database_classes.get(config['type'])
     if class_string is None:
-        raise ValueError(f'Invalid database URL {database_url}: scheme {url.scheme} is not supported')
+        raise ValueError(f'Invalid database type: {config["type"]}')
     module_name, class_name = class_string.rsplit(':', 1)
     module = importlib.import_module(module_name)
     database_class = getattr(module, class_name)
-    return database_class(url)
+    return database_class(config)
 
 
 class ExecutionDatabase:
-    def __init__(self, url):
-        self._url = url
+    def __init__(self, config):
+        self.config = config
+        self._path = None
 
     @property
-    def url(self):
-        url = urlunsplit(self._url)
-        if not self._url.netloc:
-            url = url.replace(':', '://', 1)
-        return url
+    def path(self):
+        if self._path is None:
+            self._path = self.config.get('path')
+        return self._path
+    
+    def workers_command(self, engine_id):
+        workers_command = []
+        config = self.engine_config(engine_id)
+        ssh = config.get('ssh')
+        if ssh:
+            host = ssh.get('host')
+            if not host:
+                raise ValueError('Host is mandatory in configuration for a ssh connection')
+            login = ssh.get('login')
+            if login:
+                host = f'{login}@{host}'
+            workers_command += ['ssh', '-f', host]
+        
+        casa_dir = config.get('casa_dir')
+        if casa_dir:
+            workers_command.append(f'{casa_dir}/bin/bv')
+
+        workers_command += ['python', '-m', f'capsul.engine.builtin', engine_id]
+        return workers_command
     
-    def new_execution(self, executable, execution_context, workflow, start_time):
+    def new_execution(self, executable, engine_id, execution_context, workflow, start_time):
         executable_json = json_encode(executable.json(include_parameters=False))
         execution_context_json = execution_context.json()
         workflow_parameters_json = json_encode(workflow.parameters.json())
@@ -55,6 +127,8 @@ def new_execution(self, executable, execution_context, workflow, start_time):
             else:
                 ready.append(job['uuid'])
         execution_id = self.store_execution(
+            engine_id,
+            label=executable.label,
             start_time=self._time_to_json(start_time),
             executable_json=executable_json,
             execution_context_json=execution_context_json,
@@ -65,14 +139,7 @@ def new_execution(self, executable, execution_context, workflow, start_time):
         )
         return execution_id
 
-    def _execution_context_from_json(self, execution_context_json):
-        return ExecutionContext(config=execution_context_json)
-
-    def execution_context(self, execution_id):
-        j = self.execution_context_json(execution_id)
-        if j is not None:
-            return self._execution_context_from_json(j)
-
+        
     def _executable_from_json(self, executable_json):
         return Capsul.executable(json_decode(executable_json))
     
@@ -80,20 +147,20 @@ def executable(self, execution_id):
         j = self.executable_json(execution_id)
         if j is not None:
             return self._executable_from_json(j)
-
-    def _workflow_parameters_from_json(self, workflow_parameters_json):
-        return DictWithProxy.from_json(json_decode(workflow_parameters_json))
     
-    def workflow_parameters(self, execution_id):
-        j = self.workflow_parameters_json(execution_id)
+    def workflow_parameters(self, engine_id, execution_id):
+        j = self.workflow_parameters_json(engine_id, execution_id)
         if j:
-            return self._workflow_parameters_from_json(j)
+            return DictWithProxy.from_json(json_decode(j))
+
 
-    def set_workflow_parameters(self, execution_id, workflow_parameters):
-        self.set_workflow_parameters_json(execution_id, json_encode(workflow_parameters.json()))
+    def set_workflow_parameters(self, engine_id, execution_id, workflow_parameters):
+        self.set_workflow_parameters_json(engine_id, execution_id, json_encode(workflow_parameters.json()))
+
+
+    def update_workflow_parameters(self, engine_id, execution_id, parameters_location, output_values):
+        self.update_workflow_parameters_json(engine_id, execution_id, parameters_location, json_encode(output_values))
 
-    def update_workflow_parameters(self, execution_id, parameters_location, output_values):
-        self.update_workflow_parameters_json(execution_id, parameters_location, json_encode(output_values))
 
     @staticmethod
     def _time_from_json(time_json):
@@ -103,17 +170,6 @@ def _time_from_json(time_json):
     def _time_to_json(time):
         return time.isoformat()
     
-    def start_time(self, execution_id):
-        j = self.start_time_json(execution_id)
-        if j:
-            return self._time_from_json(j)
-        return None
-
-    def end_time(self, execution_id):
-        j = self.end_time_json(execution_id)
-        if j:
-            return self._time_from_json(j)
-        return None
 
     def _job_from_json(self, job):
         for k in ('start_time', 'end_time'):
@@ -122,6 +178,7 @@ def _job_from_json(self, job):
                 job[k] = self._time_from_json(t)
         return job
 
+
     def _job_to_json(self, job):
         for k in ('start_time', 'end_time'):
             t = job.get(k)
@@ -129,21 +186,26 @@ def _job_to_json(self, job):
                 job[k] = self._time_to_json(t)
         return job
 
-    def jobs(self, execution_id):
-        return [self._job_from_json(i) for i in self.jobs_json(execution_id)]
 
-    def job(self, execution_id, job_uuid):
-        j = self.job_json(execution_id, job_uuid)
+    def job(self, engine_id, execution_id, job_uuid):
+        j = self.job_json(engine_id, execution_id, job_uuid)
         if j:
             return self._job_from_json(j)
         return None
 
-    def execution_report(self, execution_id):
-        report = self.execution_report_json(execution_id)
+
+    def execution_report(self, engine_id, execution_id):
+        report = self.execution_report_json(engine_id, execution_id)
+        report['engine_id'] = engine_id
         report['execution_id'] = execution_id
-        for n in ('executable', 'execution_context', 'workflow_parameters'):
-            convert = getattr(self, f'_{n}_from_json')
-            report[n] = convert(report[n])
+        execution_context = report['execution_context']
+        if execution_context is not None:
+            execution_context = ExecutionContext(config=execution_context)
+        report['execution_context'] = execution_context
+        workflow_parameters = report['workflow_parameters']
+        if workflow_parameters is not None:
+            workflow_parameters = DictWithProxy.from_json(workflow_parameters)
+        report['workflow_parameters'] = workflow_parameters        
         for n in ('start_time', 'end_time'):
             j = report.get(n)
             if j:
@@ -169,7 +231,7 @@ def print_execution_report(self, report, file=sys.stdout):
         print('====================\n'
               '| Execution report |\n'
               '====================\n', file=file)
-        print('executable:', report['executable'].definition, file=file)
+        print('label:', report['label'], file=file)
         print('status:', report['status'], file=file)
         print('start time:', report['start_time'], file=file)
         print('end time:', report['end_time'], file=file)
@@ -190,7 +252,6 @@ def print_execution_report(self, report, file=sys.stdout):
         print('ongoing:', report['ongoing'])
         print('done:', report['done'])
         print('failed:', report['failed'])
-
         now = datetime.now()
         for job in sorted(report['jobs'], key=lambda j: (j.get('start_time') if j.get('start_time') else now)):
             job_uuid = job['uuid']
@@ -200,7 +261,7 @@ def print_execution_report(self, report, file=sys.stdout):
             pipeline_node = '.'.join(i for i in job.get('parameters_location', '') if i != 'nodes')
             returncode = job.get('returncode')
             status = job['status']
-            command = ' '.join(f"'{i}'" for i in job.get('command', []))
+            disabled = job['disabled']
             stdout = job.get('stdout')
             stderr = job.get('stderr')
             parameters = report['workflow_parameters']
@@ -220,7 +281,7 @@ def print_execution_report(self, report, file=sys.stdout):
             print('returncode:', returncode, file=file)
             print('start time:', start_time, file=file)
             print('end time:', end_time, file=file)
-            print('command:', command, file=file)
+            print('disabled:', disabled, file=file)
             print('wait for:', wait_for, file=file)
             print('waited_by:', waited_by, file=file)
             if parameters:
@@ -245,33 +306,33 @@ def print_execution_report(self, report, file=sys.stdout):
                 print(v, file=file)
                 print(file=file)
 
-    def wait(self, execution_id, timeout=None):
+    def wait(self, engine_id, execution_id, timeout=None):
         start = time.time()
-        status = self.status(execution_id)
+        status = self.status(engine_id, execution_id)
         if status == 'ready':
-            for i in range(50):
-                time.sleep(0.1)
-                status = self.status(execution_id)
+            for i in range(100):
+                time.sleep(0.2)
+                status = self.status(engine_id, execution_id)
                 if status != 'ready':
                     break
             else:
-                self.print_execution_report(self.execution_report(execution_id), file=sys.stderr)
+                self.print_execution_report(self.execution_report(engine_id, execution_id), file=sys.stderr)
                 raise SystemError(f'workers are too slow to start execution ({datetime.now()})')
-        status = self.status(execution_id)
-        while status == 'running':
+        status = self.status(engine_id, execution_id)
+        while status != 'ended':
             if timeout is not None and (time.time() - start) > timeout:
                 raise TimeoutError('Process execution timeout')
             time.sleep(0.1)
-            status = self.status(execution_id)
+            status = self.status(engine_id, execution_id)
 
-    def raise_for_status(self, execution_id):
-        error = self.error(execution_id)
+    def raise_for_status(self, engine_id, execution_id):
+        error = self.error(engine_id, execution_id)
         if error:
-            self.print_execution_report(self.execution_report(execution_id), file=sys.stderr)
+            self.print_execution_report(self.execution_report(engine_id, execution_id), file=sys.stderr)
             raise RuntimeError(error)
 
-    def update_executable(self, executable, execution_id):
-        parameters = self.workflow_parameters(execution_id)
+    def update_executable(self, engine_id, execution_id, executable):
+        parameters = self.workflow_parameters(engine_id, execution_id)
         # print('!update_executable!')
         # from pprint import pprint
         # pprint(parameters.proxy_values)
@@ -304,8 +365,161 @@ def update_executable(self, executable, execution_id):
             if enable_parameter_links is not None:
                 executable.enable_parameter_links = enable_parameter_links
 
-    def start_one_job(self, execution_id, start_time):
-        return self.start_one_job_json(execution_id, self._time_to_json(start_time))
+
+    def __enter__(self):
+        raise NotImplementedError
+
+
+    def __exit__(self, exception_type, exception_value, exception_traceback):
+        raise NotImplementedError
+
+
+    def connect_to_engine(self, engine):
+        '''
+        Creates a connection to an engine in the database.
+        Engine is created if it does not exists.
+        '''
+        raise NotImplementedError
+
+
+    def engine_config(self, engine_id):
+        '''
+        Return the configuration dict stored for an engine
+        '''
+        raise NotImplementedError
+
+
+    def number_of_workers_to_start(self, engine_id):
+        '''
+        Return the number of workers to start in order to reach the
+        workers_count number defined in engine configuration.
+        '''
+        raise NotImplementedError
+
+
+    def worker_database_config(self, engine_id):
+        '''
+        Return database connection settings for workers. This
+        connection may be different than the client connection
+        if the database implementation create engine specific
+        internal access with restricted access rights.
+        '''
+        raise NotImplementedError
+
+
+    def worker_started(self, engine_id):
+        '''
+        Register a new worker that had been started for this engine and
+        return an identifier for it.
+        '''
+        raise NotImplementedError
+
+
+    def worker_ended(self, engine_id, worker_id):
+        '''
+        Remove a worker from the list of workers for this engine.
+        '''
+        raise NotImplementedError
+
+
+    def dispose_engine(self, engine_id):
+        '''
+        Tell Capsul that this engine will not be used anymore by any client.
+        The ressource it uses must be freed as soon as possible. If no 
+        execution is running, engine is destroyed. Otherwise, workers will
+        process ongoing executions and cleanup when done.
+        '''
+        raise NotImplementedError
+
+
+    def store_execution(self,
+            engine_id,
+            label,
+            start_time, 
+            executable_json,
+            execution_context_json,
+            workflow_parameters_json,
+            jobs,
+            ready,
+            waiting
+        ):
+        raise NotImplementedError
+
+
+    def execution_context(self, engine_id, execution_id):
+        j = self.execution_context_json(engine_id, execution_id)
+        if j is not None:
+            return ExecutionContext(config=j)
+
+
+    def execution_context_json(self, engine_id, execution_id):
+        raise NotImplementedError
+
+
+    def pop_job(self, engine_id, start_time):
+        '''
+        Convert its parameters to JSON and calls pop_job_json()
+        '''
+        return self.pop_job_json(engine_id, self._time_to_json(start_time))
     
-    def job_finished(self, execution_id, job_uuid, end_time, returncode, stdout, stderr):
-        return self.job_finished_json(execution_id, job_uuid, self._time_to_json(end_time), returncode, stdout, stderr)
+
+
+    def pop_job_json(self, engine_id, start_time):
+        raise NotImplementedError
+
+
+    def job_finished(self, engine_id, execution_id, job_uuid, end_time, returncode, stdout, stderr):
+        '''
+        Convert its parameters to JSON and calls job_finished_json()
+        '''
+        self.job_finished_json(engine_id, execution_id, job_uuid, 
+            self._time_to_json(end_time), returncode, stdout, stderr)
+
+
+    def job_finished_json(self, engine_id, execution_id, job_uuid, end_time, returncode, stdout, stderr):
+        raise NotImplementedError
+
+
+
+    def status(self, engine_id, execution_id):
+        raise NotImplementedError
+
+        
+    def workflow_parameters_json(self, engine_id, execution_id):
+        raise NotImplementedError
+
+
+    def set_workflow_parameters_json(self, engine_id, execution_id, workflow_parameters_json):
+        raise NotImplementedError
+
+
+    def update_workflow_parameters_json(self, engine_id, execution_id, parameters_location, output_values):
+        raise NotImplementedError
+
+
+    def job_json(self, engine_id, execution_id, job_uuid):
+        raise NotImplementedError
+
+   
+    def execution_report_json(self, engine_id, execution_id):
+        raise NotImplementedError
+
+
+    def dispose(self, engine_id, execution_id):
+        raise NotImplementedError
+
+
+    def start_execution(self, engine_id, execution_id, tmp):
+        raise NotImplementedError
+
+
+    def end_execution(self, engine_id, execution_id):
+        raise NotImplementedError
+
+
+    def tmp(self, engine_id, execution_id):
+        raise NotImplementedError
+
+
+    def error(self, engine_id, execution_id):
+        raise NotImplementedError
diff --git a/capsul/database/populse_db.py b/capsul/database/populse_db.py
index a5e43a38a..4ca4b5f8d 100644
--- a/capsul/database/populse_db.py
+++ b/capsul/database/populse_db.py
@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 import os
+import time
 from uuid import uuid4
 
 from populse_db import Database
@@ -7,11 +8,11 @@
 from . import ExecutionDatabase
 
 class Populse_dbExecutionDatabase(ExecutionDatabase):
-    def __init__(self, url):
+    def __init__(self, url, prefix=None):
         super().__init__(url)
-        if url.query or url.fragment:
+        if url.login or url.host or url.query or url.fragment:
             raise ValueError(f'Invalid URL: {self.url}')
-        self.sqlite_file = f'{url.netloc}{url.path}'
+        self.sqlite_file = f'{url.path}'
         self.database = Database(f'sqlite://{self.sqlite_file}')
         with self.database as session:
             if not session.has_collection('execution'):
@@ -20,8 +21,17 @@ def __init__(self, url):
                 session['global'][''] = {
                     'capsul_connection_count': 1,
                 }
+
+                session.add_collection('worker')
+                worker = session['worker']
+                worker.add_field('label', str)
+                worker.add_field('engine_config', dict)
+                worker.add_field('executions', list[str])
+                worker.add_field('status', str)
+
                 session.add_collection('execution')
                 execution = session['execution']
+                execution.add_field('label', str)
                 execution.add_field('status', str)
                 execution.add_field('error', str)
                 execution.add_field('error_detail', str)
@@ -68,7 +78,49 @@ def _set(self, execution_id, name, value):
         with self.database as session:
             session['execution'].update_document(execution_id, {name: value})
 
+
+    def new_workers_json(self, label, engine_config_json):
+        workers_id = str(uuid4())
+        with self.database as session:
+            session['worker'][workers_id] = {
+                'label': label,
+                'engine_config': engine_config_json,
+                'executions': [],
+                'status': 'starting',
+            }
+        return workers_id
+
+
+    def workers_started(self, workers_id):
+        with self.database as session:
+            session['worker'].update_document(workers_id, {'status': 'running'})
+
+
+    def workers_status(self, workers_id):
+        with self.database as session:
+            return session['worker'].document(workers_id, fields=['status'], as_list=True)[0]
+
+
+    def dispose_workers(self, workers_id):
+        with self.database as session:
+            del session['worker'][workers_id]
+
+
+    def get_execution(self, workers_id):
+        with self.database as session:
+            workers_active = session['worker'].has_document(workers_id)
+            if workers_active:
+                executions = session['worker'].document(workers_id, fields=['executions'], as_list=True)[0]
+                if executions:
+                    execution_id = executions[0]
+                    session['worker'].update_document(workers_id, {'executions': executions[1:]})
+                    return execution_id
+                return ''
+        return None
+
     def store_execution(self,
+            workers_id,
+            label,
             start_time, 
             executable_json,
             execution_context_json,
@@ -79,6 +131,7 @@ def store_execution(self,
         ):
         execution_id = str(uuid4())
         execution = {
+            'label': label,
             'status': 'ready',
             'start_time': start_time,
             'executable': executable_json,
@@ -98,6 +151,9 @@ def store_execution(self,
             session['execution'][execution_id] = execution
             for job in jobs:
                 session['job'][(execution_id, job['uuid'])] = job
+            executions = session['worker'].document(workers_id, fields=['executions'], as_list=True)[0]
+            executions.append(execution_id)
+            session['worker'].update_document(workers_id, {'executions': executions})
         return execution_id
     
     def status(self, execution_id):
@@ -222,8 +278,9 @@ def job_finished_json(self, execution_id, job_uuid, end_time, returncode, stdout
                     job = session['job'][(execution_id, uuid)]
                     job['returncode'] = 'Not started because de dependent job failed'
                     session['job'][(execution_id, uuid)] = job
-                    execution['waiting'].remove(uuid)
-                    execution['failed'].append(uuid)
+                    if uuid in execution['waiting']:
+                        execution['waiting'].remove(uuid)
+                        execution['failed'].append(uuid)
                     stack.extend(job.get('waited_by', []))
             else:
                 execution['done'].append(job_uuid)
@@ -250,7 +307,7 @@ def job_finished_json(self, execution_id, job_uuid, end_time, returncode, stdout
     def execution_report_json(self, execution_id):
         with self.database as session:
             result = dict(
-                executable = self.executable_json(execution_id),
+                label = self.label(execution_id),
                 execution_context = self.execution_context_json(execution_id),
                 status = self.status(execution_id),
                 error = self.error(execution_id),
@@ -262,7 +319,7 @@ def execution_report_json(self, execution_id):
                 ongoing = self.ongoing(execution_id),
                 done = self.done(execution_id),
                 failed = self.failed(execution_id),
-                jobs = self.jobs_json(execution_id),
+                jobs = list(self.jobs_json(execution_id)),
                 workflow_parameters = self.workflow_parameters_json(execution_id),
                 engine_debug = {}
             )
@@ -281,3 +338,6 @@ def set_tmp(self, execution_id, tmp):
 
     def tmp(self, execution_id):
         return self._get(execution_id, 'tmp')
+
+    def label(self, execution_id):
+        return self._get(execution_id, 'label')
diff --git a/capsul/database/redis.py b/capsul/database/redis.py
index 88b41c606..d72f67af4 100644
--- a/capsul/database/redis.py
+++ b/capsul/database/redis.py
@@ -46,72 +46,146 @@ def execute(self):
         return self.pipeline.execute()
 
 class RedisExecutionDatabase(ExecutionDatabase):
-    def __init__(self, url):
-        super().__init__(url)
+    def __enter__(self):
         self.uuid = str(uuid4())
-        if url.scheme != 'redis+socket':
-            raise NotImplementedError(f'Redis connection with {self.url} is not implemented')
-
-        self.redis_socket = f'{url.netloc}{url.path}'
-        if not os.path.exists(self.redis_socket):
-            # Start the redis server
-            tmp = tempfile.mkdtemp(prefix='capsul_redis_')
-            try:
-                pid_file = f'{tmp}/redis.pid'
-                log_file = f'{tmp}/redis.log'
-                cmd = [
-                    'redis-server',
-                    '--unixsocket', self.redis_socket,
-                    '--port', '0', # port 0 means no TCP connection
-                    '--daemonize', 'yes',
-                    '--pidfile', pid_file,
-                    '--dir', tmp,
-                    '--logfile', log_file,
-                    '--dbfilename', 'redis.rdb',
-                ]
-                subprocess.run(cmd)
-                for i in range(20):
-                    if os.path.exists(self.redis_socket):
-                        break
-                    time.sleep(0.1)
+
+        if self.config['type'] == 'redis+socket':
+            if not self.path:
+                raise ValueError('Database path is missing in configuration')
+            self.redis_socket = f'{self.path}.socket'
+            if not os.path.exists(self.redis_socket):
+                # Start the redis server
+                tmp = tempfile.mkdtemp(prefix='capsul_redis_')
+                try:
+                    dir, dbfilename = os.path.split(self.path)
+                    pid_file = f'{tmp}/redis.pid'
+                    log_file = f'{tmp}/redis.log'
+                    cmd = [
+                        'redis-server',
+                        '--unixsocket', self.redis_socket,
+                        '--port', '0', # port 0 means no TCP connection
+                        '--daemonize', 'yes',
+                        '--pidfile', pid_file,
+                        '--dir', dir,
+                        '--logfile', log_file,
+                        '--dbfilename', dbfilename,
+                    ]
+                    subprocess.run(cmd, cwd=tmp)
+                    for i in range(20):
+                        if os.path.exists(self.redis_socket):
+                            break
+                        time.sleep(0.1)
+                    self.redis  = redis.Redis(unix_socket_path=self.redis_socket,
+                                                decode_responses=True)
+                    self.redis.delete('capsul:shutting_down') 
+                    self.redis.set('capsul:redis_tmp', tmp)
+                    self.redis.set('capsul:redis_pid_file', pid_file)
+                except Exception:
+                    shutil.rmtree(tmp)
+            else:
                 self.redis  = redis.Redis(unix_socket_path=self.redis_socket,
-                                            decode_responses=True)
-                self.redis.set('capsul_redis_tmp', tmp)
-                self.redis.set('capsul_redis_pid_file', pid_file)
-            except Exception:
-                shutil.rmtree(tmp)
-        else:
-            self.redis  = redis.Redis(unix_socket_path=self.redis_socket,
+                                          decode_responses=True)
+        elif self.config['type'] == 'redis':
+            self.redis  = redis.Redis(host=self.config['host'], port=self.config.get('port'),
                                       decode_responses=True)
-        
-        self.redis.hset('capsul_connections', self.uuid, datetime.now().isoformat())
+            if self.config.get('login'):
+                self.redis.auth(self.config['password'], self.config['login'])
+        else:
+            raise NotImplementedError(f'Invalid Redis connection type: {self.config["type"]}')
+        if self.redis.get('capsul:shutting_down'):
+            raise RuntimeError('Cannot connect to database because it is shutting down')
+        self.redis.hset('capsul:connections', self.uuid, datetime.now().isoformat())
 
         # Some functions are implemented as a Lua script in redis
         # in order to be atomic. In redis these scripts must always
         # be registered before using them.
-        self._start_one_job = self.redis.register_script('''
-            local function key(name)
-                return 'capsul:' .. ARGV[1] .. ':' .. name
+        self._store_execution = self.redis.register_script('''
+            local engine_key = KEYS[1]
+            local execution_key = KEYS[2]
+
+            local execution_id = ARGV[1]
+            local label = ARGV[2]
+            local start_time = ARGV[3]
+            local executable_json = ARGV[4]
+            local execution_context_json = ARGV[5]
+            local workflow_parameters_json = ARGV[6]
+            local jobs = cjson.decode(ARGV[7])
+            local ready = ARGV[8]
+            local waiting = ARGV[9]
+
+            local executions = cjson.decode(redis.call('hget', engine_key, 'executions'))
+            table.insert(executions, execution_id)
+            redis.call('hset', engine_key, 'executions', cjson.encode(executions))
+
+            redis.call('hset', execution_key, 'label', label)
+            redis.call('hset', execution_key, 'status', 'ready')
+            redis.call('hset', execution_key, 'start_time', start_time)
+            redis.call('hset', execution_key, 'executable', executable_json)
+            redis.call('hset', execution_key, 'execution_context', execution_context_json)
+            redis.call('hset', execution_key, 'workflow_parameters', workflow_parameters_json)
+
+            redis.call('hset', execution_key, 'ready', ready)
+            --  An empty list modified with Redis Lua scripts may be encoded as empty dict
+            if ready  == '[]' or ready == '{}' then
+                redis.call('hset', execution_key, 'status', 'ended')
+                redis.call('hset', execution_key, 'end_time', start_time)
+            end
+
+            redis.call('hset', execution_key, 'waiting', waiting)
+            redis.call('hset', execution_key, 'ongoing', '[]')
+            redis.call('hset', execution_key, 'done', '[]')
+            redis.call('hset', execution_key, 'failed', '[]')
+
+            for _, job in pairs(jobs) do
+                redis.call('hset', execution_key, 'job:' .. job['uuid'], cjson.encode(job))
             end
+            ''')
+
+        self._pop_job = self.redis.register_script('''
+            local execution_key = KEYS[1]
+
+            local start_time = ARGV[2]
+
+            local status = redis.call('hget', execution_key, 'status')
 
-            local job_uuid = redis.call('lpop', key('ready'))
-            if job_uuid then
-                redis.call('set', key('status'), 'running')
-                local job = cjson.decode(redis.call('hget', key('jobs'), job_uuid))
-                job['start_time'] = ARGV[2]
-                job = cjson.encode(job)
-                redis.call('hset', key('jobs'), job_uuid, job)
-                redis.call('rpush', key('ongoing'), job_uuid)
-                return job
+            if status == 'ready' then
+                redis.call('hset', execution_key, 'status', 'initialization')
+                return 'start_execution'
+            end
+            if status == 'running' then
+                local ready = cjson.decode(redis.call('hget', execution_key, 'ready'))
+                if next(ready) ~= nil then
+                    local job_uuid = table.remove(ready, 1)
+                    redis.call('hset', execution_key, 'ready', cjson.encode(ready))
+                    local job_key = 'job:' .. job_uuid
+                    local job = cjson.decode(redis.call('hget', execution_key, job_key))
+                    job['start_time'] = ARGV[2]
+                    job = cjson.encode(job)
+                    redis.call('hset', execution_key, job_key, job)
+                    local ongoing = cjson.decode(redis.call('hget', execution_key, 'ongoing'))
+                    table.insert(ongoing, job_uuid)
+                    redis.call('hset', execution_key, 'ongoing', cjson.encode(ongoing))
+                    return job_uuid
+                end
+            end
+            if status == 'finalization' then
+                return 'end_execution'
             end
             return nil
         ''')
 
         self._job_finished = self.redis.register_script('''
-            local function key(name)
-                return 'capsul:' .. ARGV[1] .. ':' .. name
+            local function table_find(array, value)
+                for i, v in pairs(array) do
+                    if v == value then
+                        return i
+                    end
+                end
             end
 
+            local engine_key = KEYS[1]
+            local execution_key = KEYS[2]
+
             local execution_id = ARGV[1]
             local job_uuid = ARGV[2]
             local end_time = ARGV[3]
@@ -119,17 +193,26 @@ def __init__(self, url):
             local stdout = ARGV[5]
             local stderr = ARGV[6]
 
-            local job = cjson.decode(redis.call('hget', key('jobs'), job_uuid))
+            local job_key = 'job:' .. job_uuid
+            local job = cjson.decode(redis.call('hget', execution_key, job_key))
             job['end_time'] = end_time
             job['returncode'] = returncode
             job['stdout'] = stdout
             job['stderr'] = stderr
-            redis.call('hset', key('jobs'), job_uuid, cjson.encode(job))
+            redis.call('hset', execution_key, job_key, cjson.encode(job))
+
+            local ready = cjson.decode(redis.call('hget', execution_key, 'ready'))
+            local ongoing = cjson.decode(redis.call('hget', execution_key, 'ongoing'))
+            local failed = cjson.decode(redis.call('hget', execution_key, 'failed'))
+            local waiting = cjson.decode(redis.call('hget', execution_key, 'waiting'))
+            local done = cjson.decode(redis.call('hget', execution_key, 'done'))
 
-            redis.call('lrem', key('ongoing'), 1, job_uuid)
+
+            table.remove(ongoing, table_find(ongoing, job_uuid))
 
             if returncode ~= '0' then
-                redis.call('rpush', key('failed'), job_uuid)
+                table.insert(failed, job_uuid)
+
                 local stack = {}
                 if job['waited_by'] then
                     for key, value in pairs(job['waited_by']) do
@@ -139,11 +222,13 @@ def __init__(self, url):
 
                 local uuid = table.remove(stack)
                 while uuid do
-                    job = cjson.decode(redis.call('hget', key('jobs'), uuid))
+                    local job_key = 'job:' .. uuid
+                    job = cjson.decode(redis.call('hget', execution_key, job_key))
                     job['returncode'] = 'Not started because de dependent job failed'
-                    redis.call('hset', key('jobs'), uuid, cjson.encode(job))
-                    redis.call('lrem', key('waiting'), 1, uuid)
-                    redis.call('rpush', key('failed'), uuid)
+                    redis.call('hset', execution_key, job_key, cjson.encode(job))
+
+                    table.remove(waiting, table_find(waiting, uuid))
+                    table.insert(failed, uuid)
                     if job['waited_by'] then
                         for key, value in ipairs(job['waited_by']) do
                             table.insert(stack, value)
@@ -152,179 +237,256 @@ def __init__(self, url):
                     uuid = table.remove(stack)
                 end
             else
-                redis.call('rpush', key('done'), job_uuid)
+                table.insert(done, job_uuid)
                 if job['waited_by'] then
                     for _, waiting_uuid in ipairs(job['waited_by']) do
-                        local waiting_job = cjson.decode(redis.call('hget', key('jobs'), waiting_uuid))
+                        local waiting_job = cjson.decode(redis.call('hget', execution_key, 'job:' .. waiting_uuid))
                         local ready_to_go = true
                         if waiting_job['wait_for'] then
-                            for key2, waited in ipairs(waiting_job['wait_for']) do
-                                if not redis.call('lpos', key('done'), waited) then
+                            for _, waited in ipairs(waiting_job['wait_for']) do
+                                if not table_find(done, waited) then
                                     ready_to_go = false
                                     break
                                 end
                             end
                         end
                         if ready_to_go then
-                        redis.call('lrem', key('waiting'), 1, waiting_uuid)
-                            redis.call('rpush', key('ready'), waiting_uuid)
+                            table.remove(waiting, table_find(waiting, waiting_uuid))
+                            table.insert(ready, waiting_uuid)
                         end
                     end
                 end
             end
 
-            if (redis.call('llen', key('ongoing')) ~= 0) or (redis.call('llen', key('ready')) ~= 0) then
-                return false
-            else
-                redis.call('hdel', 'capsul_ongoing_executions', execution_id)
-                if redis.call('hexists', 'capsul_undisposed_executions', execution_id) then
-                    if redis.call('llen', key('failed')) ~= 0 then
-                        redis.call('set', key('error'), 'Some jobs failed')
-                    end
-                    redis.call('set', key('status'), 'ended')
-                    redis.call('set', key('end_time'), end_time)
-                    return true
-                else
-                    -- TODO: reuse _dispose script
-                    redis.call('del', key('executable'))
-                    redis.call('del', key('execution_context'))
-                    redis.call('del', key('status'))
-                    redis.call('del', key('error'))
-                    redis.call('del', key('error_detail'))
-                    redis.call('del', key('start_time'))
-                    redis.call('del', key('end_time'))
-                    redis.call('del', key('waiting'))
-                    redis.call('del', key('ready'))
-                    redis.call('del', key('ongoing'))
-                    redis.call('del', key('done'))
-                    redis.call('del', key('failed'))
-                    redis.call('del', key('jobs'))
-                    redis.call('del', key('workflow_parameters'))
-                    local tmp = redis.call('get', key('tmp'))
-                    redis.call('del', key('tmp'))
-                    return tmp
+            redis.call('hset', execution_key, 'ready', cjson.encode(ready))
+            redis.call('hset', execution_key, 'ongoing', cjson.encode(ongoing))
+            redis.call('hset', execution_key, 'failed', cjson.encode(failed))
+            redis.call('hset', execution_key, 'waiting', cjson.encode(waiting))
+            redis.call('hset', execution_key, 'done', cjson.encode(done))
+
+            if (next(ongoing) == nil) and (next(ready) == nil) then
+                if next(failed) ~= nil then
+                    redis.call('hset', execution_key, 'error', 'Some jobs failed')
                 end
+                redis.call('hset', execution_key, 'status', 'finalization')
+                redis.call('hset', execution_key, 'end_time', end_time)
             end
         ''')
 
-        self._full_report = self.redis.register_script('''
-            local function key(name)
-                return 'capsul:' .. ARGV[1] .. ':' .. name
-            end
+        self._update_workflow_parameters = self.redis.register_script('''
+            local execution_key = KEYS[1]
 
-            local executable = redis.call('get', key('executable'))
-            local execution_context = redis.call('get', key('execution_context'))
-            local workflow_parameters = redis.call('get', key('workflow_parameters'))
-            local status = redis.call('get', key('status'))
-            local error = redis.call('get', key('error'))
-            local error_detail = redis.call('get', key('error_detail'))
-            local start_time = redis.call('get', key('start_time'))
-            local end_time = redis.call('get', key('end_time'))
-            local waiting = redis.call('lrange', key('waiting'), 0, -1)
-            local ready = redis.call('lrange', key('ready'), 0, -1)
-            local ongoing = redis.call('lrange', key('ongoing'), 0, -1)
-            local done = redis.call('lrange', key('done'), 0, -1)
-            local failed = redis.call('lrange', key('failed'), 0, -1)
-            local jobs = {}
-            for i, v in ipairs(redis.call('hgetall', key('jobs'))) do
-                if i % 2 == 0 then
-                    table.insert(jobs, v)
+            local parameters_location = cjson.decode(ARGV[1])
+            local output_parameters = cjson.decode(ARGV[2])
+            local workflow_parameters = cjson.decode(redis.call('hget', execution_key, 'workflow_parameters'))
+
+            local parameters = workflow_parameters['content']
+            for index, value in ipairs(parameters_location) do
+                local i = tonumber(value)
+                if i then
+                    parameters = parameters[i+1]
+                else
+                    parameters = parameters[value]
                 end
             end
-            return {executable, execution_context, workflow_parameters, 
-                status, error, error_detail, start_time, end_time, 
-                waiting, ready, ongoing, done, failed, jobs}                
+
+            for k, v in pairs(output_parameters) do
+                workflow_parameters['proxy_values'][parameters[k][2]+1] = v
+            end
+            redis.call('hset', execution_key, 'workflow_parameters', cjson.encode(workflow_parameters))
             ''')
 
+
         self._dispose = self.redis.register_script('''
-            local function key(name)
-                return 'capsul:' .. ARGV[1] .. ':' .. name
+            local function table_find(array, value)
+                for i, v in pairs(array) do
+                    if v == value then
+                        return i
+                    end
+                end
             end
 
+            local engine_key = KEYS[1]
+            local execution_key = KEYS[2]
+
             local execution_id = ARGV[1]
 
-            redis.call('hdel', 'capsul_undisposed_executions', execution_id)
-            if redis.call('get', key('status')) == 'ended' then
-                redis.call('del', key('executable'))
-                redis.call('del', key('execution_context'))
-                redis.call('del', key('status'))
-                redis.call('del', key('error'))
-                redis.call('del', key('error_detail'))
-                redis.call('del', key('start_time'))
-                redis.call('del', key('end_time'))
-                redis.call('del', key('waiting'))
-                redis.call('del', key('ready'))
-                redis.call('del', key('ongoing'))
-                redis.call('del', key('done'))
-                redis.call('del', key('failed'))
-                redis.call('del', key('jobs'))
-                redis.call('del', key('workflow_parameters'))
-                local tmp = redis.call('get', key('tmp'))
-                redis.call('del', key('tmp'))
-                return tmp
-            else
-                return false
+            redis.call('hset', execution_key, 'dispose', 1)
+            if redis.call('hget', execution_key, 'status') == 'ended' then
+                redis.call('del', execution_key)
+                local executions = cjson.decode(redis.call('hget', engine_key, 'executions'))
+                table.remove(executions, table_find(executions, execution_id))
+                redis.call('hset', engine_key, 'executions', cjson.encode(executions))
             end
             ''')
 
-        self._check_shutdown = self.redis.register_script('''
-            if redis.call('hlen', 'capsul_connections') == 0 and
-                redis.call('hlen', 'capsul_ongoing_executions') == 0 and
-                redis.call('llen', 'capsul_undisposed_executions') == 0
-            then
-                -- setting capsul_connection to a string value will prevent
-                -- the creation of new connections because they will use
-                -- hset command that raises an error value type is not a hash
-                redis.call('set', 'capsul_connections', 'shutting down')
-                return true
-            else
-                return false
+        self._worker_started = self.redis.register_script('''
+            local function table_find(array, value)
+                for i, v in pairs(array) do
+                    if v == value then
+                        return i
+                    end
+                end
             end
-            ''')
 
-        self._update_workflow_parameters = self.redis.register_script('''
-            local function key(name)
-                return 'capsul:' .. ARGV[1] .. ':' .. name
-            end
+            local engine_key = KEYS[1]
 
-            local execution_id = ARGV[1]
-            local parameters_location = cjson.decode(ARGV[2])
-            local output_parameters = cjson.decode(ARGV[3])
-            local workflow_parameters = cjson.decode(redis.call('get', key('workflow_parameters')))
+            local worker_id = ARGV[1]
 
-            local parameters = workflow_parameters['content']
-            for index, value in ipairs(parameters_location) do
-                local i = tonumber(value)
-                if i then
-                    parameters = parameters[i+1]
-                else
-                    parameters = parameters[value]
+            local workers = cjson.decode(redis.call('hget', engine_key, 'workers'))
+            table.insert(workers, worker_id)
+            redis.call('hset', engine_key, 'workers', cjson.encode(workers))
+        ''')
+
+        self._worker_ended = self.redis.register_script('''
+            local function table_find(array, value)
+                for i, v in pairs(array) do
+                    if v == value then
+                        return i
+                    end
                 end
             end
 
-            for k, v in pairs(output_parameters) do
-                workflow_parameters['proxy_values'][parameters[k][2]+1] = v
+            local engine_key = KEYS[1]
+
+            local worker_id = ARGV[1]
+
+            local workers = redis.call('hget', engine_key, 'workers')
+            if workers then
+                workers = cjson.decode(workers)
+                table.remove(workers, table_find(workers, worker_id))
+                redis.call('hset', engine_key, 'workers', cjson.encode(workers))
             end
-            redis.call('set', key('workflow_parameters'), cjson.encode(workflow_parameters))
+        ''')
+
+
+        self._full_report = self.redis.register_script('''
+            local execution_key = KEYS[1]
+
+            local label = redis.call('hget', execution_key, 'label')
+            local execution_context = redis.call('hget', execution_key, 'execution_context')
+            local workflow_parameters = redis.call('hget', execution_key, 'workflow_parameters')
+            local status = redis.call('hget', execution_key, 'status')
+            local error = redis.call('hget', execution_key, 'error')
+            local error_detail = redis.call('hget', execution_key, 'error_detail')
+            local start_time = redis.call('hget', execution_key, 'start_time')
+            local end_time = redis.call('hget', execution_key, 'end_time')
+            local waiting = cjson.decode(redis.call('hget', execution_key, 'waiting'))
+            local ready = cjson.decode(redis.call('hget', execution_key, 'ready'))
+            local ongoing = cjson.decode(redis.call('hget', execution_key, 'ongoing'))
+            local done = cjson.decode(redis.call('hget', execution_key, 'done'))
+            local failed = cjson.decode(redis.call('hget', execution_key, 'failed'))
+            local jobs = {}
+            local cursor = 0
+            repeat
+                local result = redis.call('hscan', execution_key, cursor, 'MATCH', 'job:*')
+                cursor = tonumber(result[1])
+                for i, job in ipairs(result[2]) do
+                    if i % 2 == 0 then
+                        table.insert(jobs, job)
+                    end
+                end
+            until cursor == 0
+
+            return {label, execution_context, workflow_parameters, 
+                status, error, error_detail, start_time, end_time, 
+                waiting, ready, ongoing, done, failed, jobs}                
             ''')
-      
-    def close(self):
-        self.redis.hdel('capsul_connections', self.uuid)
+
+
+        self._check_shutdown = self.redis.register_script('''
+            if redis.call('get', 'capsul:redis_pid_file') and
+               redis.call('hlen', 'capsul:connections') == 0
+            then
+                redis.call('set', 'capsul:shutting_down', 1)
+                return true
+            else
+                return false
+            end
+            ''')
+
+        return self
+    
+
+    def __exit__(self, exception_type, exception_value, exception_traceback):
+        self.redis.hdel('capsul:connections', self.uuid)
         self.check_shutdown()
+        del self.redis
    
-    def key(self, execution_id, name):
-        return f'capsul:{execution_id}:{name}'
 
-    def get(self, execution_id, name):
-        return self.redis.get(self.key(execution_id, name))
+    def connect_to_engine(self, engine):
+        engine_id = self.redis.hget('capsul:engine', engine.label)
+        if not engine_id:
+            # Create new engine in database
+            engine_id = str(uuid4())
+            # Engine configuration dictionary
+            self.redis.hset(f'capsul:{engine_id}', 'config', json.dumps(engine.config.json()))
+            # List of running workers
+            self.redis.hset(f'capsul:{engine_id}', 'workers', '[]')
+            # List of executions_id
+            self.redis.hset(f'capsul:{engine_id}', 'executions', '[]')
+            # Create association between label and engine_id
+            self.redis.hset(f'capsul:{engine_id}', 'label', engine.label)
+            self.redis.hset('capsul:engine', engine.label, engine_id)
+        return engine_id
     
-    def set(self, execution_id, name, value):
-        self.redis.set(self.key(execution_id, name), value)
 
-    def pipeline(self, execution_id):
-        return RedisExecutionPipeline(self.redis.pipeline(), execution_id)
+    def engine_config(self, engine_id):
+        config = self.redis.hget(f'capsul:{engine_id}', 'config')
+        if config:
+            return json.loads(config)
+        raise ValueError(f'Invalid engine_id: {engine_id}')
+
+
+    def number_of_workers_to_start(self, engine_id):
+        config = self.engine_config(engine_id)
+        workers = json.loads(self.redis.hget(f'capsul:{engine_id}', 'workers'))
+        requested = config.get('start_workers', {}).get('count', 0)
+        return max(0, requested - len(workers))
+
+
+    def worker_database_config(self, engine_id):
+        return self.config
+
+
+    def worker_started(self, engine_id):
+        worker_id = str(uuid4())
+        keys = [
+            f'capsul:{engine_id}',
+        ]
+        args = [
+            worker_id,
+        ]
+        self._worker_started(keys=keys, args=args)
+        return worker_id
+
+
+    def worker_ended(self, engine_id, worker_id):
+        keys = [
+            f'capsul:{engine_id}',
+        ]
+        args = [
+            worker_id,
+        ]
+        self._worker_ended(keys=keys, args=args)
+
+
+    def dispose_engine(self, engine_id):
+        label = self.redis.hget(f'capsul:{engine_id}', 'label')
+        if label:
+            # Removes association between label and engine_id
+            self.redis.hdel('capsul:engine', label)
+            self.redis.hdel(f'capsul:{engine_id}', 'label')
+            # Check if some executions had been submited or are ongoing
+            # An empty list modified with Redis Lua scripts may be encoded as empty dict
+            if self.redis.hget(f'capsul:{engine_id}', 'executions') in ('{}', '[]'):
+                # Nothing is ongoing, completely remove engine
+                self.redis.delete(f'capsul:{engine_id}')
+
 
     def store_execution(self,
+            engine_id,
+            label,
             start_time, 
             executable_json,
             execution_context_json,
@@ -334,107 +496,99 @@ def store_execution(self,
             waiting
         ):
         execution_id = str(uuid4())
-        pipe = self.pipeline(execution_id)
-        pipe.pipeline.hset('capsul_ongoing_executions', execution_id, datetime.now().isoformat())
-        pipe.pipeline.hset('capsul_undisposed_executions', execution_id, datetime.now().isoformat())
-        pipe.set( 'status', ('ready' if ready else 'ended'))
-        pipe.set('start_time', start_time)
-        pipe.set('executable', json.dumps(executable_json))
-        pipe.set('execution_context', json.dumps(execution_context_json))
-        pipe.set('workflow_parameters', json.dumps(workflow_parameters_json))
-        pipe.set_list('ready', ready)
-        pipe.set_list('waiting', waiting)
-        pipe.delete('ongoing')
-        pipe.delete('done')
-        pipe.delete('failed')
-        if not ready:
-            pipe.set('end_time', start_time)
-        for job in jobs:
-            pipe.hset('jobs', job['uuid'], json.dumps(job))
-        pipe.execute()
+        keys = [
+            f'capsul:{engine_id}',
+            f'capsul:{engine_id}:{execution_id}'
+        ]
+        args = [
+            execution_id,
+            label,
+            start_time,
+            json.dumps(executable_json),
+            json.dumps(execution_context_json),
+            json.dumps(workflow_parameters_json),
+            json.dumps(jobs),
+            json.dumps(ready),
+            json.dumps(waiting),
+        ]
+        self._store_execution(keys=keys, args=args)
         return execution_id
-    
-    def status(self, execution_id):
-        return self.get(execution_id, 'status')
 
-    def error(self, execution_id):
-        return self.get(execution_id, 'error')
 
-    def error_detail(self, execution_id):
-        return self.get(execution_id, 'error_detail')
+    def execution_context_json(self, engine_id, execution_id):
+        return json.loads(self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'execution_context'))
 
-    def set_error(self, execution_id, error, error_detail=None):
-        pipe = self.pipeline(execution_id)
-        pipe.set( 'error', error)
-        pipe.set( 'error_detail', error_detail)
-        pipe.execute()
-    
-    def execution_context_json(self, execution_id):
-        return json.loads(self.get(execution_id, 'execution_context'))
 
-    def executable_json(self, execution_id):
-        return json.loads(self.get(execution_id, 'executable'))
-    
-    def workflow_parameters_json(self, execution_id):
-        return json.loads(self.get(execution_id, 'workflow_parameters'))
-
-    def set_workflow_parameters_json(self, execution_id, workflow_parameters_json):
-        self.set(execution_id, 'workflow_parameters', json.dumps(workflow_parameters_json))
-
-    def update_workflow_parameters_json(self, execution_id, parameters_location, output_values):
-        self._update_workflow_parameters(args=[execution_id, json.dumps(parameters_location), json.dumps(output_values)])
-    
-    def start_time_json(self, execution_id):
-        return self.get(execution_id, 'start_time')
-
-    def end_time_json(self, execution_id):
-        return self.get(execution_id, 'end_time')
-
-    def jobs_json(self, execution_id):
-        jobs_key = self.key(execution_id, 'jobs')
-        for job_uuid in self.redis.hkeys(jobs_key):
-            yield json.loads(self.redis.hget(jobs_key, job_uuid))
+    def pop_job_json(self, engine_id, start_time):
+        executions = self.redis.hget(f'capsul:{engine_id}','executions')
+        if executions is None:
+            # engine_id does not exists anymore
+            # return None to say to workers that they can die
+            return None, None
+        else:
+            for execution_id in json.loads(executions):
+                keys = [
+                    f'capsul:{engine_id}:{execution_id}'
+                ]
+                args=[start_time]
+                job_uuid = self._pop_job(keys=keys, args=args)
+                if job_uuid:
+                    return execution_id, job_uuid
+            # Empty string means "no job ready yet"
+            return '', ''
+
+    def job_finished_json(self, engine_id, execution_id, job_uuid, 
+                          end_time, returncode, stdout, stderr):
+        keys = [
+            f'capsul:{engine_id}',
+            f'capsul:{engine_id}:{execution_id}'
+        ]
+        args = [
+            execution_id, 
+            job_uuid, 
+            end_time, 
+            returncode, 
+            stdout, 
+            stderr
+        ]
+        self._job_finished(keys=keys, args=args)
+
+
+    def status(self, engine_id, execution_id):
+        return self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'status')
 
+        
+    def workflow_parameters_json(self, engine_id, execution_id):
+        return json.loads(self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'workflow_parameters') or 'null')
 
-    def job_json(self, execution_id, job_uuid):
-        job = json.loads(self.redis.hget(self.key(execution_id, 'jobs'), job_uuid))
-        return job
 
-    def ready(self ,execution_id):
-        return self.get_list(execution_id, 'ready')
+    def set_workflow_parameters_json(self, engine_id, execution_id, workflow_parameters_json):
+        self.redis.hset(f'capsul:{engine_id}:{execution_id}', 'workflow_parameters',
+                        json.dumps(workflow_parameters_json))
 
-    def waiting(self ,execution_id):
-        return self.get_list(execution_id, 'waiting')
 
-    def ongoing(self ,execution_id):
-        return self.get_list(execution_id, 'ongoing')
+    def update_workflow_parameters_json(self, engine_id, execution_id, parameters_location, output_values):
+        keys = [
+            f'capsul:{engine_id}:{execution_id}'
+        ]
+        args = [
+            json.dumps(parameters_location),
+            json.dumps(output_values)
+        ]
+        self._update_workflow_parameters(keys=keys, args=args)
 
-    def done(self ,execution_id):
-        return self.get_list(execution_id, 'done')
 
-    def failed(self ,execution_id):
-        return self.get_list(execution_id, 'failed')
-   
-    def start_one_job_json(self, execution_id, start_time):
-        job = self._start_one_job(args=[execution_id, start_time])
-        if job is not None:
-            job = json.loads(job)
+    def job_json(self, engine_id, execution_id, job_uuid):
+        job = json.loads(self.redis.hget(f'capsul:{engine_id}:{execution_id}', f'job:{job_uuid}'))
         return job
-    
-    def job_finished_json(self, execution_id, job_uuid, end_time, returncode, stdout, stderr):
-        r = self._job_finished(args=[execution_id, job_uuid, end_time, returncode, stdout, stderr])
-        if isinstance(r, str) and os.path.exists(r):
-            shutil.rmtree(r)
-            return True
-        else:
-            return bool(r)
 
-    def execution_report_json(self, execution_id):
-        (executable, execution_context, workflow_parameters, status, error,
+
+    def execution_report_json(self, engine_id, execution_id):
+        (label, execution_context, workflow_parameters, status, error,
          error_detail, start_time, end_time, waiting, ready,
-         ongoing, done, failed, jobs) = self._full_report(args=[execution_id])
+         ongoing, done, failed, jobs) = self._full_report(keys=[f'capsul:{engine_id}:{execution_id}'])
         result = dict(
-            executable=json.loads(executable),
+            label=label,
             execution_context=json.loads(execution_context),
             status=status,
             error=error,
@@ -451,19 +605,36 @@ def execution_report_json(self, execution_id):
             engine_debug = {},
         )
         return result
-        
-    def dispose(self, execution_id):
-        tmp = self._dispose(args=[execution_id])
-        if tmp and os.path.exists(tmp):
-            shutil.rmtree(tmp)
     
+
+    def dispose(self, engine_id, execution_id):
+        keys = [
+            f'capsul:{engine_id}',
+            f'capsul:{engine_id}:{execution_id}'
+        ]
+        args = [
+            execution_id,
+        ]
+        self._dispose(keys=keys, args=args)
+
+
     def check_shutdown(self):
-        pipe = self.redis.pipeline()
-        pipe.hlen('capsul_connections')
         if self._check_shutdown():
-            tmp = self.redis.get('capsul_redis_tmp')
-            pid_file = self.redis.get('capsul_redis_pid_file')
-            
+            pipeline = self.redis.pipeline()
+            pipeline.get('capsul:redis_tmp')
+            pipeline.delete('capsul:redis_tmp')
+            tmp = pipeline.execute()[0]
+            pid_file = self.redis.get('capsul:redis_pid_file')
+            self.redis.delete('capsul:redis_pid_file')
+
+            keys = self.redis.keys('capsul:*')
+            if keys == ['capsul:shutting_down']:
+                # Nothing in the database, do not save it
+                save= False
+            else:
+                save = True
+                self.redis.save()
+
             # Kill the redis server
             self.redis.shutdown()
 
@@ -474,11 +645,35 @@ def check_shutdown(self):
                 time.sleep(0.1)
 
             shutil.rmtree(tmp)
+            if not save and os.path.exists(self.path):
+                os.remove(self.path)
+
     
-    def set_tmp(self, execution_id, tmp):
-        self.set(execution_id, 'tmp', tmp)
+    def start_execution(self, engine_id, execution_id, tmp):
+        self.redis.hset(f'capsul:{engine_id}:{execution_id}', 'tmp', tmp)
+        self.redis.hset(f'capsul:{engine_id}:{execution_id}', 'status', 'running')
+
+
+    def end_execution(self, engine_id, execution_id):
+        self.redis.hset(f'capsul:{engine_id}:{execution_id}', 'status', 'ended')
+        tmp = self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'tmp')
+        self.redis.hdel(f'capsul:{engine_id}:{execution_id}', 'tmp')
+        if self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'dispose'):
+            executions = json.loads(self.redis.hget(f'capsul:{engine_id}', 'executions'))
+            executions.remove(execution_id)
+            self.redis.hset(f'capsul:{engine_id}', 'executions', json.dumps(executions))
+            self.redis.delete(f'capsul:{engine_id}:{execution_id}')
+            if not executions and not self.redis.hget(f'capsul:{engine_id}', 'label'):
+                # Engine is already disopsed: delete it
+                self.redis.delete(f'capsul:{engine_id}')
+        return tmp
+
+
+    def tmp(self, engine_id, execution_id):
+        return self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'tmp')
+
 
-    def tmp(self, execution_id):
-        return self.get(execution_id, 'tmp')
+    def error(self, engine_id, execution_id):
+        return self.redis.hget(f'capsul:{engine_id}:{execution_id}', 'error')
 
     
diff --git a/capsul/dataset.py b/capsul/dataset.py
index a639d9e98..5fd73a978 100644
--- a/capsul/dataset.py
+++ b/capsul/dataset.py
@@ -7,19 +7,21 @@
 '''
 
 import csv
+import fnmatch
 import functools
 import json
 import operator
 from pathlib import Path
 import re
-import fnmatch
-from capsul.pipeline.pipeline import Process, Pipeline, Switch
+import sys
+
+from capsul.pipeline.pipeline import Process, Pipeline
 from capsul.pipeline.process_iteration import ProcessIteration
 
 from soma.controller import Controller, Literal, Directory
 from soma.undefined import undefined
 
-
+global_debug = False
 
 class Dataset(Controller):
     '''
@@ -408,69 +410,85 @@ def __init_subclass__(cls, schema, process) -> None:
         schemas[schema] = cls
         cls.schema = schema
         setattr(process, 'metadata_schemas', schemas)
+                    
+
+class MetadataModifier:
+    def __init__(self, schema_name, process, parameter):
+        self.process = process
+        self.parameter = parameter
+        self.modifiers = []
+
+        process_schema = getattr(process, 'metadata_schemas', {}).get(schema_name)
+        if process_schema:
+            for pattern, modifier in getattr(process_schema, '_', {}).items():
+                if fnmatch.fnmatch(parameter, pattern):
+                    self.add_modifier(modifier)
+            modifier = getattr(process_schema, parameter, None)
+            self.add_modifier(modifier)
+        pipeline = process.get_pipeline()
+        if pipeline:
+            pipeline_schema = getattr(pipeline, 'metadata_schemas', {}).get(schema_name)
+            if pipeline_schema:
+                nodes_schema = getattr(pipeline_schema, '_nodes', None)
+                if nodes_schema:
+                    # node_modifiers = nodes_schema.get(process.name)
+                    for pattern, node_modifiers in nodes_schema.items():
+                        if fnmatch.fnmatch(process.name, pattern):
+                            for pattern, modifier in node_modifiers.items():
+                                if fnmatch.fnmatch(parameter, pattern):
+                                    self.add_modifier(modifier)
+
+    def __repr__(self):
+        return f'MetadataModifier({self.process.label}, {self.parameter}, {self.modifiers})'
     
-    @classmethod
-    def _update_metadata(cls, metadata, process, pipeline_name, parameter,
-                         iteration_index):
-        # remove top pipeline name
-        if '.' in pipeline_name:
-            pipeline_name = pipeline_name.split('.', 1)[-1]
-            long_parmater = f'{pipeline_name}.{parameter}'
+    @property
+    def is_empty(self):
+        return not self.modifiers
+    
+    def add_modifier(self, modifier):
+        if modifier is None:
+            return
+        if isinstance(modifier, dict) or callable(modifier):
+            self.modifiers.append(modifier)
         else:
-            pipeline_name = ''
-            long_parmater = parameter
-        stack = [v for p, v in getattr(cls, '_', {}).items()
-                 if fnmatch.fnmatch(long_parmater, p)] \
-            + [
-            #getattr(cls, f'{pipeline_name}.{parameter}', None),
-            getattr(cls, parameter, None),
-        ]
-        instance_process_schema = getattr(process, 'metadata_schemas', None)
-        if instance_process_schema:
-            stack += [v
-                      for p, v in getattr(instance_process_schema, '_',
-                                          {}).items()
-                      if fnmatch.fnmatch(long_parmater, p)]
-            #stack.append(getattr(instance_process_schema, f'{pipeline_name}.{parameter}', None))
-            stack.append(getattr(instance_process_schema, parameter, None))
+            raise ValueError(f'Invalid value for schema modification for parameter {self.parameter}: {modifier}')
         
-        while stack:
-            item = stack.pop(0)
-            if item is None:
-                continue
-            elif isinstance(item, list):
-                stack = item + stack
-            elif isinstance(item, dict):
-                for k, v in item.items():
+    def apply(self, metadata, process, parameter):
+        for modifier in self.modifiers:
+            if isinstance(modifier, dict):
+                for k, v in modifier.items():
                     if callable(v):
-                        v = v(metadata=metadata, 
-                              process=process,
-                              parameter=parameter,
-                              iteration_index=iteration_index)
-                    setattr(metadata, k, v)
+                        setattr(metadata, k, v(metadata=metadata, process=process, parameter=parameter))
+                    else:
+                        setattr(metadata, k, v)
             else:
-                stack.insert(0, item(
-                    metadata=metadata, 
-                    process=process,
-                    parameter=parameter,
-                    iteration_index=iteration_index))                
-                
+                modifier(metadata, process, parameter)
 
-    @staticmethod
-    def prepend(key, value, sep='_'):
-        def prepend(metadata, process, parameter, iteration_index,
-                    key=key, value=value,sep=sep):
-            current_prefix = metadata.get(key)
-            metadata[key] = value + (sep + current_prefix if current_prefix else '')
-        return prepend
+class Prepend:
+    def __init__(self, key, value, sep='_'):
+        self.key = key
+        self.value = value
+        self.sep = sep
+    
+    def __call__(self, metadata, process, parameter):
+        current_value = getattr(metadata, self.key, '')
+        setattr(metadata, self.key, self.value + (self.sep + current_value if current_value else ''))
+
+    def __repr__(self):
+        return f'Prepend({repr(self.key)}, {repr(self.value)}{("" if self.sep == "_" else ", sep=" + repr(self.sep))})'
+
+class Append:
+    def __init__(self, key, value, sep='_'):
+        self.key = key
+        self.value = value
+        self.sep = sep
+    
+    def __call__(self, metadata, process, parameter):
+        current_value = getattr(metadata, self.key, '')
+        setattr(metadata, self.key, (current_value + self.sep if current_value else '') + self.value)
 
-    @staticmethod
-    def append(key, value, sep='_'):
-        def append(metadata, process, parameter, iteration_index,
-                   key=key, value=value,sep=sep):
-            current_prefix = metadata.get(key)
-            metadata[key] = (current_prefix + sep if current_prefix else '') + value
-        return append
+    def __repr__(self):
+        return f'Append({repr(self.key)}, {repr(self.value)}{("" if self.sep == "_" else ", sep=" + repr(self.sep))})'
 
 
 def dprint(debug, *args, _frame_depth=1, **kwargs):
@@ -481,24 +499,28 @@ def dprint(debug, *args, _frame_depth=1, **kwargs):
             head = f'!{frame.function}:{frame.lineno}!'
         finally:
             del frame
-        print(head, *args, **kwargs)
+        print(head, ' '.join(f'{i}' for i in args), file=sys.stderr, **kwargs)
 
 
 def dpprint(debug, item):
     if debug:
         from pprint import pprint
-        dprint(debug=debug, _frame_depth=2)
+        dprint(debug=debug, _frame_depth=2, file=sys.stderr)
         pprint(item)
 
 
 class ProcessMetadata(Controller):
-    parameters_per_schema : dict[str, list[str]]
-    dataset_per_parameter : dict
 
-    def __init__(self, executable, execution_context, datasets=None):
+    def __init__(self, executable, execution_context, datasets=None, debug=False):
         super().__init__()
+        self.executable = executable
+        self.execution_context = execution_context
+        self.datasets = datasets
+        self._current_iteration = None
+        self.debug=debug
 
         self.parameters_per_schema = {}
+        self.schema_per_parameter = {}
         self.dataset_per_parameter = {}
         self.iterative_schemas = set()
         if isinstance(executable, ProcessIteration):
@@ -510,41 +532,18 @@ def __init__(self, executable, execution_context, datasets=None):
 
         # Associate each field to a dataset
         for field in process.user_fields():
-            inner_field = None
-            inner_field_name = None
-            if isinstance(process, Pipeline):
-                inner_item = next(process.get_linked_items(
-                                    process, field.name), None)
-            else:
-                inner_item = None
-            if inner_item is not None:
-                inner_process, inner_field_name = inner_item
-                path_type = inner_process.field(inner_field_name).path_type
-            else:
-                path_type = field.path_type
-            if path_type:
-                # Associates a Dataset name with the field
-                dataset_name = getattr(field, 'dataset', None)
-                # fallback 1: get in inner_field (of an inner process)
-                if dataset_name is None and inner_field:
-                    dataset_name = getattr(inner_field, 'dataset', None)
-                # fallback 2: get manually given datasets
-                if dataset_name is None and datasets:
-                    dataset_name = datasets.get(field.name)
-                # fallback 3: use "input" or "output"
-                if dataset_name is None and (not datasets
-                                            or field.name not in datasets):
-                    dataset_name = ('output' if field.is_output() else 'input')
-                if dataset_name is None:
-                    # no completion for this field
-                    continue
-                dataset = getattr(execution_context.dataset, dataset_name, None)
-                if dataset:
-                    self.dataset_per_parameter[field.name] = dataset_name
-                    schema = dataset.metadata_schema
-                    self.parameters_per_schema.setdefault(schema, []).append(field.name)
-                    if field.name in iterative_parameters:
-                        self.iterative_schemas.add(schema)
+            dataset_name = self.parameter_dataset_name(executable, field)
+            if dataset_name is None:
+                # no completion for this field
+                continue
+            dataset = getattr(execution_context.dataset, dataset_name, None)
+            if dataset:
+                self.dataset_per_parameter[field.name] = dataset_name
+                schema = dataset.metadata_schema
+                self.parameters_per_schema.setdefault(schema, []).append(field.name)
+                self.schema_per_parameter[field.name] = schema
+                if field.name in iterative_parameters:
+                    self.iterative_schemas.add(schema)
         
         for schema_name in self.parameters_per_schema:
             schema_cls = Dataset.find_schema(schema_name)
@@ -557,9 +556,129 @@ def __init__(self, executable, execution_context, datasets=None):
                 self.add_field(schema_name, type_=schema_cls)
                 setattr(self, schema_name, schema_cls())
 
-    def generate_paths(self, executable, debug=False):
-        dprint(debug, executable.definition)
+        if self.debug:
+            self.dprint('Create ProcessMetadata for', self.executable.label)
+            for field in process.user_fields():
+                if field.path_type:
+                    dataset = self.dataset_per_parameter.get(field.name)
+                    schema = self.schema_per_parameter.get(field.name)
+                    self.dprint(f'  {field.name} -> dataset={dataset}, schema={schema}')
+            self.dprint('  Iterative schemas:', self.iterative_schemas)
+    
+    def dprint(self, *args, **kwargs):
+        if isinstance(self.debug, bool) and self.debug:
+            self.debug = sys.stderr
+        if self.debug:
+            print(*args, file=self.debug, **kwargs)
+    
+    def parameter_dataset_name(self, process, field):
+        '''
+        Find the name of the dataset associated to a process parameter
+        '''
+        inner_field = None
+        inner_field_name = None
+        if isinstance(process, Pipeline):
+            # inner_item = next(process.get_linked_items(
+            #                     process, field.name, direction=('links_from' if field.is_output() else 'links_to')), None)
+            inner_item = next(process.get_linked_items(
+                                process, field.name, in_outer_pipelines=False), None)
+        else:
+            inner_item = None
+        if inner_item is not None:
+            inner_process, inner_field_name = inner_item
+            path_type = inner_process.field(inner_field_name).path_type
+        else:
+            path_type = field.path_type
+        if path_type:
+            dataset_name = None
+            # 1: get manually given datasets
+            if self.datasets:
+                dataset_name = self.datasets.get(field.name)
+            # Associates a Dataset name with the field
+            if dataset_name is None:
+                dataset_name = getattr(field, 'dataset', None)
+            # fallback 1: get in inner_field (of an inner process)
+            if dataset_name is None and inner_field:
+                dataset_name = getattr(inner_field, 'dataset', None)
+            # fallback 3: use "input" or "output"
+            if dataset_name is None and (not self.datasets
+                                         or field.name not in self.datasets):
+                dataset_name = ('output' if field.is_output() else 'input')
+            return dataset_name
+        return None
+    
+    def metadata_modifications(self, process):
+        result = {}
+        if isinstance(process, ProcessIteration):
+            for iprocess in process.iterate_over_process_parmeters():
+                iresult = self.metadata_modifications(iprocess.process)
+                for k, v in iresult.items():
+                    if k in process.iterative_parameters:
+                        result.setdefault(k, []).append(v)
+                    else:
+                        result[k] = v
+        else:
+            for field in process.user_fields():                
+                if process.plugs[field.name].activated:
+                    self.dprint(f'  Parse schema modifications for {field.name}')
+                    schema = self.schema_per_parameter.get(field.name)
+                    if schema:
+                        todo = []
+                        done = set()
+                        if isinstance(process, Pipeline):
+                            stack = list(process.get_linked_items(process, field.name, in_outer_pipelines=True))
+                            while stack:
+                                node, node_parameter = stack.pop(0)
+                                self.dprint(f'    connected to {node.full_name}.{node_parameter}')
+                                todo.insert(0, (node, node_parameter))
+                                done.add(node)
+                                if field.is_output():
+                                    for node_field in node.user_fields():
+                                        if node.plugs[node_field.name].activated and not node_field.is_output():
+                                            ext = [i for i in 
+                                                   process.get_linked_items(node, 
+                                                                            node_field.name,
+                                                                            process_only=False,
+                                                                            direction='links_from')
+                                                  if isinstance(i[0], Process) and i[0] not in done]
+                                            stack.extend(ext)
+                                            done.update(i[0] for i in ext)
+                                            if self.debug:
+                                                for n, p in ext:
+                                                    self.dprint(f'        + {n.full_name}.{p}')
+                        todo.append((process, field.name))
+                        for node, node_parameter in todo:
+                            self.dprint(f'    resolve {node.name}.{node_parameter}')
+                            modifier = MetadataModifier(schema, node, node_parameter)
+                            if not modifier.is_empty:
+                                self.dprint(f'        {modifier.modifiers}')
+                                result.setdefault(field.name, []).append(modifier)
+                else:
+                    self.dprint(f'  {field.name} ignored (inactive)')
+        return result
+    
+    def get_schema(self, schema_name):
+        schema = getattr(self, schema_name)
+        if isinstance(schema, list):
+            if schema:
+                schema = schema[self._current_iteration]
+            else:
+                schema = None
+        return schema
+
+    def generate_paths(self, executable):
+        if self.debug:
+            if self._current_iteration is not None:
+                iteration = f'[{self._current_iteration}]'
+            else:
+                iteration = ''
+            self.dprint(f'Generate paths for {executable.name}{iteration}')
+            for field in executable.user_fields():
+                value = getattr(executable, field.name, undefined)
+                self.dprint('       ', field.name, '=', repr(value))
+
         if isinstance(executable, ProcessIteration):
+            empty_iterative_schema = set()
             iteration_size = 0
             for schema in self.iterative_schemas:
                 schema_iteration_size = len(getattr(self, schema))
@@ -569,160 +688,59 @@ def generate_paths(self, executable, debug=False):
                         first_schema = schema
                     elif iteration_size != schema_iteration_size:
                         raise ValueError(f'Iteration on schema {first_schema} has {iteration_size} element(s) which is not equal to schema {schema} ({schema_iteration_size} element(s))')
+                else:
+                    empty_iterative_schema.add(schema)
+            
+            for schema in empty_iterative_schema:
+                setattr(self, schema, [Dataset.find_schema(schema)() for i in range(iteration_size)])
+            
             iteration_values = {}
-            dprint(debug, f'{iteration_size}')
             for iteration in range(iteration_size):
-                metadatas = {}
-                for i in self.parameters_per_schema:
-                    if i in self.iterative_schemas:
-                        metadata_list = getattr(self, i)
-                        if metadata_list:
-                            metadatas[i] = metadata_list[iteration]
-                        else:
-                            metadatas[i] = Dataset.find_schema(i)()
-                    else:
-                        metadatas[i] = getattr(self, i)
-                self._generate_paths(executable=executable, metadatas=metadatas, iteration_index=iteration, debug=debug)
+                self._current_iteration = iteration
+                executable.select_iteration_index(iteration)
+                self.generate_paths(executable.process)
                 for i in executable.iterative_parameters:
                     if executable.field(i).path_type:
                         value = getattr(executable.process, i, undefined)
                         iteration_values.setdefault(i, []).append(value)
-            dpprint(debug, iteration_values)
             for k, v in iteration_values.items():
                 setattr(executable, k, v)
         else:
-            metadatas = dict((i, getattr(self, i)) for i in self.parameters_per_schema)
-            self._generate_paths(executable=executable, metadatas=metadatas, iteration_index=None, debug=debug)
-
-    def _generate_paths(self, executable, metadatas, iteration_index, debug):
-        dprint(debug, executable.definition, f'{iteration_index}')
-        for schema, parameters in self.parameters_per_schema.items():
-            for parameter in parameters:
-                dataset = self.dataset_per_parameter[parameter]
-                metadata = Dataset.find_schema(schema)(base_path=f'!{{dataset.{dataset}.path}}')
-                for other_schema in self.parameters_per_schema:
-                    if other_schema == schema:
-                        continue
-                    mapping_class = Dataset.find_schema_mapping(other_schema, schema)
-                    if not mapping_class:
+            for schema_name in self.parameters_per_schema:
+                for other_schema_name in self.parameters_per_schema:
+                    if other_schema_name == schema_name:
                         continue
-                    mapping_class.map_schemas(metadatas[other_schema], metadata)
-                schema_metadata = metadatas[schema]
-                for field in schema_metadata.fields():
-                    value = getattr(schema_metadata, field.name, None)
-                    if value is not None:
-                        #value = value % metadata.asdict()
-                        setattr(metadata, field.name, value)
-                dprint(debug, '->', parameter, ':', schema)
-                for path in self.follow_links(executable, parameter, debug=debug):
-                    dprint(debug, 'path length', len(path))
-                    # get all pattern rules along the path
-                    path_schemas = []
-                    for pipeline_name, process, process_parameter in path:
-                        metadata_schema = getattr(process, 'metadata_schemas',
-                                                  {}).get(schema)
-                        path_schemas.append(metadata_schema)
-                        if isinstance(process, ProcessIteration):
-                            metadata_schema = getattr(
-                                process.process, 'metadata_schemas',
-                                {}).get(schema)
-                            path_schemas.append(metadata_schema)
-
-                    # merge attributes
-                    level = len(path_schemas) - 1
-                    for pipeline_name, process, process_parameter in reversed(path):
-                        dprint(debug, 
-                               'pname:', pipeline_name, 
-                               ', def:', process.definition, 
-                               ', par:', process_parameter,
-                               ', lev:', level)
-                        if isinstance(process, ProcessIteration):
-                            metadata_schema = path_schemas[level]
-                            if metadata_schema:
-                                dprint(debug, 'iter meta_schema:', metadata_schema, level, iteration_index)
-                                metadata_schema._update_metadata(
-                                    metadata, executable, process.process.name,
-                                    process_parameter, iteration_index)
-                            level -= 1
-                        metadata_schema = path_schemas[level]
-                        if metadata_schema:
-                            metadata_schema._update_metadata(
-                                metadata, executable, pipeline_name,
-                                process_parameter, iteration_index)
-                        level -= 1
-                        # we must also call all parent pipelines schemas with
-                        # the current depth parameter name, since some upper-
-                        # level schemas may use them witht a lower-level name
-                        # (especially for pattern rules such as
-                        # "LeftPipeline.*'
-                        for ilevel in range(level, -1, -1):
-                            metadata_schema = path_schemas[level]
-                            if metadata_schema:
-                                metadata_schema._update_metadata(
-                                    metadata, executable, pipeline_name,
-                                    process_parameter, iteration_index)
-                dpprint(debug, metadata.asdict())
-                path = str(metadata.build_path()) % metadata.asdict()
-                dprint(debug, parameter, '=', path)
-                if isinstance(executable, ProcessIteration):
-                    setattr(executable.process, parameter, path)
-                else:
+                    source = self.get_schema(schema_name)
+                    dest = self.get_schema(other_schema_name)
+                    if source and dest:
+                        mapping = Dataset.find_schema_mapping(schema_name, other_schema_name)
+                        if mapping:
+                            mapping.map_schemas(source, dest)
+            if self.debug:
+                for field in self.fields():
+                    self.dprint('  Metadata for', field.name)
+                    metadata = self.get_schema(field.name)
+                    for f in metadata.fields():
+                        v = getattr(metadata, f.name, undefined)
+                        self.dprint('   ', f.name, '=', repr(v))
+            metadata_modifications = self.metadata_modifications(executable)
+            for schema, parameters in self.parameters_per_schema.items():
+                for parameter in parameters:
+                    self.dprint(f'  find value for {parameter} in schema {schema}')
+                    dataset = self.dataset_per_parameter[parameter]
+                    metadata = Dataset.find_schema(schema)(base_path=f'!{{dataset.{dataset}.path}}')
+                    s = self.get_schema(schema)
+                    if s:
+                        metadata.import_dict(s.asdict())
+                    for modifier in metadata_modifications.get(parameter, []):
+                        modifier.apply(metadata, executable, parameter)
+                    try:
+                        path = str(metadata.build_path())   
+                    except:
+                        path = undefined
+                    if self.debug:
+                        for field in metadata.fields():
+                            value = getattr(metadata, field.name, undefined)
+                            self.dprint(f'    {field.name} = {repr(value)}')
+                        self.dprint(f'    => {path}')
                     setattr(executable, parameter, path)
-
-
-    @staticmethod        
-    def follow_links(executable, parameter_name, debug=False):
-        dprint(debug, executable.definition, parameter_name)
-        main_field = executable.field(parameter_name)
-        if main_field.is_output():
-            direction = 'links_from'
-        else:
-            direction = 'links_to'
-        stack = [([], executable, main_field.name, [])]
-        done = set()
-        while stack:
-            node_name_path, node, field_name, path = stack.pop(0)
-            dprint(debug, 'p:', node_name_path, ', node:', str(node), ', field:', field_name, '...')
-            field = node.field(field_name)
-            if node in done or not node.activated:
-                continue
-            done.add(node)
-            plug = node.plugs.get(field.name)
-            followers = []
-            if isinstance(node, Switch):
-                if field.name != 'switch':
-                    for input_plug_name, output_plug_name in node.connections():
-                        dprint(debug, 'switch', field_name, input_plug_name, output_plug_name, not main_field.is_output())
-                        if not main_field.is_output():
-                            if field_name == input_plug_name:
-                                dprint(debug, 'append')
-                                next_node, next_plug = next(executable.get_linked_items(
-                                    node, output_plug_name))
-                                stack.append((node_name_path, next_node, next_plug, path))
-                        else:
-                            if field_name == output_plug_name:
-                                dprint(debug, 'append')
-                                next_node, next_plug = next(executable.get_linked_items(
-                                    node, input_plug_name))
-                                stack.append((node_name_path, next_node, next_plug, path))
-            elif isinstance(node, Process):
-                path.append(('.'.join(node_name_path + [node.name]), node, field.name))
-                if isinstance(node, Pipeline):
-                    node_name = node.name
-                else:
-                    node_name = None
-                for dest_plug_name, dest_node in (i[1:3] for i in getattr(plug, direction)):
-                    if dest_node in done or not dest_node.activated:
-                        continue
-                    followers.append((node_name, dest_node, dest_plug_name))
-                #if isinstance(node, ProcessIteration):
-                    #followers.append((node.process.name, node.process, field.name))
-            if followers:
-                for next_node_name, next_node, next_plug in followers:
-                    if next_node_name:
-                        next_node_name_path = node_name_path + [next_node_name]
-                    else:
-                        next_node_name_path = node_name_path
-                    stack.append((next_node_name_path, next_node, next_plug, path))
-            else:
-                yield path
diff --git a/capsul/engine/__init__.py b/capsul/engine/__init__.py
index dae9e2a81..fba350bb4 100644
--- a/capsul/engine/__init__.py
+++ b/capsul/engine/__init__.py
@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime
-import importlib
+import json
+import os
+import subprocess
 import sys
 
 from soma.controller import Controller
@@ -8,7 +10,7 @@
 from ..execution_context import CapsulWorkflow, ExecutionContext
 from ..config.configuration import ModuleConfiguration
 from ..dataset import Dataset
-from ..database import execution_database
+from ..database import engine_database
 
 
 def execution_context(engine_label, engine_config, executable):
@@ -67,39 +69,50 @@ def __init__(self, label, config):
         super().__init__()
         self.label = label
         self.config = config
-        workers_type = self.config.workers_type
-        try:
-            workers_module = importlib.import_module(
-                f'capsul.engine.{workers_type}')
-        except ImportError:
-            raise ValueError(f'engine type {workers_type} is not known.')
-        self.workers_class = getattr(workers_module, f'{workers_type.capitalize()}Workers')
 
     def __enter__(self):
-        self.database = execution_database(self.config.database_url)
-        self.workers = self.workers_class(self.label, self.config, self.database)
-        return self.workers
-
+        # Connect to the database
+        self.database = engine_database(self.config.database)
+        self.database.__enter__()
+        # Connect to the engine in the database. Adds the engine in
+        # the database if it does not exist.
+        self.engine_id = self.database.connect_to_engine(self)
+
+        # Starts workers if necessary
+        self.start_workers()
+        return self
+
+    def start_workers(self):
+        db_config = self.database.worker_database_config(self.engine_id)
+        env = os.environ.copy()
+        env['CAPSUL_WORKER_DATABASE'] = json.dumps(db_config)
+        for i in range(self.database.number_of_workers_to_start(self.engine_id)):
+            workers_command = self.database.workers_command(self.engine_id)
+            try:
+                subprocess.run(
+                    workers_command,
+                    capture_output=False,
+                    check=True,
+                    env=env
+                )
+            except Exception as e:
+                quote = lambda x: f"'{x}'"
+                raise RuntimeError(f'Command failed: {" ".join(quote(i) for i in workers_command)}') from e
+        
+       
     def __exit__(self, exception_type, exception_value, exception_traceback):
-        del self.workers
-        self.database.close()
-        del self.database
+        self.database.dispose_engine(self.engine_id)
+        self.database.__exit__(exception_type, exception_value, exception_traceback)
+        del self.engine_id
     
     def execution_context(self, executable):
         return execution_context(self.label, self.config, executable)
 
-
-class Workers(Controller):
-    def __init__(self, engine_label, engine_config, database):
-        self.engine_label = engine_label
-        self.engine_config = engine_config
-        self.database = database
-
-    def start(self, executable, **kwargs):
+    def start(self, executable, debug=False, **kwargs):
         for name, value in kwargs.items():
             setattr(executable, name, value)
-        econtext = execution_context(self.engine_label, self.engine_config, executable)
-        workflow = CapsulWorkflow(executable)
+        econtext = execution_context(self.label, self.config, executable)
+        workflow = CapsulWorkflow(executable, debug=debug)
         # from pprint import pprint
         # print('!start!', flush=True)
         # pprint(workflow.jobs)
@@ -108,41 +121,39 @@ def start(self, executable, **kwargs):
         # pprint(workflow.parameters.no_proxy())
         # print('----')
         # pprint(workflow.jobs)
-        execution_id = self.database.new_execution(executable, econtext, workflow, start_time=datetime.now())
-        self._start(execution_id)
+        execution_id = self.database.new_execution(executable, self.engine_id, econtext, workflow, start_time=datetime.now())
         return execution_id
 
-    def _start(self, execution_id):
-        raise NotImplementedError(
-            '_start must be implemented in Workers subclasses.')
-
-    def debug_info(self, execution_id):
-        raise NotImplementedError(
-            'debug_info must be implemented in Workers subclasses.')
 
     def status(self, execution_id):
-        return self.database.status(execution_id)
+        return self.database.status(self.engine_id, execution_id)
     
-    def wait(self, *args, **kwargs):
-        self.database.wait(*args, **kwargs)
+
+    def wait(self, execution_id, *args, **kwargs):
+        self.database.wait(self.engine_id, execution_id, *args, **kwargs)
+
 
     def raise_for_status(self, *args, **kwargs):
-        self.database.raise_for_status(*args, **kwargs)
+        self.database.raise_for_status(self.engine_id, *args, **kwargs)
+
 
     def execution_report(self, *args, **kwargs):
-        return self.database.execution_report(*args, **kwargs)
+        return self.database.execution_report(self.engine_id, *args, **kwargs)
+
 
-    def print_execution_report(self, *args, **kwargs):
-        self.database.print_execution_report(*args, **kwargs)
+    def print_execution_report(self, engine_id, *args, **kwargs):
+        self.database.print_execution_report(engine_id, *args, **kwargs)
 
     def update_executable(self, *args, **kwargs):
-        self.database.update_executable(*args, **kwargs)
+        self.database.update_executable(self.engine_id, *args, **kwargs)
+
 
     def dispose(self, *args, **kwargs):
-        self.database.dispose(*args, **kwargs)
+        self.database.dispose(self.engine_id, *args, **kwargs)
+
 
-    def run(self, executable, timeout=None, print_report=False, **kwargs):
-        execution_id = self.start(executable, **kwargs)
+    def run(self, executable, timeout=None, print_report=False, debug=False, **kwargs):
+        execution_id = self.start(executable, debug=debug, **kwargs)
         try:
             try:
                 self.wait(execution_id, timeout=timeout)
@@ -153,8 +164,15 @@ def run(self, executable, timeout=None, print_report=False, **kwargs):
             self.raise_for_status(execution_id)
             if print_report:
                 self.print_execution_report(self.execution_report(execution_id), file=sys.stdout)
-            self.update_executable(executable, execution_id)
+            self.update_executable(execution_id, executable)
         finally:
             self.dispose(execution_id)
         return status
+
+class Workers(Controller):
+    def __init__(self, engine_label, engine_config, database):
+        self.engine_label = engine_label
+        self.engine_config = engine_config
+        self.database = database
+
     
diff --git a/capsul/engine/builtin.py b/capsul/engine/builtin.py
index 87152f0aa..7b70422d6 100644
--- a/capsul/engine/builtin.py
+++ b/capsul/engine/builtin.py
@@ -1,79 +1,70 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime
+import json
 import os
 import shutil
-import subprocess
 import sys
+import time
 import tempfile
-import traceback
 
-from capsul.database import execution_database
+from capsul.database import engine_database
+from capsul.run import run_job
 
-
-from . import Workers
-
-      
-class BuiltinWorkers(Workers): 
-    def _start(self, execution_id):
-        subprocess.run(
-            [sys.executable, '-m', 'capsul.engine.builtin', self.database.url, execution_id],
-            capture_output=False,
-            check=True
-        )
-        
-    def _debug_info(self, execution_id):
-        return {}
-    
+          
 if __name__ == '__main__':
-    if len(sys.argv) != 3:
-        raise ValueError('This command must be called with two '
-            'parameters: a database URL and an execution_id')
-    database_url = sys.argv[1]
-    execution_id = sys.argv[2]
-
+    if len(sys.argv) != 2:
+        raise ValueError('This command must be called with one '
+            'parameter: an engine id')
+    engine_id = sys.argv[1]
+    db_config = os.environ.get('CAPSUL_WORKER_DATABASE')
+    if not db_config:
+        raise ValueError('Worker command requires CAPSUL_WORKER_DATABASE to be set')
     # Really detach the process from the parent.
     # Whthout this fork, performing Capsul tests shows warning
     # about child processes not properly waited for.
-    if sys.platform.startswith('win'):
-        pid = 0
-    else:
-        pid = os.fork()
+    pid = os.fork()
     if pid == 0:
-        if not sys.platform.startswith('win'):
-            os.setsid()
-        tmp = tempfile.mkdtemp(prefix='caspul_builtin_')
-        try:
-            database = execution_database(database_url)
+        os.setsid()
+        db_config = json.loads(db_config)
+        with engine_database(db_config) as database:
+            worker_id = database.worker_started(engine_id)
+            # print(f'!worker {worker_id}! started', engine_id)
             try:
-                env = os.environ.copy()
-                env['CAPSUL_DATABASE'] = database_url
-                env['CAPSUL_EXECUTION_ID'] = execution_id
-                database.set_tmp(execution_id, tmp)
-                job = database.start_one_job(execution_id, start_time=datetime.now())
-                while job is not None:
-                    command = job['command']
-                    if command is not None:
-                        result = subprocess.run(command, env=env, capture_output=True)
-                        returncode = result.returncode
-                        stdout = result.stdout.decode()
-                        stderr = result.stderr.decode()
+                execution_id, job_uuid = database.pop_job(engine_id, start_time=datetime.now())
+                while job_uuid is not None:
+                    if not job_uuid:
+                        # Empty string means no job available yet
+                        time.sleep(0.2)
+                    elif job_uuid == 'start_execution':
+                        # print(f'!worker {worker_id}! start', execution_id)
+                        # This part is done before the processing of any job
+                        tmp = os.path.join(tempfile.gettempdir(), f'capsul_execution_{execution_id}')
+                        os.mkdir(tmp)
+                        try:
+                            database.start_execution(engine_id, execution_id, tmp)
+                        except Exception:
+                            os.rmdir(tmp)
+                    elif job_uuid == 'end_execution':
+                        # print(f'!worker {worker_id}! end', execution_id)
+                        tmp = database.end_execution(engine_id, execution_id)
+                        if tmp and os.path.exists(tmp):
+                            shutil.rmtree(tmp)
                     else:
-                        returncode = stdout = stderr = None
-                    all_done = database.job_finished(execution_id, job['uuid'], 
-                        end_time=datetime.now(),
-                        returncode=returncode,
-                        stdout=stdout,
-                        stderr=stderr)
-                    if all_done:
-                        break
-                    job = database.start_one_job(execution_id, start_time=datetime.now())
-            except Exception as e:
-                database.set_error(execution_id,
-                    error=f'Builtin engine loop failure: {e}',
-                    error_detail=f'{traceback.format_exc()}'
-                )
-                raise
+                        returncode, stdout, stderr = run_job(
+                            database,
+                            engine_id,
+                            execution_id,
+                            job_uuid,
+                            same_python=True,
+                            debug=True,
+                        )
+                        # print(f'!worker {worker_id}! job', execution_id, job_uuid, database.job_finished)
+                        database.job_finished(engine_id, execution_id, job_uuid, 
+                            end_time=datetime.now(),
+                            returncode=returncode,
+                            stdout=stdout,
+                            stderr=stderr)
+                    execution_id, job_uuid = database.pop_job(engine_id, start_time=datetime.now())
             finally:
-                database.close()
-        finally:
-            shutil.rmtree(tmp)
+                # print(f'!worker {worker_id}! ended' )
+                database.worker_ended(engine_id, worker_id)
diff --git a/capsul/engine/celery.py b/capsul/engine/celery.py
deleted file mode 100644
index 4bf9da4e2..000000000
--- a/capsul/engine/celery.py
+++ /dev/null
@@ -1,126 +0,0 @@
-# -*- coding: utf-8 -*-
-from datetime import datetime
-from glob import glob
-import os
-import shutil
-import subprocess
-import tempfile
-import sys
-
-# Remove a warning message: see https://github.com/celery/kombu/issues/1339
-import warnings
-warnings.filterwarnings('ignore', 'SelectableGroups dict interface')
-from celery import Celery
-
-from ..database import execution_database
-from ..database.redis import RedisExecutionDatabase
-from . import Workers
-
-shutdown_countdown = 10
-
-database_url = os.environ.get('CAPSUL_DATABASE')
-if database_url:
-    celery_app = Celery('capsul.engine.celery', broker=database_url)
-
-    @celery_app.task(bind=True, ignore_result=True)
-    def check_shutdown(self):
-        print('!check_shutdown!')
-        global database_url
-        celery_app.control.revoke(self.request.id) # prevent this task from being executed again
-        try:
-            database = execution_database(database_url)
-            try:
-                executions_count = database.redis.llen('capsul_ongoing_executions')
-                print(f'!executions_count! {executions_count}')
-                #TODO: possible race condition here if a connection to the celery workers
-                # is done right now
-                if not executions_count:
-                    # Release special database connection
-                    print('!shutting down celery!')
-                    database.redis.hdel('capsul_connections', 'celery_workers_connection')
-                    celery_app.control.shutdown() # send shutdown signal to all workers
-                    return
-            finally:
-                database.close()
-        except Exception:
-            pass
-        celery_app.send_task('capsul.engine.celery.check_shutdown', countdown=shutdown_countdown)
-
-
-
-    @celery_app.task(ignore_result=True)
-    def execute_process(execution_id, job_uuid):
-        global database_url
-        
-        database = execution_database(database_url)
-        try:
-            job = database.job(execution_id, job_uuid)
-            print(f"execute_process {job.get('process', {}).get('definition')} (execution_id={execution_id}, job {job_uuid})")
-            command = job['command']
-            if command is not None:
-                env = os.environ.copy()
-                env['CAPSUL_DATABASE'] = database_url
-                env['CAPSUL_EXECUTION_ID'] = execution_id
-                result = subprocess.run(command, env=env, capture_output=True)
-                returncode = result.returncode
-                stdout = result.stdout.decode()
-                stderr = result.stderr.decode()
-            else:
-                returncode = stdout = stderr = None
-            database.job_finished(execution_id, job['uuid'], 
-                end_time=datetime.now(),
-                returncode=returncode,
-                stdout=stdout,
-                stderr=stderr)
-            job = database.start_one_job(execution_id, start_time=datetime.now())
-            while job is not None:
-                # print(f"  push job {job.get('process', {}).get('definition')} (job {job_uuid})")
-                celery_app.send_task('capsul.engine.celery.execute_process', args=(execution_id, job['uuid'],))
-                job = database.start_one_job(execution_id, start_time=datetime.now())
-        finally:
-            database.close()
-
-
-class CeleryWorkers(Workers):
-    def _start(self, execution_id):
-        if not isinstance(self.database, RedisExecutionDatabase):
-            raise TypeError('Celery workers can only work with a Redis execution database')
-        tmp = self.database.redis.get('capsul_redis_tmp')
-        workers_pid_file = f'{tmp}/capsul_celery_workers.pid'
-        if not os.path.exists(workers_pid_file):
-            env = os.environ.copy()
-            env.update({
-                'CAPSUL_DATABASE': self.database.url,
-            })
-            cmd = [
-                sys.executable, '-m', 'celery',
-                '-A', 'capsul.engine.celery',
-                'multi', 'start',
-                'capsul_celery_workers',
-                f'--pidfile={tmp}/%n.pid',
-                f'--logfile={tmp}/%n%I.log'
-            ]
-            subprocess.Popen(cmd, env=env, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
-            celery_app = Celery('capsul.engine.celery', broker=self.database.url)
-            celery_app.send_task('capsul.engine.celery.check_shutdown', countdown=shutdown_countdown)
-            # keep a special database connection to prevent the database to shutdown before Celery workers
-            self.database.redis.hset('capsul_connections', 'celery_workers_connection', datetime.now().isoformat())
-        else:
-            celery_app = Celery('capsul.engine.celery', broker=self.database.url)
-        tmp = tempfile.mkdtemp(prefix='caspul_celery_')
-        try:
-            self.database.set_tmp(execution_id, tmp)
-            job = self.database.start_one_job(execution_id, start_time=datetime.now())
-            while job is not None:
-                celery_app.send_task('capsul.engine.celery.execute_process', args=(execution_id, job['uuid'],))
-                job = self.database.start_one_job(execution_id, start_time=datetime.now())
-        except Exception:
-            shutil.rmtree(tmp)
-
- 
-    def _cleanup(self, execution_directory):
-        pass
-
-    def _debug_info(self, execution_id):
-        #TODO: return the workers log
-        return {}
diff --git a/capsul/execution_context.py b/capsul/execution_context.py
index 4e1ff36c3..fc020c115 100644
--- a/capsul/execution_context.py
+++ b/capsul/execution_context.py
@@ -85,9 +85,8 @@ class CapsulWorkflow(Controller):
     parameters: DictWithProxy
     jobs: dict
     
-    def __init__(self, executable):
+    def __init__(self, executable, debug=False):
         super().__init__()
-        self.find_temporary_to_generate(executable)
         self.parameters = DictWithProxy(all_proxies=True)
         self.jobs = {}
         jobs_per_process = {}
@@ -101,6 +100,7 @@ def __init__(self, executable):
             process=executable,
             parent_executables=[],
             parameters_location=[],
+            process_iterations={},
             disabled=False)
         self.parameters.content.update(job_parameters.content)
 
@@ -112,17 +112,17 @@ def __init__(self, executable):
                         aj = self.jobs[after_job]
                         aj['wait_for'].add(before_job)
                         bj = self.jobs[before_job]
-                        if bj['command'] is None:
+                        if bj['disabled']:
                             bj['waited_by'].add(after_job)
 
         # Resolve disabled jobs
-        disabled_jobs = [(uuid, job) for uuid, job in self.jobs.items() if job['command'] is None]
+        disabled_jobs = [(uuid, job) for uuid, job in self.jobs.items() if job['disabled']]
         for disabled_job in disabled_jobs:
             wait_for = set()
             stack = disabled_job[1]['wait_for']
             while stack:
                 job = stack.pop()
-                if self.jobs[job]['command'] is None:
+                if self.jobs[job]['disabled']:
                     stack.update(self.jobs[job]['wait_for'])
                 else:
                     wait_for.add(job)
@@ -130,7 +130,7 @@ def __init__(self, executable):
             stack = list(disabled_job[1]['waited_by'])
             while stack:
                 job = stack.pop(0)
-                if self.jobs[job]['command'] is None:
+                if self.jobs[job]['disabled']:
                     stack.extend(self.jobs[job]['waited_by'])
                 else:
                     waited_by.add(job)
@@ -154,6 +154,7 @@ def _create_jobs(self,
                      process,
                      parent_executables,
                      parameters_location,
+                     process_iterations,
                      disabled):
         parameters = self.parameters
         for index in parameters_location:
@@ -163,6 +164,7 @@ def _create_jobs(self,
         nodes = []
         nodes_dict = parameters.content.setdefault('nodes', {})
         if isinstance(process, Pipeline):
+            self.find_temporary_to_generate(executable, iteration_process=None)
             disabled_nodes = process.disabled_pipeline_steps_nodes()
             for node_name, node in process.nodes.items():
                 if (node is process 
@@ -180,8 +182,8 @@ def _create_jobs(self,
                     parent_executables=parent_executables + [process],
                     parameters_location=parameters_location + ['nodes',
                                                                node_name],
+                    process_iterations=process_iterations,
                     disabled=disabled or node in disabled_nodes)
-                # nodes_dict[node_name].content.update(job_parameters.content)
                 nodes.append(node)
             for field in process.user_fields():
                 for dest_node, plug_name in executable.get_linked_items(process, 
@@ -189,14 +191,17 @@ def _create_jobs(self,
                                                                         in_sub_pipelines=False):
                     if dest_node in disabled_nodes:
                         continue
-                    #print('field name:', field.name, ', plug:', plug_name, ', dest_node:', dest_node.name, dest_node.name in nodes_dict, type(dest_node), ', executable:', executable.name, ', process:', process.name)
-                    #print('parameters:', parameters.content)
                     parameters.content[field.name] = nodes_dict[dest_node.name][plug_name]
                     break
                 if field.is_output():
                     for dest_node_name, dest_plug_name, dest_node, dest_plug, is_weak in process.plugs[field.name].links_to:
                         if dest_node.activated and dest_node not in disabled_nodes and not dest_node.field(dest_plug_name).is_output():
-                            process_chronology.setdefault(dest_node.uuid, set()).add(process.uuid)
+                            if isinstance(dest_node, Pipeline):
+                                continue
+                            process_chronology.setdefault(
+                                process_iterations.get(dest_node.uuid, dest_node.uuid),
+                                set()).add(
+                                    process_iterations.get(process.uuid, process.uuid))
             for node in nodes:
                 for plug_name in node.plugs:
                     first = nodes_dict[node.name].get(plug_name)
@@ -228,8 +233,16 @@ def _create_jobs(self,
         elif isinstance(process, ProcessIteration):
             parameters['_iterations'] = []
             iteration_index = 0
+            if isinstance(process.process, Pipeline):
+                all_iterated_processes = [p for p in process.process.all_nodes() if isinstance(p, Process)]
+            else:
+                all_iterated_processes = [process.process]
             for inner_process in process.iterate_over_process_parmeters():
+                if isinstance(inner_process, Pipeline):
+                    self.find_temporary_to_generate(executable, iteration_process=process)
                 parameters['_iterations'].append({})
+                for p in all_iterated_processes:
+                    process_iterations.setdefault(p.uuid, []).append(str(iteration_index))
                 job_parameters = self._create_jobs(
                     executable=executable,
                     jobs_per_process=jobs_per_process,
@@ -238,38 +251,52 @@ def _create_jobs(self,
                     process=inner_process,
                     parent_executables=parent_executables + [process], 
                     parameters_location=parameters_location + ['_iterations', str(iteration_index)],
+                    process_iterations=process_iterations,
                     disabled=disabled)
                 for k, v in job_parameters.content.items():
                     if k in process.iterative_parameters:
                         parameters.content.setdefault(k, []).append(v)
                     elif k in process.regular_parameters:
                         parameters.content[k] = v
+                for p in all_iterated_processes:
+                    del process_iterations[p.uuid][-1]
                 iteration_index += 1
             if isinstance(executable, Pipeline):
                 for field in process.user_fields():
                     if field.is_output():
-                        for dest_node, plug_name in executable.get_linked_items(process, field.name):
-                            process_chronology.setdefault(dest_node.uuid, set()).add(process.uuid)
+                        for dest_node, plug_name in executable.get_linked_items(process, field.name, direction='links_to'):
+                            if isinstance(dest_node, Pipeline):
+                                continue
+                            process_chronology.setdefault(
+                                dest_node.uuid + ','.join(process_iterations.get(dest_node.uuid, [])),
+                                set()).add(
+                                    process.uuid + ','.join(process_iterations.get(process.uuid, []))
+                                )
         elif isinstance(process, Process):
             job_uuid = str(uuid4())
             if disabled:
                 self.jobs[job_uuid] = {
                     'uuid': job_uuid,
-                    'command': None,
+                    'disabled': True,
                     'wait_for': set(),
                     'waited_by': set(),
                 }
             else:
                 self.jobs[job_uuid] = {
                     'uuid': job_uuid,
-                    'command': ['python', '-m', 'capsul.run', 'process', job_uuid],
+                    'disabled': False,
                     'wait_for': set(),
                     'process': process.json(include_parameters=False),
                     'parameters_location': parameters_location
                 }
             for parent_executable in parent_executables:
-                jobs_per_process.setdefault(parent_executable.uuid, set()).add(job_uuid)
-            jobs_per_process.setdefault(process.uuid, set()).add(job_uuid)
+                jobs_per_process.setdefault(
+                    parent_executable.uuid + ','.join(process_iterations.get(parent_executable.uuid, [])),
+                    set()).add(job_uuid)
+            jobs_per_process.setdefault(
+                process.uuid + ','.join(process_iterations.get(process.uuid, [])),
+                set()).add(job_uuid)
+            # print('!create_job!', process.full_name)
             for field in process.user_fields():
                 if getattr(field, 'generate_temporary', False):
                     prefix = f'!{{dataset.tmp.path}}/{process.full_name}'
@@ -282,80 +309,127 @@ def _create_jobs(self,
                     value = f'{prefix}.{field.name}_{uuid}{suffix}'
                 else:
                     value = getattr(process, field.name, None)
+                # print('  ', field.name, '<-', repr(value), getattr(field, 'generate_temporary', False))
                 proxy = parameters.proxy(executable.json_value(value))
                 parameters[field.name] = proxy
-                if field.is_output() and isinstance(executable, Pipeline):
-                    for dest_node, plug_name in executable.get_linked_items(process, field.name):
-                        process_chronology.setdefault(dest_node.uuid, set()).add(process.uuid)
+                if field.is_output() and isinstance(executable, (Pipeline, ProcessIteration)):
+                    for dest_node, plug_name in executable.get_linked_items(process, field.name, direction='links_to'):
+                        if isinstance(dest_node, Pipeline):
+                            continue
+                        process_chronology.setdefault(
+                            dest_node.uuid + ','.join(process_iterations.get(dest_node.uuid, [])), 
+                            set()).add(
+                                process.uuid + ','.join(process_iterations.get(process.uuid, [])))
         return parameters
 
-    def find_temporary_to_generate(self, executable):
+    def find_temporary_to_generate(self, executable, iteration_process):
+        # print('!temporaries! ->', executable.label, iteration_process and iteration_process.label)
         if isinstance(executable, Pipeline):
-            nodes = executable.all_nodes()
+            nodes = executable.all_nodes(in_iterations=True)
+        elif isinstance(executable, ProcessIteration) and isinstance(executable.process, Pipeline):
+            nodes = executable.process.all_nodes(in_iterations=True)
         else:
             nodes = [executable]
         for node in nodes:
+            # print('!temporaries! initialize node', node.full_name)
             for field in node.user_fields():
-                field.generate_temporary = False
-        self._find_temporary_to_generate(executable)
-    
-    
-    def _find_temporary_to_generate(self, executable):
-        """Finds all output fields that are empty and linked to an input
-        and set their generate_temporary attribute to True (ans set it to
-        False for other fields).
-        """
-        if isinstance(executable, Pipeline):
-            for node in executable.nodes.values():
-                if node is executable:
-                    continue
-                if isinstance(node, NipypeProcess):
-                    #nipype processes do not use temporaries, they produce output
-                    # file names
-                    return
+                if (field.output or 
+                    not field.metadata('write', False) or
+                    not node.plugs[field.name].activated):
+                    field.generate_temporary = False
+                else:
+                    field.generate_temporary = True
+                if isinstance(node, ProcessIteration):
+                    node.process.field(field.name).generate_temporary = field.generate_temporary
+                # print('!temporaries!  ', field.name, '=', field.generate_temporary)
 
+        stack = [(executable, field) for field in executable.user_fields()]
+        while stack:
+            node, field = stack.pop(0)
+            # print('!temporaries! no temporary for', node.full_name, ':', field.name)
+            field.generate_temporary = False
+            if isinstance(node, ProcessIteration):
+                node.process.field(field.name).generate_temporary = field.generate_temporary
+            for node, parameter in executable.get_linked_items(
+                    node, field.name, direction='links_from', in_outer_pipelines=True):
                 if isinstance(node, ProcessIteration):
-                    iteration_size = node.iteration_size()
+                    stack.append((node.process, node.process.field(parameter)))
+                    # print('!temporaries!   + ', node.process.full_name, ':', parameter)
                 else:
-                    iteration_size = None
+                    stack.append((node, node.field(parameter)))
+                    # print('!temporaries!   + ', node.full_name, ':', parameter)
 
-                for plug_name, plug in node.plugs.items():
-                    value = getattr(node, plug_name, undefined)
-                    if not plug.activated or not plug.enabled:
-                        continue
-                    field = node.field(plug_name)
-                    if field.output or not field.metadata('write', False):
-                        continue
-                    if field.is_list() and field.path_type is not None:
-                        if value is not undefined and len([x for x in value if x in ('', undefined)]) == 0:
-                            continue
-                    elif value not in (undefined, '') \
-                            or (field.path_type is None
-                                or len(plug.links_to) == 0):
-                        continue
-                    # check that it is really temporary: not exported
-                    # to the main pipeline
-                    temporary = False
-                    for n, pn in executable.get_linked_items(node, plug_name, in_sub_pipelines=False, process_only=False):
-                        if n is executable:
-                            continue
-                        temporary = True
-                        break
-                    temporary = temporary or getattr(field, 'generate_temporary', False)
-                    field.generate_temporary = temporary
-                    for n, p in executable.get_linked_items(node, field.name):
-                        f = n.field(p)
-                        if f.is_output():
-                            setattr(n.field(p), 'generate_temporary', temporary)
-                self._find_temporary_to_generate(node)
-        elif isinstance(executable, ProcessIteration):
-            for name in executable.iterative_parameters:
-                temporary = getattr(executable.field(name), 'generate_temporary', False)
-                executable.process.field(name).generate_temporary = temporary
-                if isinstance(executable.process, Pipeline):
-                    for n, p in executable.process.get_linked_items(executable.process, name):
-                        f = n.field(p)
-                        if f.is_output():
-                            setattr(n.field(p), 'generate_temporary', temporary)
-            if isinstance(executable.process, Pipeline):
-                self._find_temporary_to_generate(executable.process)
+        # print('!temporaries!  parameters with temporary')
+        # for n, p in temporaries:
+            # print('!temporaries!   ', n, ':', p)
+
+
+    # def _find_temporary_to_generate(self, executable, iteration_process):
+    #     """Finds all output fields that are empty and linked to an input
+    #     and set their generate_temporary attribute to True (and set it to
+    #     False for other fields).
+    #     """
+    #     if isinstance(executable, Pipeline):
+    #         print('!temporaries! ---------', executable.full_name, iteration_process and iteration_process.full_name)
+    #         for node in executable.nodes.values():
+    #             if node is executable:
+    #                 continue
+    #             if isinstance(node, NipypeProcess):
+    #                 #nipype processes do not use temporaries, they produce output
+    #                 # file names
+    #                 return
+
+    #             print('!temporaries 2!', node.full_name)
+    #             for plug_name, plug in node.plugs.items():
+    #                 print('!temporaries 3!', plug_name)
+    #                 value = getattr(node, plug_name, undefined)
+    #                 if not plug.activated or not plug.enabled:
+    #                     continue
+    #                 field = node.field(plug_name)
+    #                 print('!  temporaries 3.1!', plug_name, field.output, field.metadata('write', False))
+    #                 if field.output or not field.metadata('write', False):
+    #                     continue
+    #                 print('!  temporaries 3.2!', plug_name)
+    #                 if field.is_list() and field.path_type:
+    #                     if value is not undefined and len([x for x in value if x in ('', undefined)]) == 0:
+    #                         continue
+    #                 elif value not in (undefined, '') \
+    #                         or (not field.path_type
+    #                             or len(plug.links_to) == 0):
+    #                     continue
+    #                 # check that it is really temporary: not exported
+    #                 # to the main pipeline
+    #                 temporary = False
+    #                 item_stack = list(executable.get_linked_items(node, plug_name, 
+    #                     in_sub_pipelines=True, process_only=False, in_outer_pipelines=True,
+    #                     direction='links_to'))
+    #                 print('!  temporaries 3.3!', node.full_name, plug_name)
+    #                 while item_stack:
+    #                     n, pn = item_stack.pop(0)
+    #                     print('!    temporary 3.3.1!  ', n.full_name, pn, n is executable, isinstance(n, Pipeline))
+    #                     if n is executable and iteration_process:
+    #                         print('!    temporary 3.3.2!  ', iteration_process.full_name, pn)
+    #                         item_stack.extend(executable.get_linked_items(iteration_process, pn, 
+    #                             in_sub_pipelines=True, process_only=False, in_outer_pipelines=True,
+    #                             direction='links_to'))
+    #                     if n is executable or isinstance(n, Pipeline):
+    #                         continue
+    #                     temporary = True
+    #                     break
+    #                 temporary = temporary or getattr(field, 'generate_temporary', False)
+    #                 print('!  temporaries 4!', plug_name, temporary)
+    #                 field.generate_temporary = temporary
+    #                 if isinstance(node, ProcessIteration):
+    #                     node.process.field(field.name).generate_temporary = temporary
+    #                 for n, p in executable.get_linked_items(node, field.name, in_outer_pipelines=False):
+    #                     print('!  temporaries 5!', n.full_name, p)
+    #                     if isinstance(n, Pipeline):
+    #                         continue
+    #                     f = n.field(p)
+    #                     if f.is_output():
+    #                         print('!  temporaries 5.1!', n.full_name, p)
+    #                         setattr(f, 'generate_temporary', temporary)
+    #             if isinstance(node, ProcessIteration):
+    #                 self._find_temporary_to_generate(node.process, node)
+    #             else:
+    #                 self._find_temporary_to_generate(node, None)
diff --git a/capsul/pipeline/pipeline.py b/capsul/pipeline/pipeline.py
index 36ab6537d..3cf446f20 100644
--- a/capsul/pipeline/pipeline.py
+++ b/capsul/pipeline/pipeline.py
@@ -190,7 +190,7 @@ def __init__(self, autoexport_nodes_parameters=None, **kwargs):
         definition: str
             The definition string defines the Node subclass in order to
             serialize it for execution. In most cases it is the module + class
-            names ("caspul.pipeline.test.test_pipeline.MyPipeline" for
+            names ("capsul.pipeline.test.test_pipeline.MyPipeline" for
             instance).
 
             For a "locally defined" pipeline, we use the "custom_pipeline"
@@ -1021,7 +1021,7 @@ def _set_node_enabled(self, is_enabled, _, node_name):
         if node:
             node.enabled = is_enabled
 
-    def all_nodes(self):
+    def all_nodes(self, in_iterations=False):
         """ Iterate over all pipeline nodes including sub-pipeline nodes.
 
         Returns
@@ -1036,6 +1036,10 @@ def all_nodes(self):
                 for sub_node in node.all_nodes():
                     if sub_node is not node:
                         yield sub_node
+            if (in_iterations and 
+               isinstance(node, ProcessIteration) and
+               isinstance(node.process, Pipeline)):
+                yield from node.process.all_nodes(in_iterations=True)
 
     def _check_local_node_activation(self, node):
         """ Try to activate a node and its plugs according to its
@@ -2192,7 +2196,8 @@ def dispatch_plugs(self, node, name):
         self.enable_parameter_links = enable_parameter_links
 
     def get_linked_items(self, node, plug_name=None, in_sub_pipelines=True,
-                         activated_only=True, process_only=True):
+                         activated_only=True, process_only=True,
+                         direction=None, in_outer_pipelines=False):
         '''Return the real process(es) node and plug connected to the given plug.
         Going through switches and inside subpipelines, ignoring nodes that are
         not activated.
@@ -2208,40 +2213,54 @@ def get_linked_items(self, node, plug_name=None, in_sub_pipelines=True,
                 continue
             plug = node.plugs.get(plug_name)
             if plug:
-                if plug.output ^ isinstance(node, Pipeline):
-                    direction = 'links_to'
+                if direction is not None:
+                    directions = (direction,)
                 else:
-                    direction = 'links_from'
-                for dest_plug_name, dest_node in (i[1:3] for i in getattr(plug, direction)):
-                    if dest_node is node or (activated_only
-                                             and not dest_node.activated):
-                        continue
-                    if isinstance(dest_node, Pipeline):
-                        if in_sub_pipelines:
-                            if dest_node is not self:
-                                for n, p in dest_node.get_linked_items(dest_node, dest_plug_name):
+                    if isinstance(node, Pipeline):
+                        if in_outer_pipelines:
+                            directions = ('links_from', 'links_to')
+                        elif plug.output:
+                            directions = ('links_from',)
+                        else:
+                            directions = ('links_to',)
+                    elif plug.output:
+                        directions = ('links_to',)
+                    else:
+                        directions = ('links_from',)
+                for current_direction in directions:
+                    for dest_plug_name, dest_node in (i[1:3] for i in getattr(plug, current_direction)):
+                        if dest_node is node or (activated_only
+                                                and not dest_node.activated):
+                            continue
+                        if isinstance(dest_node, Pipeline):
+                            if ((in_sub_pipelines and dest_node is not self) or 
+                                (in_outer_pipelines and isinstance(dest_node, Pipeline))):
+                                for n, p in self.get_linked_items(dest_node,
+                                                                  dest_plug_name,
+                                                                  in_sub_pipelines=in_sub_pipelines,
+                                                                  direction=current_direction,
+                                                                  in_outer_pipelines=in_outer_pipelines):
                                     if n is not node:
                                         yield (n, p)
-                        else:
                             yield (dest_node, dest_plug_name)
-                    elif isinstance(dest_node, Switch):
-                        if dest_plug_name == 'switch':
-                            if not process_only:
-                                yield (dest_node, dest_plug_name)
+                        elif isinstance(dest_node, Switch):
+                            if dest_plug_name == 'switch':
+                                if not process_only:
+                                    yield (dest_node, dest_plug_name)
+                            else:
+                                for input_plug_name, output_plug_name in dest_node.connections():
+                                    if plug.output ^ isinstance(node, Pipeline):
+                                        if dest_plug_name == input_plug_name:
+                                            if not process_only:
+                                                yield (dest_node, output_plug_name)
+                                            stack.append((dest_node, output_plug_name))
+                                    else:
+                                        if dest_plug_name == output_plug_name:
+                                            if not process_only:
+                                                yield (dest_node, input_plug_name)
+                                            stack.append((dest_node, input_plug_name))
                         else:
-                            for input_plug_name, output_plug_name in dest_node.connections():
-                                if plug.output ^ isinstance(node, Pipeline):
-                                    if dest_plug_name == input_plug_name:
-                                        if not process_only:
-                                            yield (dest_node, output_plug_name)
-                                        stack.append((dest_node, output_plug_name))
-                                else:
-                                    if dest_plug_name == output_plug_name:
-                                        if not process_only:
-                                            yield (dest_node, input_plug_name)
-                                        stack.append((dest_node, input_plug_name))
-                    else:
-                        yield (dest_node, dest_plug_name)
+                            yield (dest_node, dest_plug_name)
 
     def json(self, include_parameters=True):
         result = super().json(include_parameters=include_parameters)
diff --git a/capsul/pipeline/process_iteration.py b/capsul/pipeline/process_iteration.py
index e25101b8d..bf109c09d 100644
--- a/capsul/pipeline/process_iteration.py
+++ b/capsul/pipeline/process_iteration.py
@@ -12,6 +12,7 @@
 from soma.controller import undefined
 
 from capsul.process.process import Process
+import capsul.pipeline.pipeline
 
 
 class ProcessIteration(Process):
@@ -63,7 +64,11 @@ def __init__(self, definition, process, iterative_parameters,
             self._add_plug(parameter)
 
         self.metadata_schema = getattr(self.process, 'metadata_schema', {})
-
+    
+    @property
+    def label(self):
+        return self.process.name + f'[{self.iteration_size()}]'
+    
     def change_iterative_plug(self, parameter, iterative=None):
         '''
         Change a parameter to be iterative (or non-iterative)
@@ -132,7 +137,8 @@ def iterate_over_process_parmeters(self):
     
     def select_iteration_index(self, iteration_index):
         for parameter in self.regular_parameters:
-            setattr(self.process, parameter, getattr(self, parameter, undefined))
+            value = getattr(self, parameter, undefined)
+            setattr(self.process, parameter, value)
         for parameter in self.iterative_parameters:
             values = getattr(self, parameter, undefined)
             if values is not undefined and len(values) != 0:
@@ -158,3 +164,16 @@ def json(self, include_parameters=True):
         if include_parameters:
             result['parameters'] = super(Process,self).json()
         return result
+    
+    def get_linked_items(self, node, plug_name=None, in_sub_pipelines=True,
+                         activated_only=True, process_only=True, direction=None,
+                         in_outer_pipelines=False):
+        if isinstance(self.process, capsul.pipeline.pipeline.Pipeline):
+            yield from  self.process.get_linked_items(
+                node=node,
+                plug_name=plug_name,
+                in_sub_pipelines=in_sub_pipelines,
+                activated_only=activated_only,
+                process_only=process_only,
+                direction=direction,
+                in_outer_pipelines=in_outer_pipelines)
diff --git a/capsul/process/node.py b/capsul/process/node.py
index cb468495e..9341efe38 100644
--- a/capsul/process/node.py
+++ b/capsul/process/node.py
@@ -45,7 +45,7 @@ class Plug(Controller):
         the predecessor plugs of this plug
     """
     enabled : bool = True
-    activated : bool = False
+    activated : bool = True
     output : bool = False
     optional : bool = False
 
@@ -144,7 +144,7 @@ def __init__(self, definition=None, pipeline=None, name=None, inputs={},
         definition: str
             The definition string defines the Node subclass in order to
             serialize it for execution. In most cases it is the module + class
-            names ("caspul.pipeline.test.test_pipeline.MyPipeline" for
+            names ("capsul.pipeline.test.test_pipeline.MyPipeline" for
             instance).
 
             For a "locally defined" pipeline, we use the "custom_pipeline"
diff --git a/capsul/process/process.py b/capsul/process/process.py
index 5fb0edb02..8571585da 100644
--- a/capsul/process/process.py
+++ b/capsul/process/process.py
@@ -106,7 +106,7 @@ def __init__(self, definition, **kwargs):
         definition: str
             The definition string defines the Node subclass in order to
             serialize it for execution. In most cases it is the module + class
-            names ("caspul.pipeline.test.test_pipeline.MyPipeline" for
+            names ("capsul.pipeline.test.test_pipeline.MyPipeline" for
             instance).
 
             For a "locally defined" pipeline, we use the "custom_pipeline"
@@ -136,6 +136,9 @@ def uuid(self):
     def requirements(self):
         return getattr(super(), 'requirements', {})
 
+    @property
+    def label(self):
+        return self.name
 
     def json(self, include_parameters=True):
         '''
diff --git a/capsul/qt_gui/__init__.py b/capsul/qt_gui/__init__.py
index 2371db0d9..ec4bd5035 100644
--- a/capsul/qt_gui/__init__.py
+++ b/capsul/qt_gui/__init__.py
@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
 '''
-Graphical representations and tools for processes and pipelines. The module contains mainly widgets, based on Qt, using either PyQt4, PyQt5 or PySide.
+Graphical representations and tools for processes and pipelines. 
+The module contains mainly widgets, based on Qt, using either PyQt4,
+PyQt5 or PySide.
 '''
diff --git a/capsul/qt_gui/execution_gui.py b/capsul/qt_gui/execution_gui.py
new file mode 100644
index 000000000..789a9f46d
--- /dev/null
+++ b/capsul/qt_gui/execution_gui.py
@@ -0,0 +1,189 @@
+# -*- coding: utf-8 -*-
+from datetime import datetime
+from functools import partial
+from pathlib import Path
+from pprint import pformat
+
+from soma.qt_gui.qt_backend import Qt, loadUi, QtGui
+
+execution_widget_ui = Path(__file__).parent / 'execution_widget.ui'
+
+def execution_widget(database, engine_id, execution_id):
+    global execution_widget_ui
+    widget = loadUi(execution_widget_ui)
+    widget.current_job = None
+    update_execution_widget(widget, database, engine_id, execution_id)
+    update_callback = partial(update_execution_widget, widget, database, engine_id, execution_id)
+    job_callback = partial(show_job, widget=widget, database=database)
+    widget._keep_ref= (update_callback, job_callback)
+    widget.update.clicked.connect(update_callback)
+    widget.jobs.currentItemChanged.connect(job_callback)
+    return widget
+
+def update_execution_widget(widget, database, engine_id, execution_id):
+    report = database.execution_report(engine_id, execution_id)
+    widget.date.setText(f'{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
+    current_job = widget.current_job
+    status = report['status']
+    if status == 'running':
+        status = '🏃'
+    elif report['error']:
+        status = '❌'
+    else:
+        status = '✅'
+    title = f'{status} {report["label"]}'
+    widget.setWindowTitle(title)
+    widget.title.setText(title)
+    widget.waiting.setText(str(len(report['waiting'])))
+    widget.ready.setText(str(len(report['ready'])))
+    widget.ongoing.setText(str(len(report['ongoing'])))
+    widget.done.setText(str(len(report['done'])))
+    widget.failed.setText(str(len(report['failed'])))
+    text = ('<html><body>'
+        f'<b>status</b>: {report["status"]}<br>'
+        f'<b>start time</b>: {report["start_time"]}<br>'
+        f'<b>end time</b>: {report["end_time"]}<br>'
+        f'<b>engine_id</b>: {report["engine_id"]}<br>'
+        f'<b>execution_id</b>: {report["execution_id"]}<br>'
+        'execution context:<pre>\n'
+        f'{pformat(report["execution_context"].asdict(), indent=4)}\n<pre>')
+    if report['error']:
+        text += ('<p style="color: red">'
+            f'ERROR: {report["error"]}')
+        if report['error_detail']:
+            text += '<hr><pre>\n{report["error_detail"]}\n</pre>'
+        text += '</p>'
+    text += '</body></html>'
+    widget.global_text.setText(text)
+    widget.jobs.clear()
+
+    status_map = {
+        'ongoing': '🏃',
+        'done': '✅',
+        'failed': '❌',
+        'ready': '⏳',
+        'waiting': '⏳',
+    }
+    jobs = report['jobs']
+
+    widget._jobs = {job['uuid']: job for job in jobs}
+
+
+    items = {}
+    count = 0
+    for job in jobs:
+        path = []
+        parameters = job['parameters_location']
+        i = 0
+        while i < len(parameters):
+            if parameters[i] in ('nodes', '_iterations'):
+                path.append(parameters[i+1])
+                i+= 2
+            else:
+                raise ValueError(f'Cannot parse parameters_location: {parameters[i]}')
+        path = tuple(path)
+        pp = None
+        parent_item = None
+        for i in range(1, len(path)):
+            p = path[:i]
+            parent_item = items.get(p)
+            if parent_item is None:
+                parent_item = Qt.QTreeWidgetItem(pp, ['', p[-1]])
+                parent_item.engine_id = None
+                parent_item.execution_id = None
+                parent_item.job_uuid = None
+                if pp is None:
+                    widget.jobs.insertTopLevelItem(count, parent_item)
+                    count += 1
+                items[p] = parent_item
+            pp = parent_item
+        job_uuid = job['uuid']
+        process_definition = job.get('process', {}).get('definition')
+        status = job['status']
+        status = status_map.get(status, status)
+        item = items.get(path)
+        if item:
+            item.setText(0, status)
+            item.setText(1, process_definition)
+        else:
+            item = Qt.QTreeWidgetItem(parent_item, [status_map.get(status, status), process_definition])
+            items[path] = item
+        item.engine_id = engine_id
+        item.execution_id = execution_id
+        item.job_uuid = job_uuid
+        if parent_item is None:
+            widget.jobs.insertTopLevelItem(count, item)
+            count += 1
+        else:
+            if status == '❌':
+                parent_item.setText(0, '❌')
+            elif status == '🏃':
+                if parent_item.text(0) != '❌':
+                    parent_item.setText(0, '🏃')
+            elif status == '⏳':
+                if parent_item.text(0) not in ('❌', '🏃'):
+                    parent_item.setText(0, '⏳')
+            else:
+                if parent_item.text(0) == '':
+                    parent_item.setText(0, '✅')
+
+        if job_uuid == current_job:
+            widget.jobs.setCurrentItem(item)
+
+        
+    # now = datetime.now()
+    # count = 0
+    # for job in sorted(jobs, key=lambda j: (j.get('start_time') if j.get('start_time') else now)):
+    #         job_uuid = job['uuid']
+    #         process_definition = job.get('process', {}).get('definition')
+    #         status = job['status']
+
+    #         parameters = report['workflow_parameters']
+    #         if parameters:
+    #             for index in job.get('parameters_location', []):
+    #                 if index.isnumeric():
+    #                     index = int(index)
+    #                 parameters = parameters[index]
+
+    #         item = Qt.QTreeWidgetItem([status_map.get(status, status), process_definition])
+    #         item.execution_id = execution_id
+    #         item.job_uuid = job_uuid
+    #         widget.jobs.insertTopLevelItem(count, item)
+    #         count += 1
+
+def show_job(item, widget, database):
+    if not item or not item.execution_id:
+        text = ''
+    else:
+        engine_id = item.engine_id
+        execution_id = item.execution_id
+        job_uuid = item.job_uuid
+        widget.current_job = job_uuid
+        job = widget._jobs[job_uuid]
+        text = ('<html><body>'
+            f'<b>job_uuid</b>: {job["uuid"]}<br>'
+            f'<b>process</b>: {job.get("process", {}).get("definition")}<br>'
+            f'<b>pipeline node</b>: {".".join(i for i in job.get("parameters_location", "") if i != "nodes")}<br>'
+            f'<b>return code</b>: {job.get("returncode")}<br>'
+            f'<b>start time</b>: {job.get("start_time")}<br>'
+            f'<b>end time</b>: {job.get("end_time")}<br>'
+            f'<b>wait for</b>: {job.get("wait_for", [])}<br>'
+            f'<b>waited by</b>: {job.get("waited_by", [])}<br>'
+        )
+        parameters = database.workflow_parameters(engine_id, execution_id)
+        if parameters:
+            for index in job.get('parameters_location', []):
+                if index.isnumeric():
+                    index = int(index)
+                parameters = parameters[index]
+        if parameters:
+            text += f'parameters:<pre>{pformat(parameters.no_proxy(), indent=4)}</pre>'
+        else:
+            text += 'parameters: none<br>'
+        stdout = job.get('stdout')
+        stderr = job.get('stderr')
+        if stdout:
+            text += f'<br><pre>{stdout}</pre>'
+        if stderr:
+            text += f'<br><pre style="color: red">{stderr}</pre>'
+    widget.job.setText(text)
diff --git a/capsul/qt_gui/execution_widget.ui b/capsul/qt_gui/execution_widget.ui
new file mode 100644
index 000000000..b5ced94fd
--- /dev/null
+++ b/capsul/qt_gui/execution_widget.ui
@@ -0,0 +1,296 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ui version="4.0">
+ <class>Form</class>
+ <widget class="QWidget" name="Form">
+  <property name="geometry">
+   <rect>
+    <x>0</x>
+    <y>0</y>
+    <width>898</width>
+    <height>628</height>
+   </rect>
+  </property>
+  <property name="windowTitle">
+   <string>Form</string>
+  </property>
+  <layout class="QVBoxLayout" name="verticalLayout_2">
+   <item>
+    <layout class="QHBoxLayout" name="horizontalLayout">
+     <item>
+      <widget class="QLabel" name="title">
+       <property name="sizePolicy">
+        <sizepolicy hsizetype="Preferred" vsizetype="Maximum">
+         <horstretch>0</horstretch>
+         <verstretch>0</verstretch>
+        </sizepolicy>
+       </property>
+       <property name="sizeIncrement">
+        <size>
+         <width>0</width>
+         <height>0</height>
+        </size>
+       </property>
+       <property name="font">
+        <font>
+         <pointsize>18</pointsize>
+        </font>
+       </property>
+       <property name="text">
+        <string>The title</string>
+       </property>
+       <property name="alignment">
+        <set>Qt::AlignCenter</set>
+       </property>
+      </widget>
+     </item>
+     <item>
+      <widget class="QLabel" name="date">
+       <property name="sizePolicy">
+        <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+         <horstretch>0</horstretch>
+         <verstretch>0</verstretch>
+        </sizepolicy>
+       </property>
+       <property name="text">
+        <string>now</string>
+       </property>
+       <property name="alignment">
+        <set>Qt::AlignRight|Qt::AlignTrailing|Qt::AlignVCenter</set>
+       </property>
+      </widget>
+     </item>
+     <item>
+      <widget class="QPushButton" name="update">
+       <property name="sizePolicy">
+        <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+         <horstretch>0</horstretch>
+         <verstretch>0</verstretch>
+        </sizepolicy>
+       </property>
+       <property name="maximumSize">
+        <size>
+         <width>33</width>
+         <height>33</height>
+        </size>
+       </property>
+       <property name="font">
+        <font>
+         <pointsize>20</pointsize>
+        </font>
+       </property>
+       <property name="toolTip">
+        <string>Update</string>
+       </property>
+       <property name="text">
+        <string>🔄</string>
+       </property>
+      </widget>
+     </item>
+    </layout>
+   </item>
+   <item>
+    <layout class="QHBoxLayout" name="horizontalLayout_7">
+     <item>
+      <layout class="QHBoxLayout" name="horizontalLayout_2">
+       <item>
+        <widget class="QLabel" name="label_2">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="font">
+          <font>
+           <pointsize>14</pointsize>
+          </font>
+         </property>
+         <property name="text">
+          <string>🏃</string>
+         </property>
+        </widget>
+       </item>
+       <item>
+        <widget class="QLabel" name="ongoing">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>ongoing</string>
+         </property>
+        </widget>
+       </item>
+      </layout>
+     </item>
+     <item>
+      <layout class="QHBoxLayout" name="horizontalLayout_3">
+       <item>
+        <widget class="QLabel" name="label_4">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>👍</string>
+         </property>
+        </widget>
+       </item>
+       <item>
+        <widget class="QLabel" name="ready">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>ready</string>
+         </property>
+        </widget>
+       </item>
+      </layout>
+     </item>
+     <item>
+      <layout class="QHBoxLayout" name="horizontalLayout_4">
+       <item>
+        <widget class="QLabel" name="label_3">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>✋</string>
+         </property>
+        </widget>
+       </item>
+       <item>
+        <widget class="QLabel" name="waiting">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>waiting</string>
+         </property>
+        </widget>
+       </item>
+      </layout>
+     </item>
+     <item>
+      <layout class="QHBoxLayout" name="horizontalLayout_5">
+       <item>
+        <widget class="QLabel" name="label_5">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>✅</string>
+         </property>
+        </widget>
+       </item>
+       <item>
+        <widget class="QLabel" name="done">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>done</string>
+         </property>
+        </widget>
+       </item>
+      </layout>
+     </item>
+     <item>
+      <layout class="QHBoxLayout" name="horizontalLayout_6">
+       <item>
+        <widget class="QLabel" name="label_6">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>❌</string>
+         </property>
+        </widget>
+       </item>
+       <item>
+        <widget class="QLabel" name="failed">
+         <property name="sizePolicy">
+          <sizepolicy hsizetype="Preferred" vsizetype="Fixed">
+           <horstretch>0</horstretch>
+           <verstretch>0</verstretch>
+          </sizepolicy>
+         </property>
+         <property name="text">
+          <string>failed</string>
+         </property>
+        </widget>
+       </item>
+      </layout>
+     </item>
+    </layout>
+   </item>
+   <item>
+    <widget class="QSplitter" name="splitter_2">
+     <property name="orientation">
+      <enum>Qt::Horizontal</enum>
+     </property>
+     <widget class="QSplitter" name="splitter">
+      <property name="orientation">
+       <enum>Qt::Vertical</enum>
+      </property>
+      <widget class="QTextBrowser" name="global_text"/>
+      <widget class="QWidget" name="">
+       <layout class="QVBoxLayout" name="verticalLayout">
+        <item>
+         <widget class="QLabel" name="label">
+          <property name="text">
+           <string>Jobs</string>
+          </property>
+          <property name="alignment">
+           <set>Qt::AlignCenter</set>
+          </property>
+         </widget>
+        </item>
+        <item>
+         <widget class="QTreeWidget" name="jobs">
+          <column>
+           <property name="text">
+            <string>status</string>
+           </property>
+          </column>
+          <column>
+           <property name="text">
+            <string>Job</string>
+           </property>
+          </column>
+         </widget>
+        </item>
+       </layout>
+      </widget>
+     </widget>
+     <widget class="QTextBrowser" name="job"/>
+    </widget>
+   </item>
+  </layout>
+ </widget>
+ <resources/>
+ <connections/>
+</ui>
diff --git a/capsul/run.py b/capsul/run.py
index 4f9994473..5f60106cc 100644
--- a/capsul/run.py
+++ b/capsul/run.py
@@ -1,102 +1,115 @@
 # -*- coding: utf-8 -*-
 
+import contextlib
+import io
+import json
 import os
+import subprocess
 import sys
+import traceback
 
 from soma.undefined import undefined
 
 from .application import Capsul
-from .database import execution_database
+from .database import engine_database
 
-debug = False
+
+def run_job(database, engine_id, execution_id, job_uuid, same_python=False, debug=False):
+    if same_python:
+        stdout = io.StringIO()
+        stderr = io.StringIO()
+        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
+            try:
+                execute_job(database, engine_id, execution_id, job_uuid, debug=debug)
+                returncode = 0
+            except Exception as e:
+                print(traceback.format_exc(), file=stderr)
+                returncode = 1
+        stdout = stdout.getvalue()
+        stderr = stderr.getvalue()
+    else:
+        env = os.environ.copy()
+        env['CAPSUL_WORKER_DATABASE'] = json.dumps(database.config)
+        if debug:
+            env['CAPSUL_DEBUG'] = '1'
+        result = subprocess.run(['python', '-m', 'capsul.run', engine_id, job_uuid], env=env, capture_output=True)
+        returncode = result.returncode
+        stdout = result.stdout.decode()
+        stderr = result.stderr.decode()
+    return returncode, stdout, stderr
+
+
+
+def execute_job(database, engine_id, execution_id, job_uuid, debug=False):
+    execution_context = database.execution_context(engine_id, execution_id)
+    execution_context.dataset.tmp = {
+        'path': database.tmp(engine_id, execution_id),
+    }
+    job = database.job(engine_id, execution_id, job_uuid)
+    if job is None:
+        raise ValueError(f'No such job: {job_uuid}')
+    process_json = job['process']
+    parameters_location = job['parameters_location']
+    process = Capsul.executable(process_json)
+    if debug:
+        print(f'---- init {process.definition} {parameters_location} ----')
+    workflow_parameters = database.workflow_parameters(engine_id, execution_id)
+    parameters = workflow_parameters
+    for index in parameters_location:
+        if index.isnumeric():
+            index = int(index)
+        parameters = parameters[index]
+    if debug:
+        from pprint import pprint
+        pprint(parameters.no_proxy())
+        print(f'----')
+    for field in process.user_fields():
+        if field.name in parameters and parameters[field.name] is not None:
+            value = parameters.no_proxy(parameters[field.name])
+            setattr(process, field.name, value)
+    execution_context.executable = process
+    process.resolve_paths(execution_context)
+    if debug:
+        print(f'---- start {process.definition} ----')
+        for field in process.user_fields():
+            if not field.output:
+                value = getattr(process, field.name, undefined)
+                if value is not undefined:
+                    print ('   ', field.name, '=', value)
+    process.before_execute(execution_context)
+    result = process.execute(execution_context)
+    process.after_execute(result, execution_context)
+    if debug:
+        print(f'---- stop {process.definition} ----')
+        for field in process.user_fields():
+            if field.output:
+                value = getattr(process, field.name, undefined)
+                if value is not undefined:
+                    print ('   ', field.name, '=', value)
+    output_values = {}
+    for field in process.user_fields():
+        if field.output:
+            value = getattr(process, field.name, undefined)
+            if value is not undefined:
+                output_values[field.name] = value
+    if output_values is not None:
+        database.update_workflow_parameters(engine_id, execution_id, parameters_location, output_values)
 
 if __name__ == '__main__':
-    if len(sys.argv) < 2:
-        print('Capsul cannot run command because parameters are missing: '
+    if len(sys.argv) != 3:
+        print('Wrong number of paramaters, 2 expected:'
               f'command={sys.argv}',
               file=sys.stderr)
         sys.exit(1)
-    database_url = os.environ.get('CAPSUL_DATABASE')
-    if not database_url:
-        print('Capsul cannot run job because CAPSUL_DATABASE is not defined: '
+    engine_id = sys.argv[1]
+    job_uuid = sys.argv[2]
+    db_config = os.environ.get('CAPSUL_WORKER_DATABASE')
+    if not db_config:
+        print('Capsul cannot run job because CAPSUL_WORKER_DATABASE is not defined: '
               f'command={sys.argv}',
               file=sys.stderr)
-    execution_id = os.environ.get('CAPSUL_EXECUTION_ID')
-    if not execution_id:
-        print('Capsul cannot run job because CAPSUL_EXECUTION_ID is not defined: '
-              f'command={sys.argv}',
-              file=sys.stderr)
-        sys.exit(1)
-    database = execution_database(database_url)
-    try:
-        execution_context = database.execution_context(execution_id)
-        execution_context.dataset.tmp = {
-            'path': database.tmp(execution_id),
-        }
-        command = sys.argv[1]
-        args = sys.argv[2:]
-        invalid_parameters = False
-        if command == 'process':
-            if len(args) == 1:
-                job_uuid = args[0]
-                job = database.job(execution_id, job_uuid)
-                if job is None:
-                    raise ValueError(f'No such job: {job_uuid}')
-                process_json = job['process']
-                parameters_location = job['parameters_location']
-                process = Capsul.executable(process_json)
-                if debug:
-                    print(f'---- init {process.definition} {parameters_location} ----')
-                workflow_parameters = database.workflow_parameters(execution_id)
-                parameters = workflow_parameters
-                for index in parameters_location:
-                    if index.isnumeric():
-                        index = int(index)
-                    parameters = parameters[index]
-                if debug:
-                    from pprint import pprint
-                    pprint(parameters.no_proxy())
-                    print(f'----')
-                for field in process.user_fields():
-                    if field.name in parameters and parameters[field.name] is not None:
-                        value = parameters.no_proxy(parameters[field.name])
-                        setattr(process, field.name, value)
-                execution_context.executable = process
-                process.resolve_paths(execution_context)
-                if debug:
-                    print(f'---- start {process.definition} ----')
-                    for field in process.user_fields():
-                        if not field.output:
-                            value = getattr(process, field.name, undefined)
-                            if value is not undefined:
-                                print ('   ', field.name, '=', value)
-                process.before_execute(execution_context)
-                result = process.execute(execution_context)
-                process.after_execute(result, execution_context)
-                if debug:
-                    print(f'---- stop {process.definition} ----')
-                    for field in process.user_fields():
-                        if field.output:
-                            value = getattr(process, field.name, undefined)
-                            if value is not undefined:
-                                print ('   ', field.name, '=', value)
-                output_values = {}
-                for field in process.user_fields():
-                    if field.output:
-                        value = getattr(process, field.name, undefined)
-                        if value is not undefined:
-                            output_values[field.name] = value
-                if output_values is not None:
-                    database.update_workflow_parameters(execution_id, parameters_location, output_values)
-            else:
-                invalid_parameters = True  
-        else:
-            invalid_parameters = True
-    finally:
-        database.close()
-    
-    if invalid_parameters:
-        print('Capsul cannot run command because parameters are invalid: '
-            f'command={sys.argv}',
-            file=sys.stderr)
         sys.exit(1)
+    db_config = json.loads(db_config)
+    debug = 'CAPSUL_DEBUG' in os.environ
+    with engine_database(db_config) as database:
+        execute_job(database, engine_id, job_uuid, debug=debug)
diff --git a/doc/source/api/index.rst b/doc/source/api/index.rst
index b44b975bc..6175504fb 100644
--- a/doc/source/api/index.rst
+++ b/doc/source/api/index.rst
@@ -112,7 +112,7 @@ Running external software
 :mod:`capsul.subprocess`
 ------------------------
 
-This module is obsolete and will be replaced by :mod:`~capsul.in_process`. It is still used in Capsul v2 when using :class:`~caspul.study_config.study_config.StudyConfig`.
+This module is obsolete and will be replaced by :mod:`~capsul.in_process`. It is still used in Capsul v2 when using :class:`~capsul.study_config.study_config.StudyConfig`.
 
 
 :mod:`capsul.subprocess.fsl` Classes:


Test patch:
diff --git a/capsul/pipeline/test/test_pipeline.py b/capsul/pipeline/test/test_pipeline.py
index e8d174144..183e5c9a1 100644
--- a/capsul/pipeline/test/test_pipeline.py
+++ b/capsul/pipeline/test/test_pipeline.py
@@ -32,8 +32,10 @@ def __init__(self, definition=None):
         self.add_field("other_output", float, optional=True, output=True)
 
     def execute(self, context):
-        with open(self.output_image, 'w') as f:
-            f.write('dummy output.\n')
+        if os.path.exists(self.input_image):
+            with open(self.input_image) as i:
+                with open(self.output_image, 'w') as o:
+                    o.write(i.read())
         self.other_output = 24.6
 
 
@@ -132,12 +134,12 @@ def test_run_pipeline(self):
         tmp = tempfile.mkstemp('', prefix='capsul_test_pipeline')
         ofile = tmp[1]
         os.close(tmp[0])
-        os.unlink(tmp[1])
+        # os.unlink(tmp[1])
         capsul = Capsul()
         try:
             with capsul.engine() as engine:
                 engine.run(self.pipeline, timeout=5,
-                    input_image='/tmp/bloup',
+                    input_image=ofile,
                     output=ofile)
         finally:
             if os.path.exists(tmp[1]):
@@ -209,20 +211,16 @@ def test():
 
 
 if __name__ == "__main__":
-    print("RETURNCODE: ", test())
-
-    if '-v' in sys.argv[1:]:
-        import sys
-        from soma.qt_gui.qt_backend import Qt
-        from capsul.qt_gui.widgets import PipelineDeveloperView
-
-        app = Qt.QApplication.instance()
-        if not app:
-            app = Qt.QApplication(sys.argv)
-
-        pipeline = executable(MyPipeline)
-        #setattr(pipeline.nodes_activation, "node2", False)
-        view1 = PipelineDeveloperView(pipeline, allow_open_controller=True)
-        view1.show()
-        app.exec_()
-        del view1
+    from soma.qt_gui.qt_backend import Qt
+    from capsul.qt_gui.widgets import PipelineDeveloperView
+
+    app = Qt.QApplication.instance()
+    if not app:
+        app = Qt.QApplication(sys.argv)
+
+    pipeline = executable(MyPipeline)
+    setattr(pipeline.nodes_activation, "node2", True)
+    view1 = PipelineDeveloperView(pipeline, allow_open_controller=True)
+    view1.show()
+    app.exec_()
+    del view1
diff --git a/capsul/pipeline/test/test_pipeline_parameters.py b/capsul/pipeline/test/test_pipeline_parameters.py
index 7a708e1ed..fefeb07aa 100644
--- a/capsul/pipeline/test/test_pipeline_parameters.py
+++ b/capsul/pipeline/test/test_pipeline_parameters.py
@@ -727,7 +727,8 @@ def create_pipeline():
 
         with Capsul().engine() as ce:
             ce.run(pipeline1, timeout=5)
-
+        pipeline1.out
+        
         save_pipeline_parameters(self.path, pipeline1)
 
         # Reinitializing pipeline and loading parameters
diff --git a/capsul/pipeline/test/test_pipeline_with_temp.py b/capsul/pipeline/test/test_pipeline_with_temp.py
index cf9e354b5..9d78fa666 100644
--- a/capsul/pipeline/test/test_pipeline_with_temp.py
+++ b/capsul/pipeline/test/test_pipeline_with_temp.py
@@ -142,7 +142,6 @@ def test_iterative_pipeline_with_temp(self):
                                                input_name]
             self.iter_pipeline.output = output_name
 
-            # run sequentially
             with Capsul().engine() as ce:
                 ce.run(self.iter_pipeline, timeout=5)
 
diff --git a/capsul/pipeline/test/test_pipeline_workflow.py b/capsul/pipeline/test/test_pipeline_workflow.py
index d37cf3f51..7a2aaf8a2 100644
--- a/capsul/pipeline/test/test_pipeline_workflow.py
+++ b/capsul/pipeline/test/test_pipeline_workflow.py
@@ -224,7 +224,7 @@ def test_wf_run(self):
                 text = f.read()
                 self.assertEqual(len(text.split('\n')), lens[o])
 
-    def test_iter_workflow_without_temp(self):
+    def test_iter_without_temp(self):
         pipeline = Capsul.executable(DummyPipelineIterSimple)
         self.assertTrue(pipeline is not None)
         niter = 2
@@ -253,6 +253,7 @@ def test_iter_workflow_without_temp(self):
 
     def test_iter_workflow(self):
         pipeline = Capsul.executable(DummyPipelineIter)
+
         self.assertTrue(pipeline is not None)
         niter = 2
         pipeline.input = [osp.join(self.tmpdir, 'file_in%d' % i)
@@ -292,48 +293,42 @@ def test():
 
 
 if __name__ == "__main__":
-    verbose = False
-    if len(sys.argv) >= 2 and sys.argv[1] in ('-v', '--verbose'):
-        verbose = True
-
-    print("RETURNCODE: ", test())
-
-    if verbose:
-        import sys
-        from soma.qt_gui import qt_backend
-        # qt_backend.set_qt_backend('PyQt4')
-        from soma.qt_gui.qt_backend import QtGui
-        from capsul.qt_gui.widgets import PipelineDeveloperView
-
-        app = QtGui.QApplication.instance()
-        if not app:
-            app = QtGui.QApplication(sys.argv)
-        pipeline = DummyPipeline()
-        pipeline.input = '/tmp/file_in.nii'
-        pipeline.output1 = '/tmp/file_out1.nii'
-        pipeline.output2 = '/tmp/file_out2.nii'
-        pipeline.output3 = '/tmp/file_out3.nii'
-        view1 = PipelineDeveloperView(pipeline, show_sub_pipelines=True,
-                                       allow_open_controller=True)
-        view1.show()
-
-        pipeline2 = DummyPipelineIterSimple()
-        pipeline2.input = ['/tmp/file_in1.nii', '/tmp/file_in2.nii']
-        pipeline2.output = '/tmp/file_out.nii'
-        pipeline2.intermediate = ['/tmp/file_out1',
-                                  '/tmp/file_out2']
-        view2 = PipelineDeveloperView(pipeline2, show_sub_pipelines=True,
-                                       allow_open_controller=True)
-        view2.show()
-
-        pipeline3 = DummyPipelineIter()
-        pipeline3.input = ['/tmp/file_in1.nii', '/tmp/file_in2.nii']
-        pipeline3.output1 = '/tmp/file_out1.nii'
-        pipeline3.output2 = '/tmp/file_out2.nii'
-        pipeline3.output3 = '/tmp/file_out3.nii'
-        view3 = PipelineDeveloperView(pipeline3, show_sub_pipelines=True,
-                                       allow_open_controller=True)
-        view3.show()
-
-        app.exec_()
-        del view1
+    import sys
+    from soma.qt_gui import qt_backend
+    from soma.qt_gui.qt_backend import QtGui
+    from capsul.qt_gui.widgets import PipelineDeveloperView
+
+    app = QtGui.QApplication.instance()
+    if not app:
+        app = QtGui.QApplication(sys.argv)
+    pipeline = Capsul.executable(DummyPipeline)
+    pipeline.input = '/tmp/file_in.nii'
+    pipeline.output1 = '/tmp/file_out1.nii'
+    pipeline.output2 = '/tmp/file_out2.nii'
+    pipeline.output3 = '/tmp/file_out3.nii'
+    view1 = PipelineDeveloperView(pipeline, show_sub_pipelines=True,
+                                    allow_open_controller=True)
+    view1.show()
+
+    pipeline2 = Capsul.executable(DummyPipelineIterSimple)
+    pipeline2.input = ['/tmp/file_in1.nii', '/tmp/file_in2.nii']
+    pipeline2.output = '/tmp/file_out.nii'
+    pipeline2.intermediate = ['/tmp/file_out1',
+                                '/tmp/file_out2']
+    view2 = PipelineDeveloperView(pipeline2, show_sub_pipelines=True,
+                                    allow_open_controller=True)
+    view2.show()
+
+    pipeline3 = Capsul.executable(DummyPipelineIter)
+    pipeline3.input = ['/tmp/file_in1.nii', '/tmp/file_in2.nii']
+    pipeline3.output1 = '/tmp/file_out1.nii'
+    pipeline3.output2 = '/tmp/file_out2.nii'
+    pipeline3.output3 = '/tmp/file_out3.nii'
+    view3 = PipelineDeveloperView(pipeline3, show_sub_pipelines=True,
+                                    allow_open_controller=True)
+    view3.show()
+
+    app.exec_()
+    del view1
+    del view2
+    del view3
diff --git a/capsul/process/test/test_metadata_schema.py b/capsul/process/test/test_metadata_schema.py
index 42c79b780..098e60042 100644
--- a/capsul/process/test/test_metadata_schema.py
+++ b/capsul/process/test/test_metadata_schema.py
@@ -90,7 +90,8 @@ def setUp(self):
             input.mkdir()
             output = temp_home_dir / 'out'
             output.mkdir()
-            with (config / f'{app_name}.json').open('w') as f:
+            site_file = config / f'{app_name}.json'
+            with site_file.open('w') as f:
                 json.dump({
                     'builtin': {
                         'python_modules': [
@@ -109,7 +110,7 @@ def setUp(self):
                     }
                 }, f)
             Capsul.delete_singleton()
-            capsul = Capsul(app_name)
+            capsul = Capsul(app_name, site_file=site_file)
 
         except BaseException:  # clean up in case of interruption
             if old_home is None:
diff --git a/capsul/test/test_completion.py b/capsul/test/test_completion.py
index bc93e50e6..60a7a3949 100644
--- a/capsul/test/test_completion.py
+++ b/capsul/test/test_completion.py
@@ -65,17 +65,6 @@ def pipeline_definition(self):
         self.export_parameter('right_hemi', 'output', 'right_gw_mesh')
 
 
-class BiasCorrectionBrainVISA(
-        ProcessSchema, schema='brainvisa',
-        process='capsul.test.test_tiny_morphologist.BiasCorrection'):
-    output = {'prefix': 'nobias'}
-
-
-class FakeSPMNormalization12BrainVISA(
-        ProcessSchema, schema='brainvisa',
-        process='capsul.test.test_tiny_morphologist.FakeSPMNormalization12'):
-    output = {'prefix': 'normalized_fakespm12'}
-
 
 class TestPipelineBIDS(ProcessSchema, schema='bids',
                        process=TestPipeline):
@@ -87,20 +76,22 @@ class TestPipelineBrainVISA(ProcessSchema, schema='brainvisa',
                             process=TestPipeline):
     _ = {
         '*': {'process': 'test_pipeline'},
-        'left*.*': {'side': 'L'},
-        'right*.*': {'side': 'R'},
     }
+    _nodes = {
+        'split': {
+            '*': {'seg_directory': 'segmentation'},
+            'left_output':{'side': 'L', 'suffix': ''},
+            'right_output': {'side': 'R', 'suffix': ''},
+        }
+    }
+    right_gw_mesh = {'extension': 'gii', 'seg_directory': None, 'prefix': ''}
+    left_gw_mesh = {'extension': 'gii', 'seg_directory': None, 'prefix': ''}
+
+
 
 
 datasets = {
-    'input': 'input',
     'template': 'shared',
-    'nobias': 'output',
-    'normalized': 'output',
-    'left_gw_classif': 'output',
-    'left_gw_mesh': 'output',
-    'right_gw_classif': 'output',
-    'right_gw_mesh': 'output',
 }
 
 
@@ -184,19 +175,19 @@ def test_pipeline_completion(self):
 
         expected = {
             'input': '!{dataset.input.path}/rawdata/sub-aleksander/ses-m0/anat/sub-aleksander_ses-m0_T1w.nii',
-            'template': '!{fakespm.directory}/template',
+            # 'template': '!{fakespm.directory}/template',
+            'template': '!{dataset.shared.path}',
             'nobias': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/nobias_aleksander.nii',
             'normalized': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/nobias_aleksander.nii',
             'left_gw_classif': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/segmentation/Lgrey_white_aleksander.nii',
-            'left_gw_mesh': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/Laleksander.nii',
+            'left_gw_mesh': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/Laleksander.gii',
             'right_gw_classif': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/segmentation/Rgrey_white_aleksander.nii',
-            'right_gw_mesh': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/Raleksander.nii',
+            'right_gw_mesh': '!{dataset.output.path}/whaterver/aleksander/test_pipeline/m0/default_analysis/Raleksander.gii',
         }
 
-        self.maxDiff = 2000
+        self.maxDiff = 3000
         self.assertEqual(params, expected)
 
-    @unittest.skip('not working yet')
     def test_iteration_completion(self):
 
         expected_completion = {
@@ -260,21 +251,18 @@ def test():
 
 
 if __name__ == "__main__":
-    print("RETURNCODE: ", test())
-    if '-v' in sys.argv[1:]:
-        from soma.qt_gui.qt_backend import Qt
-        from capsul.qt_gui.widgets.pipeline_developer_view \
-            import PipelineDeveloperView
-
-        app = Qt.QApplication.instance()
-        if app is None:
-            app = Qt.QApplication([])
-
-        pipeline = executable('capsul.test.test_completion.TestPipeline')
-
-        pv = PipelineDeveloperView(
-            pipeline, allow_open_controller=True, enable_edition=True,
-            show_sub_pipelines=True)
-        pv.auto_dot_node_positions()
-        pv.show()
-        app.exec_()
+    from soma.qt_gui.qt_backend import Qt
+    from capsul.qt_gui.widgets.pipeline_developer_view \
+        import PipelineDeveloperView
+
+    app = Qt.QApplication.instance()
+    if app is None:
+        app = Qt.QApplication([])
+
+    pipeline = executable('capsul.test.test_completion.TestPipeline')
+    pv = PipelineDeveloperView(
+        pipeline, allow_open_controller=True, enable_edition=True,
+        show_sub_pipelines=True)
+    pv.auto_dot_node_positions()
+    pv.show()
+    app.exec_()
diff --git a/capsul/test/test_fake_morphologist.py b/capsul/test/test_fake_morphologist.py
index a8effd185..311241ea0 100644
--- a/capsul/test/test_fake_morphologist.py
+++ b/capsul/test/test_fake_morphologist.py
@@ -8,11 +8,10 @@
 import time
 import copy
 
-from soma.controller import field, File
 from soma.controller import Directory, undefined
 
 from capsul.api import Capsul, Process, Pipeline
-from capsul.config.configuration import ModuleConfiguration, default_workers_type, default_database_url
+from capsul.config.configuration import ModuleConfiguration, default_engine_database, default_engine_start_workers
 from capsul.dataset import ProcessMetadata, ProcessSchema, MetadataSchema
 
 class FakeSPMConfiguration(ModuleConfiguration):
@@ -447,24 +446,43 @@ class MorphologistBrainVISA(ProcessSchema, schema='brainvisa',
                             process=Morphologist):
     _ = {
         '*': {'process': None, 'modality': 't1mri'},
-        'GreyWhiteClassification.*': {'side': 'L'},
-        'GreyWhiteTopology.*': {'side': 'L'},
-        'GreyWhiteMesh.*': {'sidebis': 'L'},
-        'PialMesh.*': {'sidebis': 'L'},
-        'SulciSkeleton.*': {'side': 'L'},
-        'CorticalFoldsGraph.*': {'side': 'L'},
-        'SulciRecognition.*': {'side': 'L'},
-        '*_1.*': {'side': 'R'},
-        'GreyWhiteMesh_1.*': {'sidebis': 'R', 'side': None},
-        'PialMesh_1.*': {'sidebis': 'R', 'side': None},
-        'SulciRecognition*.*': {
+    }
+
+    _nodes = {
+        'GreyWhiteClassification': {'*': {'side': 'L'}},
+        'GreyWhiteTopology': {'*': {'side': 'L'}},
+        'GreyWhiteMesh' : {'*': {'sidebis': 'L'}},
+        'PialMesh': {'*': {'sidebis': 'L'}},
+        'SulciSkeleton': {'*': {'side': 'L'}},
+        'CorticalFoldsGraph': {'*': {'side': 'L'}},
+        'SulciRecognition': {'*': {'side': 'L'}},
+        '*_1': {'*': {'side': 'R'}},
+        'GreyWhiteMesh_1': {'*': {'sidebis': 'R', 'side': None}},
+        'PialMesh_1': {'*': {'sidebis': 'R', 'side': None}},
+        'SulciRecognition*': {'*': {
             'sulci_graph_version':
                 lambda **kwargs:
                     f'{kwargs["process"].CorticalFoldsGraph_graph_version}',
             'sulci_recognition_session': 'default_session_auto',
-        },
+            'prefix': None,
+            'sidebis': None,
+        }},
+    }
+    imported_t1mri = {
+        'analysis': undefined,
+        'side': None,
+        'sidebis': None,
+        'seg_directory': None,
+        'suffix': None,
+        'extension': 'nii',
+    }
+    t1mri_nobias = {
+        'side': None,
+        'sidebis': None,
+        'seg_directory': None,
+        'suffix': None,
+        'extension': 'nii',
     }
-    imported_t1mri = {'analysis': undefined}
     t1mri_referential= {
         'analysis': undefined,
         'seg_directory': 'registration',
@@ -472,14 +490,22 @@ class MorphologistBrainVISA(ProcessSchema, schema='brainvisa',
         'suffix': lambda **kwargs:
             f'{kwargs["metadata"].acquisition}',
         'extension': 'referential'}
-    reoriented_t1mri = {'analysis': undefined}
     Talairach_transform = {
         'analysis': undefined,
         'seg_directory': 'registration',
+        'prefix': '',
         'short_prefix': 'RawT1-',
         'suffix': lambda **kwargs:
             f'{kwargs["metadata"].acquisition}_TO_Talairach-ACPC',
+        'side': None,
+        'sidebis': None,
         'extension': 'trm'}
+    left_labelled_graph = {
+        'part': 'left_hemi'
+    }
+    right_labelled_graph = {
+        'part': 'right_hemi'
+    }
 
 
 class MorphologistShared(ProcessSchema, schema='shared', process=Morphologist):
@@ -707,7 +733,7 @@ def setUp(self):
         with self.config_file.open('w') as f:
             json.dump(config, f)
 
-        self.capsul = Capsul('test_fake_morphologist', site_file=self.config_file)
+        self.capsul = Capsul('test_fake_morphologist', site_file=self.config_file, user_file=None)
         return super().setUp()
 
     def tearDown(self):
@@ -722,8 +748,8 @@ def test_fake_morphologist_config(self):
         self.maxDiff = 2000
         expected_config = {
             'builtin': {
-                'workers_type': default_workers_type,
-                'database_url': default_database_url,
+                'database': default_engine_database,
+                'start_workers': default_engine_start_workers,
                 'dataset': {
                     'input': {
                         'path': str(self.tmp / 'bids'),
@@ -1009,7 +1035,7 @@ def test_path_generation(self):
 
         bv_schema_diff = [
           ([], []),
-          (['normalized_t1mri', 'MNI_transform'], ['reoriented_t1mri']),
+          (['normalized_t1mri', 'MNI_transform'], []),
           (['normalized_t1mri', 'MNI_transform',
             'normalization_spm_native_transformation',
             'normalization_spm_native_transformation_pass1'],
@@ -1069,7 +1095,7 @@ def test_path_generation(self):
                 })
 
             t0 = time.time()
-            metadata.generate_paths(morphologist, debug=False)
+            metadata.generate_paths(morphologist)
             t1 = time.time()
             print('completion time:', t1 - t0, 's')
 
@@ -1405,4 +1431,14 @@ def test():
     return runtime.wasSuccessful()
 
 if __name__ == '__main__':
-    test()
+    morphologist = Capsul.executable('capsul.pipeline.test.fake_morphologist.morphologist.Morphologist')
+    parent = morphologist.nodes['SulciRecognition'].get_pipeline()
+    print(parent)
+    # t = TestFakeMorphologist()
+    # t.subjects = [f'subject_{i:04}' for i in range(2000)]
+    # print(f'Setting up config and data files for {len(t.subjects)} subjects and 3 time points')
+    # t.setUp()
+    # try:
+    #     ...
+    # finally:
+    #     t.tearDown()
diff --git a/capsul/test/test_tiny_morphologist.py b/capsul/test/test_tiny_morphologist.py
index b151ef7e8..a1f2f872c 100644
--- a/capsul/test/test_tiny_morphologist.py
+++ b/capsul/test/test_tiny_morphologist.py
@@ -5,13 +5,14 @@
 import shutil
 import tempfile
 import unittest
+from capsul.qt_gui.execution_gui import execution_widget
 
 from soma.controller import field, File
 from soma.controller import Directory, undefined
 
 from capsul.api import Capsul, Process, Pipeline
-from capsul.config.configuration import ModuleConfiguration, default_workers_type, default_database_url
-from capsul.dataset import ProcessMetadata, ProcessSchema
+from capsul.config.configuration import ModuleConfiguration, default_engine_database, default_engine_start_workers
+from capsul.dataset import ProcessMetadata, ProcessSchema, Prepend, Append
 
 
 class FakeSPMConfiguration(ModuleConfiguration):
@@ -46,10 +47,10 @@ def execute(self, context):
             f.write(content)
 
 class BiasCorrectionBIDS(ProcessSchema, schema='bids', process=BiasCorrection):
-    output = {'part': 'nobias'}
+    output = Prepend('part', 'nobias')
 
 class BiasCorrectionBrainVISA(ProcessSchema, schema='brainvisa', process=BiasCorrection):
-    output = {'prefix': 'nobias'}
+    output = Prepend('prefix', 'nobias')
 
 
 class FakeSPMNormalization12(Process):
@@ -80,10 +81,10 @@ def execute(self, context):
             f.write(content)
 
 class FakeSPMNormalization12BIDS(ProcessSchema, schema='bids', process=FakeSPMNormalization12):
-    output = {'part': 'normalized_fakespm12'}
+    output = Prepend('part', 'normalized_fakespm12')
 
 class FakeSPMNormalization12BrainVISA(ProcessSchema, schema='brainvisa', process=FakeSPMNormalization12):
-    output = {'prefix': 'normalized_fakespm12'}
+    output = Prepend('prefix', 'normalized_fakespm12')
 
 
 class FakeSPMNormalization8(FakeSPMNormalization12):
@@ -94,10 +95,10 @@ class FakeSPMNormalization8(FakeSPMNormalization12):
     }
 
 class FakeSPMNormalization8BIDS(ProcessSchema, schema='bids', process=FakeSPMNormalization8):
-    output = {'part': 'normalized_fakespm8'}
+    output = Prepend('part', 'normalized_fakespm8')
 
 class FakeSPMNormalization8BrainVISA(ProcessSchema, schema='brainvisa', process=FakeSPMNormalization8):
-    output = {'prefix': 'normalized_fakespm8'}
+    output = Prepend('prefix', 'normalized_fakespm8')
 
 
 class AimsNormalization(Process):
@@ -114,10 +115,10 @@ def execute(self, context):
             f.write(content)
 
 class AimsNormalizationBIDS(ProcessSchema, schema='bids', process=AimsNormalization):
-    output = {'part': 'normalized_aims'}
+    output = Prepend('part', 'normalized_aims')
 
 class AimsNormalizationBrainVISA(ProcessSchema, schema='brainvisa', process=AimsNormalization):
-    output = {'prefix': 'normalized_aims'}
+    output = Prepend('prefix', 'normalized_aims')
 
 class SplitBrain(Process):
     input: field(type_=File, extensions=('.nii',))
@@ -135,12 +136,13 @@ def execute(self, context):
                 f.write(side_content)
 
 
-class SplitBrainBIDS(ProcessSchema, schema='bids', process=SplitBrain):
-    output = {'part': 'split'}
-
-
 class SplitBrainBrainVISA(ProcessSchema, schema='brainvisa', process=SplitBrain):
-    output = {'prefix': 'split'}
+    _ = { 
+        'right_output': Append('suffix', 'right'),
+        'left_output': Append('suffix', 'left'),
+    }
+    right_output = Prepend('prefix', 'split')
+    left_output = Prepend('prefix', 'split')
 
 
 class ProcessHemisphere(Process):
@@ -155,6 +157,10 @@ def execute(self, context):
             f.write(content)
 
 
+class ProcessHemisphereBrainVISA(ProcessSchema, schema='brainvisa', process=ProcessHemisphere):
+    output = Prepend('prefix', 'hemi')
+
+
 class TinyMorphologist(Pipeline):
     def pipeline_definition(self):
         self.add_process('nobias', BiasCorrection)
@@ -189,19 +195,28 @@ def pipeline_definition(self):
 
 class TinyMorphologistBIDS(ProcessSchema, schema='bids', process=TinyMorphologist):
     _ = {
-        '*': {'process': 'tinymorphologist'}
+        '*': {'process': 'tinymorphologist'},
+        'split.right_output': {'part': 'right_hemi'},
+        'split.left_output': {'part': 'left_hemi'},
     }
     input = {'process': None}
-    left_hemisphere = {'part': 'left_hemi'}
-    right_hemisphere = {'part': 'right_hemi'}
 
 class TinyMorphologistBrainVISA(ProcessSchema, schema='brainvisa', process=TinyMorphologist):
     _ = {
-        '*': {'process': 'tinymorphologist'}
+        '*': {'process': 'tinymorphologist'},
+    }
+    _nodes = {
+        # 'split': {
+        #     'right_output': {'prefix': 'right_hemi'},
+        #     'left_output': {'prefix': 'left_hemi'}
+        # },
+        # 'fakespm_normalization_12': {'*': Append('suffix', 'fakespm12')},
+        # 'fakespm_normalization_8': {'*': Append('suffix', 'fakespm8')},
+        # 'aims_normalization': {'*': Append('suffix', 'aims')},
     }
     input = {'process': None}
-    left_hemisphere = {'prefix': 'left_hemi'}
-    right_hemisphere = {'prefix': 'right_hemi'}
+    # left_hemisphere = {'prefix': 'left_hemi'}
+    # right_hemisphere = {'prefix': 'right_hemi'}
 
 def concatenate(inputs: list[File], result: File):
     with open(result, 'w') as o:
@@ -332,7 +347,7 @@ def setUp(self):
         with self.config_file.open('w') as f:
             json.dump(config, f)
 
-        self.capsul = Capsul('test_tiny_morphologist', site_file=self.config_file)
+        self.capsul = Capsul('test_tiny_morphologist', site_file=self.config_file, user_file=None)
         return super().setUp()
 
     def tearDown(self):
@@ -345,8 +360,8 @@ def test_tiny_morphologist_config(self):
         self.maxDiff = 2000
         expected_config = {
             'builtin': {
-                'workers_type': default_workers_type,
-                'database_url': default_database_url,
+                'database': default_engine_database,
+                'start_workers': default_engine_start_workers,
                 'dataset': {
                     'input': {
                         'path': str(self.tmp / 'bids'),
@@ -423,31 +438,32 @@ def test_tiny_path_generation(self):
                 'template': '!{fakespm.directory}/template',
                 'nobias': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
                 'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
-                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander.nii',
-                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander.nii',
+                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_nobias_aleksander_left.nii',
             },
             'aims': {
                 'template': '!{fakespm.directory}/template',
                 'nobias': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
-                'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/normalized_aims_aleksander.nii',
-                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander.nii',
-                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander.nii',
+                'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/normalized_aims_nobias_aleksander.nii',
+                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
             },
             'fakespm12': {
                 'template': '!{fakespm.directory}/template',
                 'nobias': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
-                'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/normalized_fakespm12_aleksander.nii',
-                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander.nii',
-                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander.nii',
+                'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/normalized_fakespm12_nobias_aleksander.nii',
+                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm12_nobias_aleksander_right.nii',
+                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm12_nobias_aleksander_left.nii',
             },
             'fakespm8': {
                 'template': '!{fakespm.directory}/template',
                 'nobias': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
-                'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/normalized_fakespm8_aleksander.nii',
-                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander.nii',
-                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander.nii',
+                'normalized': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/normalized_fakespm8_nobias_aleksander.nii',
+                'right_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                'left_hemisphere': '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
             },
         }
+
         tiny_morphologist = self.capsul.executable('capsul.test.test_tiny_morphologist.TinyMorphologist')            
         engine = self.capsul.engine()
         execution_context = engine.execution_context(tiny_morphologist)
@@ -498,16 +514,13 @@ def test_tiny_path_generation(self):
                     'nobias', 'normalized', 'right_hemisphere', 'left_hemisphere'))
             self.maxDiff = 2000
             self.assertEqual(params, expected[normalization])
-            # for field in tiny_morphologist.fields():
-            #     value = getattr(tiny_morphologist, field.name, undefined)
-            #     print(f'!{normalization}!', field.name, value)
 
             with self.capsul.engine() as engine:
                 status = engine.run(tiny_morphologist, timeout=5)
                 self.assertEqual(status, 'ended')
 
 
-    def test_pipeline_iteration(self):
+    def test_tiny_morphologist_iteration(self):
         expected_completion = {
             'input': [
                         '!{dataset.input.path}/rawdata/sub-aleksander/ses-m0/anat/sub-aleksander_ses-m0_T1w.nii',
@@ -530,44 +543,44 @@ def test_pipeline_iteration(self):
                         '!{dataset.input.path}/rawdata/sub-casimiro/ses-m24/anat/sub-casimiro_ses-m24_T1w.nii',
             ],
             'left_hemisphere': [
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander_{executable.normalization[0]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander_{executable.normalization[1]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander_{executable.normalization[2]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/left_hemi_aleksander_{executable.normalization[3]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/left_hemi_aleksander_{executable.normalization[4]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/left_hemi_aleksander_{executable.normalization[5]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/left_hemi_aleksander_{executable.normalization[6]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/left_hemi_aleksander_{executable.normalization[7]}.nii',
-                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/left_hemi_aleksander_{executable.normalization[8]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/left_hemi_casimiro_{executable.normalization[9]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/left_hemi_casimiro_{executable.normalization[10]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/left_hemi_casimiro_{executable.normalization[11]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/left_hemi_casimiro_{executable.normalization[12]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/left_hemi_casimiro_{executable.normalization[13]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/left_hemi_casimiro_{executable.normalization[14]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/left_hemi_casimiro_{executable.normalization[15]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/left_hemi_casimiro_{executable.normalization[16]}.nii',
-                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/left_hemi_casimiro_{executable.normalization[17]}.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_casimiro_left.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_left.nii',
             ],
             'nobias': [
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/aleksander_{executable.normalization[0]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/aleksander_{executable.normalization[1]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/aleksander_{executable.normalization[2]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/aleksander_{executable.normalization[3]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/aleksander_{executable.normalization[4]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/aleksander_{executable.normalization[5]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/aleksander_{executable.normalization[6]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/aleksander_{executable.normalization[7]}.nii',
-                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/aleksander_{executable.normalization[8]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/casimiro_{executable.normalization[9]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/casimiro_{executable.normalization[10]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/casimiro_{executable.normalization[11]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/casimiro_{executable.normalization[12]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/casimiro_{executable.normalization[13]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/casimiro_{executable.normalization[14]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/casimiro_{executable.normalization[15]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/casimiro_{executable.normalization[16]}.nii',
-                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/casimiro_{executable.normalization[17]}.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/nobias_aleksander.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/nobias_casimiro.nii',
+                        '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/nobias_casimiro.nii',
             ],
             'normalization': ['none',
                                 'aims',
@@ -588,24 +601,24 @@ def test_pipeline_iteration(self):
                                 'aims',
                                 'fakespm8'],
             'right_hemisphere': [
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander_{executable.normalization[0]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander_{executable.normalization[1]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander_{executable.normalization[2]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/right_hemi_aleksander_{executable.normalization[3]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/right_hemi_aleksander_{executable.normalization[4]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/right_hemi_aleksander_{executable.normalization[5]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/right_hemi_aleksander_{executable.normalization[6]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/right_hemi_aleksander_{executable.normalization[7]}.nii',
-                                    '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/right_hemi_aleksander_{executable.normalization[8]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/right_hemi_casimiro_{executable.normalization[9]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/right_hemi_casimiro_{executable.normalization[10]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/right_hemi_casimiro_{executable.normalization[11]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/right_hemi_casimiro_{executable.normalization[12]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/right_hemi_casimiro_{executable.normalization[13]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/right_hemi_casimiro_{executable.normalization[14]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/right_hemi_casimiro_{executable.normalization[15]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/right_hemi_casimiro_{executable.normalization[16]}.nii',
-                                    '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/right_hemi_casimiro_{executable.normalization[17]}.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_casimiro_right.nii',
+                                '!{dataset.output.path}/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_right.nii',
             ],
         }
 
@@ -631,44 +644,44 @@ def test_pipeline_iteration(self):
                         f'{self.tmp}/bids/rawdata/sub-casimiro/ses-m24/anat/sub-casimiro_ses-m24_T1w.nii',
             ],
             'left_hemisphere': [
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander_none.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander_aims.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/left_hemi_aleksander_fakespm8.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/left_hemi_aleksander_none.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/left_hemi_aleksander_aims.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/left_hemi_aleksander_fakespm8.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/left_hemi_aleksander_none.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/left_hemi_aleksander_aims.nii',
-                                f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/left_hemi_aleksander_fakespm8.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/left_hemi_casimiro_none.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/left_hemi_casimiro_aims.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/left_hemi_casimiro_fakespm8.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/left_hemi_casimiro_none.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/left_hemi_casimiro_aims.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/left_hemi_casimiro_fakespm8.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/left_hemi_casimiro_none.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/left_hemi_casimiro_aims.nii',
-                                f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/left_hemi_casimiro_fakespm8.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_casimiro_left.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_left.nii',
             ],
             'nobias': [
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/aleksander_none.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/aleksander_aims.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/aleksander_fakespm8.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/aleksander_none.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/aleksander_aims.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/aleksander_fakespm8.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/aleksander_none.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/aleksander_aims.nii',
-                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/aleksander_fakespm8.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/casimiro_none.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/casimiro_aims.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/casimiro_fakespm8.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/casimiro_none.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/casimiro_aims.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/casimiro_fakespm8.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/casimiro_none.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/casimiro_aims.nii',
-                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/casimiro_fakespm8.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/nobias_aleksander.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/nobias_casimiro.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/nobias_casimiro.nii',
             ],
             'normalization': ['none',
                                 'aims',
@@ -689,24 +702,24 @@ def test_pipeline_iteration(self):
                                 'aims',
                                 'fakespm8'],
             'right_hemisphere': [
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander_none.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander_aims.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/right_hemi_aleksander_fakespm8.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/right_hemi_aleksander_none.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/right_hemi_aleksander_aims.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/right_hemi_aleksander_fakespm8.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/right_hemi_aleksander_none.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/right_hemi_aleksander_aims.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/right_hemi_aleksander_fakespm8.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/right_hemi_casimiro_none.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/right_hemi_casimiro_aims.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/right_hemi_casimiro_fakespm8.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/right_hemi_casimiro_none.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/right_hemi_casimiro_aims.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/right_hemi_casimiro_fakespm8.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/right_hemi_casimiro_none.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/right_hemi_casimiro_aims.nii',
-                                    f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/right_hemi_casimiro_fakespm8.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/aleksander/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_aleksander_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_aims_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m0/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_aims_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m12/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_aims_nobias_casimiro_right.nii',
+                        f'{self.tmp}/brainvisa/whaterver/casimiro/tinymorphologist/m24/default_analysis/hemi_split_normalized_fakespm8_nobias_casimiro_right.nii',
             ]
         }
 
@@ -715,12 +728,12 @@ def test_pipeline_iteration(self):
             non_iterative_plugs=['template'],
         )
 
-        class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', process=tiny_morphologist_iteration):
-            _ = {
-                '*': {
-                    'suffix': lambda iteration_index, **kwargs: f'{{executable.normalization[{iteration_index}]}}',
-                }
-            }
+        # class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', process=tiny_morphologist_iteration):
+        #     _ = {
+        #         '*': {
+        #             'suffix': lambda iteration_index, **kwargs: f'{{executable.normalization[{iteration_index}]}}',
+        #         }
+        #     }
 
         engine = self.capsul.engine()
         execution_context = engine.execution_context(tiny_morphologist_iteration)
@@ -734,8 +747,6 @@ class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', proc
             input_metadata = execution_context.dataset['input'].schema.metadata(path)
             inputs.extend([input_metadata]*3)
             normalizations += ['none', 'aims', 'fakespm8']
-        # Set the input data
-#        tiny_morphologist_iteration.input = inputs
         tiny_morphologist_iteration.normalization = normalizations
 
         
@@ -751,38 +762,68 @@ class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', proc
             self.assertEqual(getattr(tiny_morphologist_iteration, name), value,
                              f'Differing value for parameter {name}')
 
-        # run it
-        # Note: to run via soma-workflow, just set this:
-        # self.capsul.config.builtin.engine_type = 'soma_workflow'
+        with self.capsul.engine() as engine:
+            status = engine.run(tiny_morphologist_iteration, timeout=10)
 
-        #status = None
-        #try:
-            #with self.capsul.engine() as engine:
-                #status = engine.run(tiny_morphologist_iteration)
-        #except Exception:
-            #import traceback
-            #traceback.print_exc()
-
-        #self.assertEqual(
-            #status,
-            #{'status': 'ended', 'error': None, 'error_detail': None,
-              #'engine_output': ''})
+        self.assertEqual(
+            status,
+            'ended')
 
 
 if __name__ == '__main__':
-    # import sys
-    # sys.stdout.flush()
-    # from soma.qt_gui.qt_backend import QtGui
-    # from capsul.qt_gui.widgets import PipelineDeveloperView
-    # tiny_morphologist = Capsul.executable('capsul.test.test_tiny_morphologist.TinyMorphologist')
-    # app = QtGui.QApplication.instance()
-    # if not app:
-    #     app = QtGui.QApplication(sys.argv)
-    # view1 = PipelineDeveloperView(tiny_morphologist, show_sub_pipelines=True, allow_open_controller=True, enable_edition=True)
-    # view1.show()
-    # app.exec_()
-    # del view1
-    t = TestTinyMorphologist()
-    t.setUp()
-    t.test_tiny_path_generation()
-    t.tearDown()
+    import sys
+    from soma.qt_gui.qt_backend import Qt
+    
+    qt_app = Qt.QApplication.instance()
+    if not qt_app:
+        qt_app = Qt.QApplication(sys.argv)
+    self = TestTinyMorphologist()
+    # self.subjects = [f'subject{i:04}' for i in range(1)]
+    print(f'Setting up config and data files for {len(self.subjects)} subjects and 3 time points')
+    self.setUp()
+    try:
+        tiny_morphologist_iteration = self.capsul.executable_iteration(
+            'capsul.test.test_tiny_morphologist.TinyMorphologist',
+            non_iterative_plugs=['template'],
+        )
+
+        engine = self.capsul.engine()
+        execution_context = engine.execution_context(tiny_morphologist_iteration)
+
+        # Parse the dataset with BIDS-specific query (here "suffix" is part
+        #  of BIDS specification). The object returned contains info for main
+        # BIDS fields (sub, ses, acq, etc.)
+        inputs = []
+        normalizations = []
+        for path in sorted(self.capsul.config.builtin.dataset.input.find(suffix='T1w', extension='nii')):
+            input_metadata = execution_context.dataset['input'].schema.metadata(path)
+            inputs.extend([input_metadata]*3)
+            normalizations += ['none', 'aims', 'fakespm8']
+        tiny_morphologist_iteration.normalization = normalizations
+
+        
+        metadata = ProcessMetadata(tiny_morphologist_iteration, execution_context)
+        metadata.bids = inputs
+        metadata.generate_paths(tiny_morphologist_iteration)
+
+        with self.capsul.engine() as engine:
+            execution_id = engine.start(tiny_morphologist_iteration)
+            try:
+                widget = execution_widget(engine.database, engine.engine_id, execution_id)
+                widget.show()
+                from capsul.qt_gui.widgets import PipelineDeveloperView
+                tiny_morphologist = Capsul.executable('capsul.test.test_tiny_morphologist.TinyMorphologist')
+                # view1 = PipelineDeveloperView(tiny_morphologist, show_sub_pipelines=True, allow_open_controller=True, enable_edition=True)
+                # view1.show()
+                qt_app.exec_()
+                del widget
+                # del view1
+                engine.wait(execution_id, timeout=1000)
+                # engine.raise_for_status(execution_id)
+            except TimeoutError:
+                # engine.print_execution_report(engine.execution_report(engine.engine_id, execution_id))
+                raise
+            finally:
+                engine.dispose(execution_id)
+    finally:
+        self.tearDown()