diff --git a/esmvalcore/_recipe/recipe.py b/esmvalcore/_recipe/recipe.py
index 375f750725..f15e61e92a 100644
--- a/esmvalcore/_recipe/recipe.py
+++ b/esmvalcore/_recipe/recipe.py
@@ -1340,7 +1340,7 @@ def run(self) -> None:
             file.prepare()
         logger.info("Successfully downloaded missing data")
 
-        self.tasks.run(max_parallel_tasks=self.session["max_parallel_tasks"])
+        self.tasks.run(self.session)
         logger.info(
             "Wrote recipe with version numbers and wildcards to:\nfile://%s",
             filled_recipe,
diff --git a/esmvalcore/_task.py b/esmvalcore/_task.py
index 7f8d2a66a8..5e50ae847f 100644
--- a/esmvalcore/_task.py
+++ b/esmvalcore/_task.py
@@ -19,16 +19,21 @@
 from copy import deepcopy
 from pathlib import Path
 from shutil import which
+from typing import TYPE_CHECKING
 
 import dask
+import dask.config
 import psutil
 import yaml
 from distributed import Client
 
-from ._citation import _write_citation_files
-from ._provenance import TrackedFile, get_task_provenance
-from .config._dask import get_distributed_client
-from .config._diagnostics import DIAGNOSTICS, TAGS
+from esmvalcore._citation import _write_citation_files
+from esmvalcore._provenance import TrackedFile, get_task_provenance
+from esmvalcore.config._dask import get_distributed_client
+from esmvalcore.config._diagnostics import DIAGNOSTICS, TAGS
+
+if TYPE_CHECKING:
+    from esmvalcore.config import Session
 
 logger = logging.getLogger(__name__)
 
@@ -270,20 +275,24 @@ def flatten(self):
         tasks.add(self)
         return tasks
 
-    def run(self, input_files: list[str] | None = None) -> None:
+    def run(
+        self,
+        session: Session,
+        input_files: list[str] | None = None,
+    ) -> list[str]:
         """Run task."""
         if not self.output_files:
             if input_files is None:
                 input_files = []
             for task in self.ancestors:
-                input_files.extend(task.run())
+                input_files.extend(task.run(session))
             logger.info(
                 "Starting task %s in process [%s]",
                 self.name,
                 os.getpid(),
             )
             start = datetime.datetime.now()
-            self.output_files = self._run(input_files)
+            self.output_files = self._run(input_files, session=session)
             runtime = datetime.datetime.now() - start
             logger.info(
                 "Successfully completed task %s (priority %s) in %s",
@@ -295,7 +304,7 @@ def run(self, input_files: list[str] | None = None) -> None:
         return self.output_files
 
     @abc.abstractmethod
-    def _run(self, input_files: list[str]) -> list[str]:
+    def _run(self, input_files: list[str], session: Session) -> list[str]:
         """Run task."""
 
     def get_product_attributes(self) -> dict:
@@ -349,7 +358,11 @@ def __init__(self, prev_preproc_dir, preproc_dir, name):
 
         super().__init__(ancestors=None, name=name, products=products)
 
-    def _run(self, _: list[str]) -> list[str]:
+    def _run(
+        self,
+        input_files: list[str],  # noqa: ARG002
+        session: Session,  # noqa: ARG002
+    ) -> list[str]:
         """Return the result of a previous run."""
         metadata = self.get_product_attributes()
 
@@ -588,7 +601,11 @@ def _start_diagnostic_script(self, cmd, env):
             env=complete_env,
         )
 
-    def _run(self, input_files):
+    def _run(
+        self,
+        input_files: list[str],
+        session: Session,  # noqa: ARG002
+    ) -> list[str]:
         """Run the diagnostic script."""
         if self.script is None:  # Run only preprocessor
             return []
@@ -803,17 +820,17 @@ def get_independent(self) -> TaskSet:
                 independent_tasks.add(task)
         return independent_tasks
 
-    def run(self, max_parallel_tasks: int | None = None) -> None:
+    def run(self, session: Session) -> None:
         """Run tasks.
 
         Parameters
         ----------
-        max_parallel_tasks : int
-            Number of processes to run. If `1`, run the tasks sequentially.
+        session:
+            The current session.
         """
-        with get_distributed_client() as client:
+        with get_distributed_client(session) as client:
             if client is None:
-                address = None
+                address: str | None = None
             else:
                 address = client.scheduler.address
                 for task in self.flatten():
@@ -825,19 +842,19 @@ def run(self, max_parallel_tasks: int | None = None) -> None:
                         # Python script.
                         task.settings["scheduler_address"] = address
 
-            if max_parallel_tasks == 1:
-                self._run_sequential()
+            if session["max_parallel_tasks"] == 1:
+                self._run_sequential(session)
             else:
-                self._run_parallel(address, max_parallel_tasks)
+                self._run_parallel(address, session)
 
-    def _run_sequential(self) -> None:
+    def _run_sequential(self, session: Session) -> None:
         """Run tasks sequentially."""
         n_tasks = len(self.flatten())
         logger.info("Running %s tasks sequentially", n_tasks)
 
         tasks = self.get_independent()
         for task in sorted(tasks, key=lambda t: t.priority):
-            task.run()
+            task.run(session)
 
     def _get_dask_config(self, max_parallel_tasks: int) -> dict:
         """Configure the threaded Dask scheduler.
@@ -884,14 +901,19 @@ def _get_dask_config(self, max_parallel_tasks: int) -> dict:
         )
         return {"num_workers": n_workers}
 
-    def _run_parallel(self, scheduler_address, max_parallel_tasks):
+    def _run_parallel(
+        self,
+        scheduler_address: str | None,
+        session: Session,
+    ) -> None:
         """Run tasks in parallel."""
         scheduled = self.flatten()
-        running = {}
+        running: dict[BaseTask, multiprocessing.pool.ApplyResult] = {}
 
         n_tasks = n_scheduled = len(scheduled)
         n_running = 0
 
+        max_parallel_tasks = session["max_parallel_tasks"]
         if max_parallel_tasks is None:
             max_parallel_tasks = available_cpu_count()
         max_parallel_tasks = min(max_parallel_tasks, n_tasks)
@@ -930,7 +952,12 @@ def done(task):
                         if all(done(t) for t in task.ancestors):
                             future = pool.apply_async(
                                 _run_task,
-                                [task, scheduler_address, scheduler_lock],
+                                [
+                                    task,
+                                    session,
+                                    scheduler_address,
+                                    scheduler_lock,
+                                ],
                             )
                             running[task] = future
                             scheduled.remove(task)
@@ -971,7 +998,12 @@ def _copy_results(task, future):
     task.output_files, task.products = future.get()
 
 
-def _run_task(task, scheduler_address, scheduler_lock):
+def _run_task(
+    task: BaseTask,
+    session: Session,
+    scheduler_address: str | None,
+    scheduler_lock: threading.Lock | None,
+) -> tuple[list[str], set[TrackedFile]]:
     """Run task and return the result."""
     if scheduler_address is None:
         client = contextlib.nullcontext()
@@ -980,6 +1012,6 @@ def _run_task(task, scheduler_address, scheduler_lock):
 
     with client:
         task.scheduler_lock = scheduler_lock
-        output_files = task.run()
+        output_files = task.run(session)
 
     return output_files, task.products
diff --git a/esmvalcore/config/_dask.py b/esmvalcore/config/_dask.py
index 10c2195101..c71d4cae7f 100644
--- a/esmvalcore/config/_dask.py
+++ b/esmvalcore/config/_dask.py
@@ -12,16 +12,15 @@
 import dask.config
 from distributed import Client
 
-from esmvalcore.config import CFG
-from esmvalcore.exceptions import (
-    InvalidConfigParameter,
-)
+from esmvalcore.exceptions import InvalidConfigParameter
 
 if TYPE_CHECKING:
     from collections.abc import Generator
 
     from distributed.deploy import Cluster
 
+    from esmvalcore.config import Session
+
 logger = logging.getLogger(__name__)
 
 
@@ -76,9 +75,9 @@ def validate_dask_config(dask_config: Mapping) -> None:
 
 
 @contextlib.contextmanager
-def get_distributed_client() -> Generator[None | Client]:
+def get_distributed_client(session: Session) -> Generator[None | Client]:
     """Get a Dask distributed client."""
-    dask_config = deepcopy(CFG["dask"])
+    dask_config = deepcopy(session["dask"])
     validate_dask_config(dask_config)
 
     # Set up cluster and client according to the selected profile
diff --git a/esmvalcore/config/_logging.py b/esmvalcore/config/_logging.py
index 9332c133f2..c77d4232b6 100644
--- a/esmvalcore/config/_logging.py
+++ b/esmvalcore/config/_logging.py
@@ -78,7 +78,7 @@ def _purge_file_handlers(cfg: dict) -> None:
 def _get_log_files(
     cfg: dict,
     output_dir: os.PathLike | str | None = None,
-) -> list:
+) -> list[str]:
     """Initialize log files for the file handlers."""
     log_files = []
 
@@ -114,21 +114,21 @@ def configure_logging(
     cfg_file: os.PathLike | str | None = None,
     output_dir: os.PathLike | str | None = None,
     console_log_level: str | None = None,
-) -> list:
+) -> list[str]:
     """Configure logging.
 
     Parameters
     ----------
-    cfg_file : str, optional
+    cfg_file:
         Logging config file. If `None`, defaults to `configure-logging.yml`
-    output_dir : str, optional
+    output_dir:
         Output directory for the log files. If `None`, log only to the console.
-    console_log_level : str, optional
+    console_log_level:
         If `None`, use the default (INFO).
 
     Returns
     -------
-    log_files : list
+    :
         Filenames that will be logged to.
     """
     if cfg_file is None:
diff --git a/esmvalcore/preprocessor/__init__.py b/esmvalcore/preprocessor/__init__.py
index 798edb36a8..3c17265530 100644
--- a/esmvalcore/preprocessor/__init__.py
+++ b/esmvalcore/preprocessor/__init__.py
@@ -117,6 +117,7 @@
     from dask.delayed import Delayed
     from iris.cube import CubeList
 
+    from esmvalcore.config import Session
     from esmvalcore.dataset import Dataset
     from esmvalcore.typing import FacetValue
 
@@ -855,7 +856,11 @@ def __init__(
         self.debug = debug
         self.write_ncl_interface = write_ncl_interface
 
-    def _run(self, _: list[str]) -> list[str]:  # noqa: C901,PLR0912
+    def _run(  # noqa: C901,PLR0912
+        self,
+        input_files: list[str],  # noqa: ARG002
+        session: Session,
+    ) -> list[str]:
         """Run the preprocessor."""
         for product in self.products:
             product.activity = self.activity
@@ -906,7 +911,11 @@ def _run(self, _: list[str]) -> list[str]:  # noqa: C901,PLR0912
                 "Computing and saving data for preprocessing task %s",
                 self.name,
             )
-            _compute_with_progress(delayeds, description=self.name)
+            _compute_with_progress(
+                delayeds,
+                session=session,
+                description=self.name,
+            )
         finally:
             if self.scheduler_lock is not None:
                 self.scheduler_lock.release()
diff --git a/esmvalcore/preprocessor/_dask_progress.py b/esmvalcore/preprocessor/_dask_progress.py
index 58edde625d..bcfa7ffb63 100644
--- a/esmvalcore/preprocessor/_dask_progress.py
+++ b/esmvalcore/preprocessor/_dask_progress.py
@@ -8,12 +8,14 @@
 import time
 from typing import TYPE_CHECKING, Any
 
+import dask
 import dask.diagnostics
+import dask.utils
 import distributed
 import distributed.diagnostics.progressbar
 import rich.progress
 
-from esmvalcore.config import CFG
+from esmvalcore.config._logging import configure_logging
 
 if TYPE_CHECKING:
     import contextlib
@@ -21,15 +23,14 @@
 
     from dask.delayed import Delayed
 
+    from esmvalcore.config import Session
+
 logger = logging.getLogger(__name__)
 
 
 class RichProgressBar(dask.diagnostics.Callback):
     """Progress bar using `rich` for the Dask threaded scheduler."""
 
-    # Disable warnings about design choices that have been made in the base class.
-    # pylint: disable=method-hidden,super-init-not-called,too-few-public-methods,unused-argument,useless-suppression
-
     # Adapted from https://github.com/dask/dask/blob/0f3e5ff6e642e7661b3f855bfd192a6f6fb83b49/dask/diagnostics/progress.py#L32-L153
     def __init__(self):
         self.progress = rich.progress.Progress(
@@ -87,9 +88,6 @@ class RichDistributedProgressBar(
 ):
     """Progress bar using `rich` for the Dask distributed scheduler."""
 
-    # Disable warnings about design choices that have been made in the base class.
-    # pylint: disable=too-few-public-methods,unused-argument,useless-suppression
-
     def __init__(self, keys) -> None:  # noqa: ANN001
         self.progress = rich.progress.Progress(
             rich.progress.TaskProgressColumn(),
@@ -121,14 +119,17 @@ def _draw_stop(self, **kwargs):
 class ProgressLogger(dask.diagnostics.ProgressBar):
     """Progress logger for the Dask threaded scheduler."""
 
-    # Disable warnings about design choices that have been made in the base class.
-    # pylint: disable=too-few-public-methods,unused-argument,useless-suppression
-
     def __init__(
         self,
         log_interval: str | float = "1s",
+        log_level: str = "INFO",
+        log_dir: str | None = None,
         description: str = "",
     ) -> None:
+        configure_logging(
+            output_dir=log_dir,
+            console_log_level=log_level,
+        )
         self._desc = f"{description} " if description else description
         self._log_interval = dask.utils.parse_timedelta(
             log_interval,
@@ -159,15 +160,18 @@ class DistributedProgressLogger(
 ):
     """Progress logger for the Dask distributed scheduler."""
 
-    # Disable warnings about design choices that have been made in the base class.
-    # pylint: disable=too-few-public-methods,unused-argument,useless-suppression
-
     def __init__(
         self,
         keys,  # noqa: ANN001
         log_interval: str | float = "1s",
+        log_level: str = "INFO",
+        log_dir: str | None = None,
         description: str = "",
     ) -> None:
+        configure_logging(
+            output_dir=log_dir,
+            console_log_level=log_level,
+        )
         self._desc = f"{description} " if description else description
         self._log_interval = dask.utils.parse_timedelta(
             log_interval,
@@ -204,6 +208,7 @@ def _draw_stop(self, **kwargs):
 
 def _compute_with_progress(
     delayeds: Iterable[Delayed],
+    session: Session,
     description: str,
 ) -> None:
     """Compute delayeds while displaying a progress bar."""
@@ -213,13 +218,12 @@ def _compute_with_progress(
     except ValueError:
         use_distributed = False
 
-    log_progress_interval = CFG["logging"]["log_progress_interval"]
+    log_progress_interval = session["logging"]["log_progress_interval"]
     if isinstance(log_progress_interval, (str, datetime.timedelta)):
         log_progress_interval = dask.utils.parse_timedelta(
             log_progress_interval,
         )
-
-    if CFG["max_parallel_tasks"] != 1 and log_progress_interval == 0.0:
+    if session["max_parallel_tasks"] != 1 and log_progress_interval == 0.0:
         # Enable progress logging if `max_parallel_tasks` > 1 to avoid clutter.
         log_progress_interval = 10.0
 
@@ -238,8 +242,10 @@ def _compute_with_progress(
         else:
             DistributedProgressLogger(
                 futures,
-                log_interval=log_progress_interval,
                 description=description,
+                log_interval=log_progress_interval,
+                log_level=session["log_level"],
+                log_dir=session.run_dir,
             )
         dask.compute(futures)
     else:
@@ -249,6 +255,8 @@ def _compute_with_progress(
             ctx = ProgressLogger(
                 description=description,
                 log_interval=log_progress_interval,
+                log_level=session["log_level"],
+                log_dir=session.run_dir,
             )
         with ctx:
             dask.compute(delayeds)
diff --git a/tests/integration/preprocessor/test_preprocessing_task.py b/tests/integration/preprocessor/test_preprocessing_task.py
index 3340a6c5eb..835fd3f0e6 100644
--- a/tests/integration/preprocessor/test_preprocessing_task.py
+++ b/tests/integration/preprocessor/test_preprocessing_task.py
@@ -1,6 +1,9 @@
 """Tests for `esmvalcore.preprocessor.PreprocessingTask`."""
 
+from __future__ import annotations
+
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import iris
 import iris.cube
@@ -12,9 +15,19 @@
 from esmvalcore.io.local import LocalFile
 from esmvalcore.preprocessor import PreprocessingTask, PreprocessorFile
 
+if TYPE_CHECKING:
+    import pytest_mock
+
+    from esmvalcore.config import Session
+
 
 @pytest.mark.parametrize("scheduler_lock", [False, True])
-def test_load_save_task(tmp_path, mocker, scheduler_lock):
+def test_load_save_task(
+    tmp_path: Path,
+    mocker: pytest_mock.MockerFixture,
+    session: Session,
+    scheduler_lock: bool,
+) -> None:
     """Test that a task that just loads and saves a file."""
     # Prepare a test dataset
     cube = iris.cube.Cube(data=[273.0], var_name="tas", units="K")
@@ -22,7 +35,12 @@ def test_load_save_task(tmp_path, mocker, scheduler_lock):
     iris.save(cube, in_file)
     dataset = Dataset(short_name="tas")
     dataset.files = [in_file]
-    dataset.load = lambda: in_file.to_iris()[0]
+    mocker.patch.object(
+        dataset,
+        "load",
+        autospec=True,
+        side_effect=lambda: in_file.to_iris()[0],
+    )
 
     # Create task
     task = PreprocessingTask(
@@ -44,7 +62,8 @@ def test_load_save_task(tmp_path, mocker, scheduler_lock):
     if scheduler_lock:
         task.scheduler_lock = mocker.Mock()
 
-    task.run()
+    session.run_dir.mkdir(parents=True)
+    task.run(session)
 
     assert len(task.products) == 1
     preproc_file = task.products.pop().filename
@@ -60,7 +79,11 @@ def test_load_save_task(tmp_path, mocker, scheduler_lock):
         assert task.scheduler_lock is None
 
 
-def test_load_save_and_other_task(tmp_path, monkeypatch):
+def test_load_save_and_other_task(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    session: Session,
+) -> None:
     """Test that a task just copies one file and preprocesses another file."""
     # Prepare test datasets
     in_cube = iris.cube.Cube(data=[0.0], var_name="tas", units="degrees_C")
@@ -74,11 +97,19 @@ def test_load_save_and_other_task(tmp_path, monkeypatch):
 
     dataset1 = Dataset(short_name="tas", dataset="dataset1")
     dataset1.files = [file1]
-    dataset1.load = lambda: file1.to_iris()[0]
+    monkeypatch.setattr(
+        dataset1,
+        "load",
+        lambda: file1.to_iris()[0],
+    )
 
     dataset2 = Dataset(short_name="tas", dataset="dataset1")
     dataset2.files = [file2]
-    dataset2.load = lambda: file2.to_iris()[0]
+    monkeypatch.setattr(
+        dataset2,
+        "load",
+        lambda: file2.to_iris()[0],
+    )
 
     # Create some mock preprocessor functions and patch
     # `esmvalcore.preprocessor` so it uses them.
@@ -166,7 +197,8 @@ def multi_preproc_func(products, output_products):
     activity = provenance.activity("software:esmvalcore")
     task.initialize_provenance(activity)
 
-    task.run()
+    session.run_dir.mkdir(parents=True)
+    task.run(session)
 
     # Check that three files were saved and the preprocessor functions were
     # only applied to the second one.
diff --git a/tests/integration/recipe/test_recipe.py b/tests/integration/recipe/test_recipe.py
index 17a0521120..e0a7b8690c 100644
--- a/tests/integration/recipe/test_recipe.py
+++ b/tests/integration/recipe/test_recipe.py
@@ -2596,9 +2596,7 @@ def test_recipe_run(tmp_path, patched_datafinder, session, mocker):
 
     esmvalcore.io.esgf.download.assert_called()
     esmvalcore.io.local.LocalFile.prepare.assert_called()
-    recipe.tasks.run.assert_called_once_with(
-        max_parallel_tasks=session["max_parallel_tasks"],
-    )
+    recipe.tasks.run.assert_called_once_with(session)
     recipe.write_filled_recipe.assert_called_once()
     recipe.write_html_summary.assert_called_once()
 
diff --git a/tests/integration/test_diagnostic_run.py b/tests/integration/test_diagnostic_run.py
index 7d8f783975..71d20c76c0 100644
--- a/tests/integration/test_diagnostic_run.py
+++ b/tests/integration/test_diagnostic_run.py
@@ -19,7 +19,7 @@ def get_mock_distributed_client(monkeypatch):
     """Mock `get_distributed_client` to avoid starting a Dask cluster."""
 
     @contextlib.contextmanager
-    def get_distributed_client():
+    def get_distributed_client(session):
         yield None
 
     monkeypatch.setattr(
diff --git a/tests/integration/test_task.py b/tests/integration/test_task.py
index 4329ec4310..0ddf260ba2 100644
--- a/tests/integration/test_task.py
+++ b/tests/integration/test_task.py
@@ -32,7 +32,7 @@
 
 
 class MockBaseTask(BaseTask):
-    def _run(self, input_files):
+    def _run(self, input_files, session):
         tmp_path = self._tmp_path
         output_file = tmp_path / self.name
 
@@ -77,7 +77,7 @@ def get_distributed_client_mock(client):
     """Mock `get_distributed_client` to avoid starting a Dask cluster."""
 
     @contextmanager
-    def get_distributed_client():
+    def get_distributed_client(session):
         yield client
 
     return get_distributed_client
@@ -93,7 +93,13 @@ def get_distributed_client():
         ("spawn", 2),
     ],
 )
-def test_run_tasks(monkeypatch, max_parallel_tasks, example_tasks, mpmethod):
+def test_run_tasks(
+    monkeypatch,
+    session,
+    max_parallel_tasks,
+    example_tasks,
+    mpmethod,
+):
     """Check that tasks are run correctly."""
     monkeypatch.setattr(
         esmvalcore._task,
@@ -105,14 +111,20 @@ def test_run_tasks(monkeypatch, max_parallel_tasks, example_tasks, mpmethod):
         "Pool",
         multiprocessing.get_context(mpmethod).Pool,
     )
-    example_tasks.run(max_parallel_tasks=max_parallel_tasks)
+    session["max_parallel_tasks"] = max_parallel_tasks
+    example_tasks.run(session=session)
 
     for task in example_tasks:
         print(task.name, task.output_files)
         assert task.output_files
 
 
-def test_diag_task_updated_with_address(monkeypatch, mocker, tmp_path):
+def test_diag_task_updated_with_address(
+    monkeypatch,
+    mocker,
+    tmp_path,
+    session,
+):
     """Test that the scheduler address is passed to the diagnostic tasks."""
     # Set up mock Dask distributed client
     client = mocker.Mock()
@@ -134,7 +146,8 @@ def test_diag_task_updated_with_address(monkeypatch, mocker, tmp_path):
     mocker.patch.object(TaskSet, "_run_sequential")
     tasks = TaskSet()
     tasks.add(task)
-    tasks.run(max_parallel_tasks=1)
+    session["max_parallel_tasks"] = 1
+    tasks.run(session=session)
 
     # Check that the scheduler address was added to the
     # diagnostic task settings.
@@ -149,15 +162,15 @@ def test_diag_task_updated_with_address(monkeypatch, mocker, tmp_path):
         partial(
             TaskSet._run_parallel,
             scheduler_address=None,
-            max_parallel_tasks=1,
         ),
     ],
 )
-def test_runner_uses_priority(monkeypatch, runner, example_tasks):
+def test_runner_uses_priority(monkeypatch, session, runner, example_tasks):
     """Check that the runner tries to respect task priority."""
+    session["max_parallel_tasks"] = 1
     order = []
 
-    def _run(self, input_files):
+    def _run(self, input_files, session):
         print(f"running task {self.name} with priority {self.priority}")
         order.append(self.priority)
         return [f"{self.name}_test.nc"]
@@ -165,14 +178,14 @@ def _run(self, input_files):
     monkeypatch.setattr(MockBaseTask, "_run", _run)
     monkeypatch.setattr(esmvalcore._task.multiprocessing, "Pool", ThreadPool)
 
-    runner(example_tasks)
+    runner(example_tasks, session=session)
     print(order)
     assert len(order) == 12
     assert order == sorted(order)
 
 
 @pytest.mark.parametrize("address", [None, "localhost:1234"])
-def test_run_task(mocker, address):
+def test_run_task(mocker, session, address):
     # Set up mock Dask distributed client
     mocker.patch.object(esmvalcore._task, "Client")
 
@@ -185,6 +198,7 @@ def test_run_task(mocker, address):
         task,
         scheduler_address=address,
         scheduler_lock=scheduler_lock,
+        session=session,
     )
     assert output_files == task.run.return_value
     assert products == task.products
diff --git a/tests/sample_data/experimental/test_run_recipe.py b/tests/sample_data/experimental/test_run_recipe.py
index 03e290ea21..baa7be067c 100644
--- a/tests/sample_data/experimental/test_run_recipe.py
+++ b/tests/sample_data/experimental/test_run_recipe.py
@@ -39,7 +39,7 @@ def get_mock_distributed_client(monkeypatch):
     """Mock `get_distributed_client` to avoid starting a Dask cluster."""
 
     @contextmanager
-    def get_distributed_client():
+    def get_distributed_client(session):
         yield None
 
     monkeypatch.setattr(
diff --git a/tests/unit/config/test_dask.py b/tests/unit/config/test_dask.py
index 70f5ca9ef8..386a5a9df4 100644
--- a/tests/unit/config/test_dask.py
+++ b/tests/unit/config/test_dask.py
@@ -1,14 +1,27 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 import pytest
 
-from esmvalcore.config import CFG, _dask
+from esmvalcore.config import _dask
 from esmvalcore.exceptions import (
     InvalidConfigParameter,
 )
 
+if TYPE_CHECKING:
+    from unittest.mock import Mock
+
+    import pytest_mock
+
+    from esmvalcore.config import Session
+
 
 @pytest.fixture
-def mock_dask_config_set(mocker):
-    dask_config_dict = {}
+def mock_dask_config_set(
+    mocker: pytest_mock.MockerFixture,
+) -> Mock:
+    dask_config_dict: dict = {}
     mock_dask_set = mocker.patch("dask.config.set", autospec=True)
     mock_dask_set.side_effect = dask_config_dict.update
     mock_dask_get = mocker.patch("dask.config.get", autospec=True)
@@ -16,29 +29,24 @@ def mock_dask_config_set(mocker):
     return mock_dask_set
 
 
-def test_get_no_distributed_client(ignore_existing_user_config):
-    with _dask.get_distributed_client() as client:
+def test_get_no_distributed_client(session: Session) -> None:
+    with _dask.get_distributed_client(session) as client:
         assert client is None
 
 
 def test_get_distributed_client_external(
-    monkeypatch,
-    mocker,
-    mock_dask_config_set,
-    ignore_existing_user_config,
-):
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "external",
-            "profiles": {
-                "external": {
-                    "scheduler_address": "tcp://127.0.0.1:42021",
-                },
+    mocker: pytest_mock.MockerFixture,
+    mock_dask_config_set: Mock,
+    session: Session,
+) -> None:
+    session["dask"] = {
+        "use": "external",
+        "profiles": {
+            "external": {
+                "scheduler_address": "tcp://127.0.0.1:42021",
             },
         },
-    )
+    }
 
     # Create mock distributed.Client
     mock_client = mocker.Mock()
@@ -49,10 +57,10 @@ def test_get_distributed_client_external(
         return_value=mock_client,
     )
 
-    with _dask.get_distributed_client() as client:
+    with _dask.get_distributed_client(session) as client:
         assert client is mock_client
     _dask.Client.assert_called_once_with()
-    mock_client.close.assert_called_once_with()
+    mock_client.close.assert_called_once_with()  # type: ignore[union-attr]
     assert (
         mocker.call({"scheduler_address": "tcp://127.0.0.1:42021"})
         in mock_dask_config_set.mock_calls
@@ -61,31 +69,26 @@ def test_get_distributed_client_external(
 
 @pytest.mark.parametrize("shutdown_timeout", [False, True])
 def test_get_distributed_client_slurm(
-    monkeypatch,
-    mocker,
-    mock_dask_config_set,
-    ignore_existing_user_config,
-    shutdown_timeout,
-):
+    mocker: pytest_mock.MockerFixture,
+    mock_dask_config_set: Mock,
+    session: Session,
+    shutdown_timeout: bool,
+) -> None:
     slurm_cluster = {
         "type": "dask_jobqueue.SLURMCluster",
         "queue": "interactive",
         "cores": "8",
         "memory": "16GiB",
     }
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "slurm_cluster",
-            "profiles": {
-                "slurm_cluster": {
-                    "cluster": slurm_cluster,
-                    "num_workers": 42,
-                },
+    session["dask"] = {
+        "use": "slurm_cluster",
+        "profiles": {
+            "slurm_cluster": {
+                "cluster": slurm_cluster,
+                "num_workers": 42,
             },
         },
-    )
+    }
 
     # Create mock distributed.Client
     mock_client = mocker.Mock()
@@ -108,9 +111,9 @@ def test_get_distributed_client_slurm(
     mock_cluster = mock_cluster_cls.return_value
     if shutdown_timeout:
         mock_cluster.close.side_effect = TimeoutError
-    with _dask.get_distributed_client() as client:
+    with _dask.get_distributed_client(session) as client:
         assert client is mock_client
-    mock_client.close.assert_called_once_with()
+    mock_client.close.assert_called_once_with()  # type: ignore[union-attr]
     _dask.Client.assert_called_once_with()
     args = {k: v for k, v in slurm_cluster.items() if k != "type"}
     mock_cluster_cls.assert_called_once_with(**args)
@@ -123,21 +126,16 @@ def test_get_distributed_client_slurm(
 
 
 def test_custom_default_scheduler(
-    monkeypatch,
-    mock_dask_config_set,
-    ignore_existing_user_config,
-):
+    mock_dask_config_set: Mock,
+    session: Session,
+) -> None:
     default_scheduler = {"num_workers": 42, "scheduler": "processes"}
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "process_scheduler",
-            "profiles": {"process_scheduler": default_scheduler},
-        },
-    )
+    session["dask"] = {
+        "use": "process_scheduler",
+        "profiles": {"process_scheduler": default_scheduler},
+    }
 
-    with _dask.get_distributed_client() as client:
+    with _dask.get_distributed_client(session) as client:
         assert client is None
 
     mock_dask_config_set.assert_called_with(
@@ -145,30 +143,30 @@ def test_custom_default_scheduler(
     )
 
 
-def test_invalid_dask_config_no_profiles(monkeypatch):
-    monkeypatch.setitem(CFG, "dask", {})
+def test_invalid_dask_config_no_profiles(session: Session) -> None:
+    session["dask"] = {}
 
     msg = "Key 'profiles' needs to be defined for 'dask' configuration"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
 
 
-def test_invalid_dask_config_no_use(monkeypatch):
-    monkeypatch.setitem(CFG, "dask", {"profiles": {}})
+def test_invalid_dask_config_no_use(session: Session) -> None:
+    session["dask"] = {"profiles": {}}
 
     msg = "Key 'use' needs to be defined for 'dask' configuration"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
 
 
-def test_invalid_dask_config_invalid_profiles(monkeypatch):
-    monkeypatch.setitem(CFG, "dask", {"use": "test", "profiles": 1})
+def test_invalid_dask_config_invalid_profiles(session: Session) -> None:
+    session["dask"] = {"use": "test", "profiles": 1}
 
     msg = "Key 'dask.profiles' needs to be a mapping, got"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
 
 
@@ -177,75 +175,62 @@ def test_invalid_dask_config_invalid_profiles(monkeypatch):
     ["scheduler_address", "scheduler-address"],
 )
 def test_invalid_dask_config_profile_with_cluster_and_address(
-    monkeypatch,
-    address_name,
-):
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "test",
-            "profiles": {
-                "test": {"cluster": {}, address_name: "8786"},
+    session: Session,
+    address_name: str,
+) -> None:
+    session["dask"] = {
+        "use": "test",
+        "profiles": {
+            "test": {
+                "cluster": {},
+                address_name: "8786",
             },
         },
-    )
+    }
 
     msg = "Key 'dask.profiles.test' uses 'cluster' and 'scheduler_address'"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
 
 
-def test_invalid_dask_config_profile_invalid_cluster(monkeypatch):
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "test",
-            "profiles": {
-                "test": {"cluster": 1},
-            },
+def test_invalid_dask_config_profile_invalid_cluster(session: Session) -> None:
+    session["dask"] = {
+        "use": "test",
+        "profiles": {
+            "test": {"cluster": 1},
         },
-    )
+    }
 
     msg = "Key 'dask.profiles.test.cluster' needs to be a mapping"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
 
 
-def test_invalid_dask_config_cluster_no_type(monkeypatch):
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "test",
-            "profiles": {
-                "test": {"cluster": {}},
-            },
+def test_invalid_dask_config_cluster_no_type(session: Session) -> None:
+    session["dask"] = {
+        "use": "test",
+        "profiles": {
+            "test": {"cluster": {}},
         },
-    )
+    }
 
     msg = "Key 'dask.profiles.test.cluster' does not have a 'type'"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
 
 
-def test_invalid_dask_config_invalid_use(monkeypatch):
-    monkeypatch.setitem(
-        CFG,
-        "dask",
-        {
-            "use": "not_in_profiles",
-            "profiles": {
-                "test": {},
-            },
+def test_invalid_dask_config_invalid_use(session: Session) -> None:
+    session["dask"] = {
+        "use": "not_in_profiles",
+        "profiles": {
+            "test": {},
         },
-    )
+    }
 
     msg = "Key 'dask.use' needs to point to an element of 'dask.profiles'"
     with pytest.raises(InvalidConfigParameter, match=msg):
-        with _dask.get_distributed_client():
+        with _dask.get_distributed_client(session):
             pass
diff --git a/tests/unit/preprocessor/test_dask_progress.py b/tests/unit/preprocessor/test_dask_progress.py
index 04aeaafad8..22ac8ccd55 100644
--- a/tests/unit/preprocessor/test_dask_progress.py
+++ b/tests/unit/preprocessor/test_dask_progress.py
@@ -1,7 +1,9 @@
 """Test :mod:`esmvalcore.preprocessor._dask_progress`."""
 
-import logging
+from __future__ import annotations
+
 import time
+from typing import TYPE_CHECKING
 
 import dask
 import distributed
@@ -9,43 +11,50 @@
 
 from esmvalcore.preprocessor import _dask_progress
 
+if TYPE_CHECKING:
+    from esmvalcore.config import Session
+
 
 @pytest.mark.parametrize("use_distributed", [False, True])
 @pytest.mark.parametrize("interval", [-1, 0.0, 0.2])
 def test_compute_with_progress(
-    capsys,
-    caplog,
-    monkeypatch,
-    use_distributed,
-    interval,
-):
-    caplog.set_level(logging.INFO)
+    capsys: pytest.CaptureFixture,
+    session: Session,
+    use_distributed: bool,
+    interval: float,
+) -> None:
     if use_distributed:
         client = distributed.Client(n_workers=1, threads_per_worker=1)
     else:
         client = None
 
-    monkeypatch.setitem(_dask_progress.CFG, "max_parallel_tasks", 1)
-    monkeypatch.setitem(
-        _dask_progress.CFG["logging"],
-        "log_progress_interval",
-        f"{interval}s" if interval > 0 else interval,
+    session["log_level"] = "INFO"
+    session["max_parallel_tasks"] = 1
+    session["logging"]["log_progress_interval"] = (
+        f"{interval}s" if interval > 0 else interval
     )
+    session.run_dir.mkdir(parents=True)
 
     def func(delay: float) -> None:
         time.sleep(delay)
 
     delayeds = [dask.delayed(func)(0.11)]
-    _dask_progress._compute_with_progress(delayeds, description="test")
-    if interval == 0.0:  # noqa: SIM108
-        # Assert that some progress bar has been written to stdout.
-        progressbar = capsys.readouterr().out
-    else:
-        # Assert that some progress bar has been logged.
-        progressbar = caplog.text
+    _dask_progress._compute_with_progress(
+        delayeds,
+        session=session,
+        description="test",
+    )
+    progressbar = capsys.readouterr().out
+    print(progressbar)
     if interval < 0.0:
         assert not progressbar
     else:
         assert "100%" in progressbar
+        if interval == 0.0:
+            # Assert that Rich progress bar has been written to stdout.
+            assert "1/1" in progressbar
+        else:
+            # Assert that progress bar has been logged.
+            assert "####" in progressbar
     if client is not None:
         client.shutdown()
diff --git a/tests/unit/task/test_diagnostic_task.py b/tests/unit/task/test_diagnostic_task.py
index 3566d67922..ea8333d37e 100644
--- a/tests/unit/task/test_diagnostic_task.py
+++ b/tests/unit/task/test_diagnostic_task.py
@@ -17,6 +17,8 @@
 if TYPE_CHECKING:
     from pytest_mock import MockerFixture
 
+    from esmvalcore.config import Session
+
 
 def test_write_ncl_settings(tmp_path):
     """Test minimally write_ncl_settings()."""
@@ -349,6 +351,7 @@ def test_collect_provenance_ancestor_hint(mocker, caplog, diagnostic_task):
 def test_run_fails_with_nonzero_returncode(
     mocker: MockerFixture,
     tmp_path: Path,
+    session: Session,
 ) -> None:
     """Test that DiagnosticError is raised when script fails."""
     mocker.patch.object(
@@ -395,4 +398,4 @@ def test_run_fails_with_nonzero_returncode(
 
     msg = r"Diagnostic script test.py failed with return code 1"
     with pytest.raises(DiagnosticError, match=re.escape(msg)):
-        task._run(input_files=[])
+        task._run(input_files=[], session=session)
diff --git a/tests/unit/task/test_resume_task.py b/tests/unit/task/test_resume_task.py
index 9b35c00558..1d5092b9da 100644
--- a/tests/unit/task/test_resume_task.py
+++ b/tests/unit/task/test_resume_task.py
@@ -1,9 +1,18 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 import yaml
 
 from esmvalcore._task import ResumeTask
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from esmvalcore.config import Session
+
 
-def test_run(tmp_path):
+def test_run(tmp_path: Path, session: Session) -> None:
     """Test `esmvalcore._task.ResumeTask.run`."""
     task_name = "diagnostic_name/var_name"
     prev_output_dir = tmp_path / "recipe_test_20210911_102100"
@@ -28,7 +37,7 @@ def test_run(tmp_path):
         task_name,
     )
 
-    result = task.run()
+    result = task.run(session)
 
     metadata_file = preproc_dir / "metadata.yml"