Repository: AgnostiqHQ/covalent
Pull request #1690 (2023-06-07 03:02:52)

Memory improvements (3/3): Introduce new runner and executor API
<!--
⚠️ Your pull request title should be short detailed and understandable for all.
⚠️ Increment the version number in the /VERSION file. If this is a bugfix, increment the patch version (the rightmost number), for example 0.18.2 becomes 0.18.3. If it's a new feature, increment the minor version (the middle number), for example 0.18.2 becomes 0.19.0.
⚠️ Add a note to /CHANGELOG.md with your version number and the date, summarizing the changes.
⚠️ If your pull request fixes an open issue, please link to the issue.
⚠️ Rebase the latest changes from the develop branch. Assuming origin points to https://github.com/AgnostiqHQ/covalent, you should run git rebase origin/develop.
-->

- [ ] I have added the tests to cover my changes.
- [ ] I have updated the documentation and CHANGELOG accordingly.
- [ ] I have read the CONTRIBUTING document.

Closes https://github.com/AgnostiqHQ/covalent-staging/issues/709


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1690?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) Report
> Merging [#1690](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1690?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (8b65f60) into [develop](https://app.codecov.io/gh/AgnostiqHQ/covalent/commit/b6bb6b938ca1cda50b9a2549e1e2bad45606ad53?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (b6bb6b9) will **decrease** coverage by `4.12%`.
> The diff coverage is `80.67%`.

> :exclamation: Current head 8b65f60 differs from pull request most recent head 21952ab. Consider uploading reports for the commit 21952ab to get more accurate results

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1690/graphs/tree.svg?width=650&height=150&src=pr&token=YGHCB3DE4P&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1690?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)

```diff
@@             Coverage Diff             @@
##           develop    #1690      +/-   ##
===========================================
- Coverage    87.79%   83.67%   -4.12%     
===========================================
  Files          159      216      +57     
  Lines         6801    10636    +3835     
  Branches       151      151              
===========================================
+ Hits          5971     8900    +2929     
- Misses         725     1631     +906     
  Partials       105      105              
```

| Flag | Coverage Δ | |
|---|---|---|
| Dispatcher | `83.55% <83.31%> (-4.03%)` | :arrow_down: |
| Functional_Tests | `54.25% <51.17%> (-4.27%)` | :arrow_down: |
| SDK | `77.84% <57.99%> (-13.10%)` | :arrow_down: |
| UI_Frontend | `72.64% <ø> (ø)` | |

</details>
kessler-frost: Converting to draft to add more tests.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ff32a7cc..5a67bd55 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -38,11 +38,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Documentation and test cases for database triggers.
 - Added the `__pow__` method to the `Electron` class
+- New Runner and executor API to bypass server-side memory when running tasks.
 
 ### Docs
 
 - Added federated learning showcase code
-- Updated tutorial for redispatching workflows with Streamlit 
+- Updated tutorial for redispatching workflows with Streamlit
 
 ### Tests
 
diff --git a/covalent/_serialize/lattice.py b/covalent/_serialize/lattice.py
index 5aefbb61..ebe405ac 100644
--- a/covalent/_serialize/lattice.py
+++ b/covalent/_serialize/lattice.py
@@ -42,8 +42,8 @@ ASSET_TYPES = {
     "inputs": AssetType.TRANSPORTABLE,
     "named_args": AssetType.TRANSPORTABLE,
     "named_kwargs": AssetType.TRANSPORTABLE,
-    "cova_imports": AssetType.OBJECT,
-    "lattice_imports": AssetType.OBJECT,
+    "cova_imports": AssetType.JSONABLE,
+    "lattice_imports": AssetType.TEXT,
     "deps": AssetType.JSONABLE,
     "call_before": AssetType.JSONABLE,
     "call_after": AssetType.JSONABLE,
diff --git a/covalent/_shared_files/schemas/lattice.py b/covalent/_shared_files/schemas/lattice.py
index f3c2a352..ffc38654 100644
--- a/covalent/_shared_files/schemas/lattice.py
+++ b/covalent/_shared_files/schemas/lattice.py
@@ -62,8 +62,8 @@ LATTICE_RESULTS_FILENAME = "results.tobj"
 LATTICE_DEPS_FILENAME = "deps.json"
 LATTICE_CALL_BEFORE_FILENAME = "call_before.json"
 LATTICE_CALL_AFTER_FILENAME = "call_after.json"
-LATTICE_COVA_IMPORTS_FILENAME = "cova_imports.pkl"
-LATTICE_LATTICE_IMPORTS_FILENAME = "lattice_imports.pkl"
+LATTICE_COVA_IMPORTS_FILENAME = "cova_imports.json"
+LATTICE_LATTICE_IMPORTS_FILENAME = "lattice_imports.txt"
 LATTICE_STORAGE_TYPE = "file"
 
 
diff --git a/covalent/_shared_files/utils.py b/covalent/_shared_files/utils.py
index 48e26870..e7bd6036 100644
--- a/covalent/_shared_files/utils.py
+++ b/covalent/_shared_files/utils.py
@@ -21,7 +21,7 @@ import inspect
 import shutil
 import socket
 from datetime import timedelta
-from typing import Any, Callable, Dict, Set, Tuple
+from typing import Any, Callable, Dict, List, Tuple
 
 import cloudpickle
 from pennylane._device import Device
@@ -37,9 +37,6 @@ DEFAULT_UI_ADDRESS = get_config("user_interface.address")
 DEFAULT_UI_PORT = get_config("user_interface.port")
 
 
-# Dictionary to map Dask clients to their scheduler addresses
-_address_client_mapper = {}
-
 _IMPORT_PATH_SEPARATOR = ":"
 
 
@@ -141,7 +138,7 @@ def get_serialized_function_str(function):
     return function_str + "\n\n"
 
 
-def get_imports(func: Callable) -> Tuple[str, Set[str]]:
+def get_imports(func: Callable) -> Tuple[str, List[str]]:
     """
     Given an input workflow function, find the imports that were used, and determine
         which ones are Covalent-related.
@@ -155,7 +152,7 @@ def get_imports(func: Callable) -> Tuple[str, Set[str]]:
     """
 
     imports_str = ""
-    cova_imports = set()
+    cova_imports = []
     for i, j in func.__globals__.items():
         if inspect.ismodule(j) or (
             inspect.isfunction(j) and j.__name__ in ["lattice", "electron"]
@@ -167,7 +164,7 @@ def get_imports(func: Callable) -> Tuple[str, Set[str]]:
 
             if j.__name__ in ["covalent", "lattice", "electron"]:
                 import_line = f"# {import_line}"
-                cova_imports.add(i)
+                cova_imports.append(i)
 
             imports_str += import_line
 
diff --git a/covalent/_workflow/lattice.py b/covalent/_workflow/lattice.py
index 3339ffa2..aa09ac03 100644
--- a/covalent/_workflow/lattice.py
+++ b/covalent/_workflow/lattice.py
@@ -82,7 +82,6 @@ class Lattice:
         self.named_kwargs = None
         self.electron_outputs = {}
         self.lattice_imports, self.cova_imports = get_imports(self.workflow_function)
-        self.cova_imports.update({"electron"})
 
         self.workflow_function = TransportableObject.make_transportable(self.workflow_function)
 
@@ -110,13 +109,11 @@ class Lattice:
         for node_name, output in self.electron_outputs.items():
             attributes["electron_outputs"][node_name] = output.to_dict()
 
-        attributes["cova_imports"] = list(self.cova_imports)
         return json.dumps(attributes)
 
     @staticmethod
     def deserialize_from_json(json_data: str) -> None:
         attributes = json.loads(json_data)
-        attributes["cova_imports"] = set(attributes["cova_imports"])
 
         for node_name, object_dict in attributes["electron_outputs"].items():
             attributes["electron_outputs"][node_name] = TransportableObject.from_dict(object_dict)
@@ -211,6 +208,7 @@ class Lattice:
         self.inputs = TransportableObject({"args": args, "kwargs": kwargs})
         self.named_args = TransportableObject(named_args)
         self.named_kwargs = TransportableObject(named_kwargs)
+        self.lattice_imports, self.cova_imports = get_imports(workflow_function)
 
         # Set any lattice metadata not explicitly set by the user
         constraint_names = {"executor", "workflow_executor", "deps", "call_before", "call_after"}
diff --git a/covalent/executor/__init__.py b/covalent/executor/__init__.py
index afde6717..ffc139f9 100644
--- a/covalent/executor/__init__.py
+++ b/covalent/executor/__init__.py
@@ -32,7 +32,7 @@ import pkg_resources
 from .._shared_files import logger
 from .._shared_files.config import get_config, update_config
 from ..quantum import QCluster, Simulator
-from .base import BaseExecutor, wrapper_fn
+from .base import BaseExecutor
 
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
diff --git a/covalent/executor/base.py b/covalent/executor/base.py
index 23ddfb34..562af7b2 100644
--- a/covalent/executor/base.py
+++ b/covalent/executor/base.py
@@ -27,93 +27,25 @@ import queue
 from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from typing import (
-    Any,
-    Callable,
-    ContextManager,
-    Dict,
-    Iterable,
-    List,
-    Literal,
-    Optional,
-    Tuple,
-    Union,
-)
+from typing import Any, Callable, ContextManager, Dict, Iterable, List, Optional, Union
 
 import aiofiles
 
+from covalent._shared_files.exceptions import TaskCancelledError
+from covalent.executor.schemas import ResourceMap, TaskSpec
+from covalent.executor.utils import Signals
+
 from .._shared_files import TaskRuntimeError, logger
 from .._shared_files.context_managers import active_dispatch_info_manager
-from .._shared_files.exceptions import TaskCancelledError
 from .._shared_files.qelectron_utils import remove_qelectron_db
-from .._shared_files.util_classes import RESULT_STATUS, DispatchInfo
-from .._workflow.depscall import RESERVED_RETVAL_KEY__FILES
-from .._workflow.transport import TransportableObject
-from .utils import Signals
+from .._shared_files.util_classes import RESULT_STATUS, DispatchInfo, Status
+from .schemas import TaskUpdate
 
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
 TypeJSON = Union[str, int, float, bool, None, Dict[str, Any], List[Any]]
 
 
-def wrapper_fn(
-    function: TransportableObject,
-    call_before: List[Tuple[TransportableObject, TransportableObject, TransportableObject]],
-    call_after: List[Tuple[TransportableObject, TransportableObject, TransportableObject]],
-    *args,
-    **kwargs,
-):
-    """Wrapper for serialized callable.
-
-    Execute preparatory shell commands before deserializing and
-    running the callable. This is the actual function to be sent to
-    the various executors.
-
-    """
-
-    cb_retvals = {}
-    for tup in call_before:
-        serialized_fn, serialized_args, serialized_kwargs, retval_key = tup
-        cb_fn = serialized_fn.get_deserialized()
-        cb_args = serialized_args.get_deserialized()
-        cb_kwargs = serialized_kwargs.get_deserialized()
-        retval = cb_fn(*cb_args, **cb_kwargs)
-
-        # we always store cb_kwargs dict values as arrays to factor in non-unique values
-        if retval_key and retval_key in cb_retvals:
-            cb_retvals[retval_key].append(retval)
-        elif retval_key:
-            cb_retvals[retval_key] = [retval]
-
-    # if cb_retvals key only contains one item this means it is a unique (non-repeated) retval key
-    # so we only return the first element however if it is a 'files' kwarg we always return as a list
-    cb_retvals = {
-        key: value[0] if len(value) == 1 and key != RESERVED_RETVAL_KEY__FILES else value
-        for key, value in cb_retvals.items()
-    }
-
-    fn = function.get_deserialized()
-
-    new_args = [arg.get_deserialized() for arg in args]
-
-    new_kwargs = {k: v.get_deserialized() for k, v in kwargs.items()}
-
-    # Inject return values into kwargs
-    for key, val in cb_retvals.items():
-        new_kwargs[key] = val
-
-    output = fn(*new_args, **new_kwargs)
-
-    for tup in call_after:
-        serialized_fn, serialized_args, serialized_kwargs, retval_key = tup
-        ca_fn = serialized_fn.get_deserialized()
-        ca_args = serialized_args.get_deserialized()
-        ca_kwargs = serialized_kwargs.get_deserialized()
-        ca_fn(*ca_args, **ca_kwargs)
-
-    return TransportableObject(output)
-
-
 class _AbstractBaseExecutor(ABC):
     """
     Private parent class for BaseExecutor and AsyncBaseExecutor
@@ -127,6 +59,8 @@ class _AbstractBaseExecutor(ABC):
 
     """
 
+    SUPPORTS_MANAGED_EXECUTION = False
+
     def __init__(
         self,
         log_stdout: str = "",
@@ -285,6 +219,18 @@ class BaseExecutor(_AbstractBaseExecutor):
         """
         return self._notify_sync(Signals.GET, "cancel_requested")
 
+    def get_version_info(self) -> Dict:
+        """
+        Query the database for the task's Python and Covalent version
+
+        Arg:
+            dispatch_id: Dispatch ID of the lattice
+
+        Returns:
+            {"python": python_version, "covalent": covalent_version}
+        """
+        return self._notify_sync(Signals.GET, "version_info")
+
     def set_job_handle(self, handle: TypeJSON) -> Any:
         """
         Save the job_id/handle returned by the backend executing the task
@@ -297,6 +243,32 @@ class BaseExecutor(_AbstractBaseExecutor):
         """
         return self._notify_sync(Signals.PUT, ("job_handle", json.dumps(handle)))
 
+    def _set_job_status_sync(self, status: Status) -> bool:
+        if self.validate_status(status):
+            return self._notify_sync(Signals.PUT, ("job_status", str(status)))
+        else:
+            return False
+
+    def validate_status(self, status: Status) -> bool:
+        """Overridable filter"""
+        return True
+
+    async def set_job_status(self, status: Status) -> bool:
+        """
+        Sets the job state
+
+        For use with send/receive API
+
+        Return(s)
+            Whether the action succeeded
+        """
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(
+            None,
+            self._set_job_status_sync,
+            status,
+        )
+
     def write_streams_to_file(
         self,
         stream_strings: Iterable[str],
@@ -432,7 +404,7 @@ class BaseExecutor(_AbstractBaseExecutor):
 
         raise NotImplementedError
 
-    def cancel(self, task_metadata: Dict, job_handle: Any) -> Literal[False]:
+    def cancel(self, task_metadata: Dict, job_handle: Any) -> bool:
         """
         Method to cancel the job identified uniquely by the `job_handle` (base class)
 
@@ -446,7 +418,7 @@ class BaseExecutor(_AbstractBaseExecutor):
         app_log.debug(f"Cancel not implemented for executor {type(self)}")
         return False
 
-    async def _cancel(self, task_metadata: Dict, job_handle: Any) -> Any:
+    async def _cancel(self, task_metadata: Dict, job_handle: Any) -> bool:
         """
         Cancel the task in a non-blocking manner
 
@@ -467,6 +439,77 @@ class BaseExecutor(_AbstractBaseExecutor):
         """Placeholder to run any executor specific cleanup/teardown actions"""
         pass
 
+    async def send(
+        self,
+        task_specs: List[Dict],
+        resources: ResourceMap,
+        task_group_metadata: Dict,
+    ):
+        """Submit a list of task references to the compute backend.
+
+        Args:
+            task_specs: a list of TaskSpecs
+            resources: a ResourceMap mapping task assets to URIs
+            task_group_metadata: a dictionary of metadata for the task group.
+                                 Current keys are `dispatch_id`, `node_ids`,
+                                 and `task_group_id`.
+
+        The return value of `send()` will be passed directly into `poll()`.
+        """
+        # Schemas:
+        #
+        # Task spec:
+        # {
+        #     "function_id": int,
+        #     "args_ids": List[int],
+        #     "kwargs_ids": Dict[str, int],
+        #     "deps_id": str,
+        #     "call_before_id": str,
+        #     "call_after_id": str,
+        # }
+
+        # resources:
+        # {
+        #     "functions": Dict[int, str],
+        #     "inputs": Dict[int, str],
+        #     "deps": Dict[str, str]
+        # }
+
+        # task_group_metadata:
+        # {
+        #     "dispatch_id": str,
+        #     "node_ids": List[int],
+        #     "task_group_id": int,
+        # }
+
+        # Assets are will be accessible by the compute backend
+        # at the provided URIs
+
+        # Covalent will upload all assets before invoking `send()`.
+
+        raise NotImplementedError
+
+    async def poll(self, task_group_metadata: Dict, data: Any):
+        # To be run as a background task.  A callback will be
+        # registered with the runner to invoke the receive()
+
+        raise NotImplementedError
+
+    async def receive(
+        self,
+        task_group_metadata: Dict,
+        data: Any,
+    ) -> List[TaskUpdate]:
+        # Returns (output_uri, stdout_uri, stderr_uri,
+        # exception_raised)
+
+        # Job should have reached a terminal state by the time this is invoked.
+
+        raise NotImplementedError
+
+    def get_upload_uri(self, task_metadata: Dict, object_key: str):
+        return ""
+
 
 class AsyncBaseExecutor(_AbstractBaseExecutor):
     """Async base executor class to be used for defining any executor
@@ -566,6 +609,18 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
         """
         return await self._notify_sync(Signals.GET, "cancel_requested")
 
+    async def get_version_info(self) -> Dict:
+        """
+        Query the database for dispatch version metadata.
+
+        Arg:
+            dispatch_id: Dispatch ID of the lattice
+
+        Returns:
+            {"python": python_version, "covalent": covalent_version}
+        """
+        return await self._notify_sync(Signals.GET, "version_info")
+
     async def set_job_handle(self, handle: TypeJSON) -> Any:
         """
         Save the job handle to database
@@ -578,6 +633,24 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
         """
         return await self._notify_sync(Signals.PUT, ("job_handle", json.dumps(handle)))
 
+    def validate_status(self, status: Status) -> bool:
+        """Overridable filter"""
+        return True
+
+    async def set_job_status(self, status: Status) -> bool:
+        """
+        Validates and sets the job state
+
+        For use with send/receive API
+
+        Return(s)
+            Whether the action succeeded
+        """
+        if self.validate_status(status):
+            return await self._notify_sync(Signals.PUT, ("job_status", str(status)))
+        else:
+            return False
+
     async def write_streams_to_file(
         self,
         stream_strings: Iterable[str],
@@ -695,13 +768,13 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
 
         raise NotImplementedError
 
-    async def cancel(self, task_metadata: Dict, job_handle: Any) -> Literal[False]:
+    async def cancel(self, task_metadata: Dict, job_handle: Any) -> bool:
         """
-        Executor specific task cancellation method
+        Method to cancel the job identified uniquely by the `job_handle` (base class)
 
         Arg(s)
-            task_metadata: Metadata associated with the task to be cancelled
-            job_handle: Unique ID assigned to the job by the backend
+            task_metadata: Metadata of the task to be cancelled
+            job_handle: Unique ID of the job assigned by the backend
 
         Return(s)
             False by default
@@ -709,17 +782,132 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
         app_log.debug(f"Cancel not implemented for executor {type(self)}")
         return False
 
-    async def _cancel(self, task_metadata: Dict, job_handle: Any) -> Literal[False]:
+    async def _cancel(self, task_metadata: Dict, job_handle: Any) -> bool:
         """
-        Cancel the task in a non-blocking manner and teardown the infrastructure
+        Cancel the task in a non-blocking manner
 
         Arg(s)
-            task_metadata: Metadata associated with the task to be cancelled
-            job_handle: Unique ID assigned to the job by the backend
+            task_metadata: Metadata of the task to be cancelled
+            job_handle: Unique ID of the job assigned by the backend
 
         Return(s)
-            Result from cancelling the task
+           Result of the task cancellation
         """
-        cancel_result = await self.cancel(task_metadata, job_handle)
-        await self.teardown(task_metadata)
-        return cancel_result
+        return await self.cancel(task_metadata, job_handle)
+
+    async def send(
+        self,
+        task_specs: List[TaskSpec],
+        resources: ResourceMap,
+        task_group_metadata: Dict,
+    ) -> Any:
+        """Submit a list of task references to the compute backend.
+
+        Args:
+            task_specs: a list of TaskSpecs
+            resources: a ResourceMap mapping task assets to URIs
+            task_group_metadata: A dictionary of metadata for the task group.
+                                 Current keys are `dispatch_id`, `node_ids`,
+                                 and `task_group_id`.
+
+        The return value of `send()` will be passed directly into `poll()`.
+        """
+        # Schemas:
+        #
+        # Task spec:
+        # {
+        #     "function_id": int,
+        #     "args_ids": List[int],
+        #     "kwargs_ids": Dict[str, int],
+        #     "deps_id": str,
+        #     "call_before_id": str,
+        #     "call_after_id": str,
+        # }
+
+        # resources:
+        # {
+        #     "functions": Dict[int, str],
+        #     "inputs": Dict[int, str],
+        #     "deps": Dict[str, str]
+        # }
+
+        # task_group_metadata:
+        # {
+        #     "dispatch_id": str,
+        #     "node_ids": List[int],
+        #     "task_group_id": int,
+        # }
+
+        # Assets are assumed to be accessible by the compute backend
+        # at the provided URIs
+
+        # Covalent will upload all assets before invoking send().
+
+        raise NotImplementedError
+
+    async def poll(self, task_group_metadata: Dict, data: Any) -> Any:
+        """Block until the job has reached a terminal state.
+
+        Args:
+            task_group_metadata: A dictionary of metadata for the task group.
+                                 Current keys are `dispatch_id`, `node_ids`,
+                                 and `task_group_id`.
+            data: The return value of send().
+
+        The return value of `poll()` will be passed directly into receive().
+
+        Raise `NotImplementedError` to indicate that the compute backend
+        will notify the Covalent server asynchronously of job completion.
+
+        """
+
+        raise NotImplementedError
+
+    async def receive(
+        self,
+        task_group_metadata: Dict,
+        data: Any,
+    ) -> List[TaskUpdate]:
+        """Return a list of task updates.
+
+        Each task must have reached a terminal state by the time this is invoked.
+
+        Args:
+            task_group_metadata: A dictionary of metadata for the task group.
+                                 Current keys are `dispatch_id`, `node_ids`,
+                                 and `task_group_id`.
+            data: The return value of poll() or the request body of `/jobs/update`.
+
+        Returns:
+            Returns a list of task results, each a TaskUpdate dataclass
+            of the form
+
+            {
+              "dispatch_id": dispatch_id,
+              "node_id": node_id,
+              "status": status,
+              "assets": {
+                  "output":  {
+                    "remote_uri": output_uri,
+                  },
+                  "stdout":  {
+                    "remote_uri": stdout_uri,
+                  },
+                  "stderr":  {
+                    "remote_uri": stderr_uri,
+                  },
+                },
+            }
+
+            corresponding to the node ids (task_ids) specified in the
+            `task_group_metadata`.  This might be a subset of the node
+            ids in the originally submitted task group as jobs may
+            notify Covalent asynchronously of completed tasks before
+            the entire task group finishes running.
+
+        """
+
+        raise NotImplementedError
+
+    def get_upload_uri(self, task_group_metadata: Dict, object_key: str):
+        return ""
diff --git a/covalent/executor/executor_plugins/dask.py b/covalent/executor/executor_plugins/dask.py
index 2344d4c2..d23abc5e 100644
--- a/covalent/executor/executor_plugins/dask.py
+++ b/covalent/executor/executor_plugins/dask.py
@@ -21,18 +21,26 @@ and waits for execution to finish then returns the result.
 This is a plugin executor module; it is loaded if found and properly structured.
 """
 
+import asyncio
+import json
 import os
-from typing import Callable, Dict, List, Literal, Optional
+from enum import Enum
+from typing import Any, Callable, Dict, List, Literal, Optional
 
 from dask.distributed import CancelledError, Client, Future
+from pydantic import BaseModel
 
 from covalent._shared_files import TaskRuntimeError, logger
 
 # Relative imports are not allowed in executor plugins
 from covalent._shared_files.config import get_config
 from covalent._shared_files.exceptions import TaskCancelledError
+from covalent._shared_files.util_classes import RESULT_STATUS, Status
+from covalent._shared_files.utils import format_server_url
 from covalent.executor.base import AsyncBaseExecutor
+from covalent.executor.schemas import ResourceMap, TaskSpec, TaskUpdate
 from covalent.executor.utils.wrappers import io_wrapper as dask_wrapper
+from covalent.executor.utils.wrappers import run_task_from_uris_alt
 
 # The plugin class name must be given by the executor_plugin_name attribute:
 EXECUTOR_PLUGIN_NAME = "DaskExecutor"
@@ -54,15 +62,38 @@ _EXECUTOR_PLUGIN_DEFAULTS = {
     "create_unique_workdir": False,
 }
 
-# Temporary
-_address_client_mapper = {}
+# See https://github.com/dask/distributed/issues/5667
+_clients = {}
+
+# See
+# https://stackoverflow.com/questions/62164283/why-do-my-dask-futures-get-stuck-in-pending-and-never-finish
+_futures = {}
+
+MANAGED_EXECUTION = os.environ.get("COVALENT_USE_OLD_DASK") != "1"
+
+# Dictionary to map Dask clients to their scheduler addresses
+_address_client_map = {}
+
+
+# Valid terminal statuses
+class StatusEnum(str, Enum):
+    CANCELLED = str(RESULT_STATUS.CANCELLED)
+    COMPLETED = str(RESULT_STATUS.COMPLETED)
+    FAILED = str(RESULT_STATUS.FAILED)
+    READY = "READY"
+
+
+class ReceiveModel(BaseModel):
+    status: StatusEnum
 
 
 class DaskExecutor(AsyncBaseExecutor):
     """
-    Dask executor class that submits the input function to a running dask cluster.
+    Dask executor class that submits the input function to a running LOCAL dask cluster.
     """
 
+    SUPPORTS_MANAGED_EXECUTION = MANAGED_EXECUTION
+
     def __init__(
         self,
         scheduler_address: str = "",
@@ -111,7 +142,7 @@ class DaskExecutor(AsyncBaseExecutor):
         if not self.scheduler_address:
             try:
                 self.scheduler_address = get_config("dask.scheduler_address")
-            except KeyError as ex:
+            except KeyError:
                 app_log.debug(
                     "No dask scheduler address found in config. Address must be set manually."
                 )
@@ -123,11 +154,11 @@ class DaskExecutor(AsyncBaseExecutor):
         dispatch_id = task_metadata["dispatch_id"]
         node_id = task_metadata["node_id"]
 
-        dask_client = _address_client_mapper.get(self.scheduler_address)
+        dask_client = _address_client_map.get(self.scheduler_address)
 
         if not dask_client:
             dask_client = Client(address=self.scheduler_address, asynchronous=True)
-            _address_client_mapper[self.scheduler_address] = dask_client
+            _address_client_map[self.scheduler_address] = dask_client
             await dask_client
 
         if self.create_unique_workdir:
@@ -165,14 +196,156 @@ class DaskExecutor(AsyncBaseExecutor):
         Return(s)
             True by default
         """
-        dask_client = _address_client_mapper.get(self.scheduler_address)
+
+        if not self.scheduler_address:
+            try:
+                self.scheduler_address = get_config("dask.scheduler_address")
+            except KeyError:
+                app_log.debug(
+                    "No dask scheduler address found in config. Address must be set manually."
+                )
+
+        dask_client = _address_client_map.get(self.scheduler_address)
 
         if not dask_client:
             dask_client = Client(address=self.scheduler_address, asynchronous=True)
-            _address_client_mapper[self.scheduler_address] = dask_client
-            await dask_client
+            await asyncio.wait_for(dask_client, timeout=5)
 
         fut: Future = Future(key=job_handle, client=dask_client)
-        await fut.cancel()
+
+        # https://stackoverflow.com/questions/46278692/dask-distributed-how-to-cancel-tasks-submitted-with-fire-and-forget
+        await dask_client.cancel([fut], asynchronous=True, force=True)
         app_log.debug(f"Cancelled future with key {job_handle}")
         return True
+
+    async def send(
+        self,
+        task_specs: List[TaskSpec],
+        resources: ResourceMap,
+        task_group_metadata: dict,
+    ):
+        # Assets are assumed to be accessible by the compute backend
+        # at the provided URIs
+
+        # The Asset Manager is responsible for uploading all assets
+        # Returns a job handle (should be JSONable)
+
+        if not self.scheduler_address:
+            try:
+                self.scheduler_address = get_config("dask.scheduler_address")
+            except KeyError:
+                app_log.debug(
+                    "No dask scheduler address found in config. Address must be set manually."
+                )
+
+        dask_client = Client(address=self.scheduler_address, asynchronous=True)
+        await asyncio.wait_for(dask_client, timeout=5)
+
+        dispatch_id = task_group_metadata["dispatch_id"]
+        task_ids = task_group_metadata["node_ids"]
+        gid = task_group_metadata["task_group_id"]
+        output_uris = []
+        for node_id in task_ids:
+            result_uri = os.path.join(self.cache_dir, f"result_{dispatch_id}-{node_id}.pkl")
+            stdout_uri = os.path.join(self.cache_dir, f"stdout_{dispatch_id}-{node_id}.txt")
+            stderr_uri = os.path.join(self.cache_dir, f"stderr_{dispatch_id}-{node_id}.txt")
+            output_uris.append((result_uri, stdout_uri, stderr_uri))
+
+        server_url = format_server_url()
+
+        key = f"dask_job_{dispatch_id}:{gid}"
+
+        await self.set_job_handle(key)
+
+        future = dask_client.submit(
+            run_task_from_uris_alt,
+            list(map(lambda t: t.dict(), task_specs)),
+            resources.dict(),
+            output_uris,
+            self.cache_dir,
+            task_group_metadata,
+            server_url,
+            key=key,
+        )
+
+        _clients[key] = dask_client
+        _futures[key] = future
+
+        return future.key
+
+    async def poll(self, task_group_metadata: Dict, poll_data: Any):
+        fut = _futures.pop(poll_data)
+        app_log.debug(f"Future {fut}")
+        try:
+            await fut
+        except CancelledError as e:
+            raise TaskCancelledError() from e
+
+        _clients.pop(poll_data)
+
+        return {"status": StatusEnum.READY.value}
+
+    async def receive(self, task_group_metadata: Dict, data: Any) -> List[TaskUpdate]:
+        # Job should have reached a terminal state by the time this is invoked.
+        dispatch_id = task_group_metadata["dispatch_id"]
+        task_ids = task_group_metadata["node_ids"]
+
+        task_results = []
+
+        if not data:
+            terminal_status = RESULT_STATUS.CANCELLED
+        else:
+            received = ReceiveModel.parse_obj(data)
+            terminal_status = Status(received.status.value)
+
+        for task_id in task_ids:
+            # TODO: Handle the case where the job was cancelled before the task started running
+            app_log.debug(f"Receive called for task {dispatch_id}:{task_id} with data {data}")
+
+            if terminal_status == RESULT_STATUS.CANCELLED:
+                output_uri = ""
+                stdout_uri = ""
+                stderr_uri = ""
+
+            else:
+                result_path = os.path.join(self.cache_dir, f"result-{dispatch_id}:{task_id}.json")
+                with open(result_path, "r") as f:
+                    result_summary = json.load(f)
+                    node_id = result_summary["node_id"]
+                    output_uri = result_summary["output_uri"]
+                    stdout_uri = result_summary["stdout_uri"]
+                    stderr_uri = result_summary["stderr_uri"]
+                    exception_raised = result_summary["exception_occurred"]
+
+                terminal_status = (
+                    RESULT_STATUS.FAILED if exception_raised else RESULT_STATUS.COMPLETED
+                )
+
+            task_result = {
+                "dispatch_id": dispatch_id,
+                "node_id": task_id,
+                "status": terminal_status,
+                "assets": {
+                    "output": {
+                        "remote_uri": output_uri,
+                    },
+                    "stdout": {
+                        "remote_uri": stdout_uri,
+                    },
+                    "stderr": {
+                        "remote_uri": stderr_uri,
+                    },
+                },
+            }
+
+            task_results.append(TaskUpdate(**task_result))
+
+        app_log.debug(f"Returning results for tasks {dispatch_id}:{task_ids}")
+        return task_results
+
+    def get_upload_uri(self, task_group_metadata: Dict, object_key: str):
+        dispatch_id = task_group_metadata["dispatch_id"]
+        task_group_id = task_group_metadata["task_group_id"]
+
+        filename = f"asset_{dispatch_id}-{task_group_id}_{object_key}.pkl"
+        return os.path.join("file://", self.cache_dir, filename)
diff --git a/covalent/executor/executor_plugins/local.py b/covalent/executor/executor_plugins/local.py
index 3352a48f..aaba7a04 100644
--- a/covalent/executor/executor_plugins/local.py
+++ b/covalent/executor/executor_plugins/local.py
@@ -20,19 +20,27 @@ Module for defining a local executor that directly invokes the input python func
 This is a plugin executor module; it is loaded if found and properly structured.
 """
 
-
+import asyncio
 import os
 from concurrent.futures import ProcessPoolExecutor
+from enum import Enum
 from typing import Any, Callable, Dict, List, Optional
 
-# Relative imports are not allowed in executor plugins
+import requests
+from pydantic import BaseModel
+
 from covalent._shared_files import TaskCancelledError, TaskRuntimeError, logger
 from covalent._shared_files.config import get_config
+from covalent._shared_files.util_classes import RESULT_STATUS, Status
+from covalent._shared_files.utils import format_server_url
 from covalent.executor import BaseExecutor
 
+# Relative imports are not allowed in executor plugins
+from covalent.executor.schemas import ResourceMap, TaskSpec, TaskUpdate
+
 # Store the wrapper function in an external module to avoid module
 # import errors during pickling
-from covalent.executor.utils.wrappers import io_wrapper
+from covalent.executor.utils.wrappers import io_wrapper, run_task_from_uris
 
 # The plugin class name must be given by the executor_plugin_name attribute:
 EXECUTOR_PLUGIN_NAME = "LocalExecutor"
@@ -58,11 +66,27 @@ _EXECUTOR_PLUGIN_DEFAULTS = {
 proc_pool = ProcessPoolExecutor()
 
 
+# Valid terminal statuses
+class StatusEnum(str, Enum):
+    CANCELLED = str(RESULT_STATUS.CANCELLED)
+    COMPLETED = str(RESULT_STATUS.COMPLETED)
+    FAILED = str(RESULT_STATUS.FAILED)
+
+
+class ReceiveModel(BaseModel):
+    status: StatusEnum
+
+
+MANAGED_EXECUTION = True
+
+
 class LocalExecutor(BaseExecutor):
     """
     Local executor class that directly invokes the input function.
     """
 
+    SUPPORTS_MANAGED_EXECUTION = MANAGED_EXECUTION
+
     def __init__(
         self, workdir: str = "", create_unique_workdir: Optional[bool] = None, *args, **kwargs
     ) -> None:
@@ -130,3 +154,118 @@ class LocalExecutor(BaseExecutor):
             raise TaskRuntimeError(tb)
 
         return output
+
+    def _send(
+        self,
+        task_specs: List[TaskSpec],
+        resources: ResourceMap,
+        task_group_metadata: dict,
+    ):
+        dispatch_id = task_group_metadata["dispatch_id"]
+        task_ids = task_group_metadata["node_ids"]
+        gid = task_group_metadata["task_group_id"]
+        output_uris = []
+        for node_id in task_ids:
+            result_uri = os.path.join(self.cache_dir, f"result_{dispatch_id}-{node_id}.pkl")
+            stdout_uri = os.path.join(self.cache_dir, f"stdout_{dispatch_id}-{node_id}.txt")
+            stderr_uri = os.path.join(self.cache_dir, f"stderr_{dispatch_id}-{node_id}.txt")
+            output_uris.append((result_uri, stdout_uri, stderr_uri))
+
+        server_url = format_server_url()
+
+        app_log.debug(f"Running task group {dispatch_id}:{task_ids}")
+        future = proc_pool.submit(
+            run_task_from_uris,
+            list(map(lambda t: t.dict(), task_specs)),
+            resources.dict(),
+            output_uris,
+            self.cache_dir,
+            task_group_metadata,
+            server_url,
+        )
+
+        def handle_cancelled(fut):
+            app_log.debug(f"In done callback for {dispatch_id}:{gid}, future {fut}")
+            if fut.cancelled():
+                for task_id in task_ids:
+                    url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/job"
+                    requests.put(url, json={"status": "CANCELLED"})
+
+        future.add_done_callback(handle_cancelled)
+
+    def _receive(self, task_group_metadata: Dict, data: Any) -> List[TaskUpdate]:
+        # Returns (output_uri, stdout_uri, stderr_uri,
+        # exception_raised)
+
+        # Job should have reached a terminal state by the time this is invoked.
+        dispatch_id = task_group_metadata["dispatch_id"]
+        task_ids = task_group_metadata["node_ids"]
+
+        task_results = []
+
+        for task_id in task_ids:
+            # Handle the case where the job was cancelled before the task started running
+            app_log.debug(f"Receive called for task {dispatch_id}:{task_id} with data {data}")
+
+            if not data:
+                terminal_status = RESULT_STATUS.CANCELLED
+            else:
+                received = ReceiveModel.parse_obj(data)
+                terminal_status = Status(received.status.value)
+
+            task_result = {
+                "dispatch_id": dispatch_id,
+                "node_id": task_id,
+                "status": terminal_status,
+                "assets": {
+                    "output": {
+                        "remote_uri": "",
+                    },
+                    "stdout": {
+                        "remote_uri": "",
+                    },
+                    "stderr": {
+                        "remote_uri": "",
+                    },
+                },
+            }
+
+            task_results.append(TaskUpdate(**task_result))
+
+        app_log.debug(f"Returning results for tasks {dispatch_id}:{task_ids}")
+        return task_results
+
+    async def send(
+        self,
+        task_specs: List[TaskSpec],
+        resources: ResourceMap,
+        task_group_metadata: dict,
+    ):
+        # Assets are assumed to be accessible by the compute backend
+        # at the provided URIs
+
+        # The Asset Manager is responsible for uploading all assets
+        # Returns a job handle (should be JSONable)
+
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(
+            None,
+            self._send,
+            task_specs,
+            resources,
+            task_group_metadata,
+        )
+
+    async def receive(self, task_group_metadata: Dict, data: Any) -> List[TaskUpdate]:
+        # Returns (output_uri, stdout_uri, stderr_uri,
+        # exception_raised)
+
+        # Job should have reached a terminal state by the time this is invoked.
+        loop = asyncio.get_running_loop()
+
+        return await loop.run_in_executor(
+            None,
+            self._receive,
+            task_group_metadata,
+            data,
+        )
diff --git a/covalent/executor/schemas.py b/covalent/executor/schemas.py
new file mode 100644
index 00000000..ef99c9c7
--- /dev/null
+++ b/covalent/executor/schemas.py
@@ -0,0 +1,118 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Types defining the runner-executor interface
+"""
+
+from typing import Dict, List
+
+from pydantic import BaseModel, validator
+
+from covalent._shared_files.schemas.asset import AssetUpdate
+from covalent._shared_files.util_classes import RESULT_STATUS, Status
+
+
+class TaskUpdate(BaseModel):
+    """Represents a task status update.
+
+    Attributes:
+        dispatch_id: The id of the dispatch.
+        node_id: The id of the task.
+        status: A Status dataclass representing the task's terminal status.
+        assets: A map from asset keys to AssetUpdate objects
+    """
+
+    dispatch_id: str
+    node_id: int
+    status: Status
+    assets: Dict[str, AssetUpdate]
+
+    @validator("status")
+    def validate_status(cls, v):
+        if RESULT_STATUS.is_terminal(v):
+            return v
+        else:
+            raise ValueError(f"Illegal status update {v}")
+
+
+class TaskSpec(BaseModel):
+    """An abstract description of a runnable task.
+
+    Attributes:
+        function_id: The `node_id` of the function.
+        args_ids: The `node_id`s of the function's args
+        kwargs_ids: The `node_id`s of the function's kwargs {key: node_id}
+        deps_id: An opaque string representing the task's deps.
+        call_before_id: An opaque string representing the task's call_before.
+        call_after_id: An opaque string representing the task's call_before.
+
+    The attribute values can be used in conjunction with a
+    `ResourceMap` to locate the actual resources in the compute
+    environment.
+    """
+
+    function_id: int
+    args_ids: List[int]
+    kwargs_ids: Dict[str, int]
+    deps_id: str
+    call_before_id: str
+    call_after_id: str
+
+
+class ResourceMap(BaseModel):
+
+    """Map resource identifiers to URIs.
+
+    The resources may be loaded in the compute environment from these
+    URIs.
+
+    Resource identifiers are the attribute values of TaskSpecs. For
+    instance, if ts is a `TaskSpec` and rm is the corresponding
+    `ResourceMap`, then
+    - the serialized function has URI `rm.functions[ts.function_id]`
+    - the serialized args have URIs `rm.inputs[ts.args_ids[i]]`
+    - the call_before has URI `rm.deps[ts.call_before_id]`
+
+    Attributes:
+        functions: A map from node id to the corresponding URI.
+        inputs: A map from node id to the corresponding URI
+        deps: A map from deps resource ids to their corresponding URIs.
+
+    """
+
+    # Map node_id to URI
+    functions: Dict[int, str]
+
+    # Map node_id to URI
+    inputs: Dict[int, str]
+
+    # Includes deps, call_before, call_after
+    deps: Dict[str, str]
+
+
+class TaskGroup(BaseModel):
+    """Description of a group of runnable graph nodes.
+
+    Attributes:
+        dispatch_id: The `dispatch_id`.
+        task_ids: The graph nodes comprising the task group.
+        task_group_id: The task group identifier.
+    """
+
+    dispatch_id: str
+    task_ids: List[int]
+    task_group_id: int
diff --git a/covalent/executor/utils/__init__.py b/covalent/executor/utils/__init__.py
index 7c5c5585..fb7160c1 100644
--- a/covalent/executor/utils/__init__.py
+++ b/covalent/executor/utils/__init__.py
@@ -15,4 +15,4 @@
 # limitations under the License.
 
 from .context import get_context, set_context
-from .wrappers import Signals
+from .enums import Signals
diff --git a/covalent/executor/utils/enums.py b/covalent/executor/utils/enums.py
new file mode 100644
index 00000000..8246cfd9
--- /dev/null
+++ b/covalent/executor/utils/enums.py
@@ -0,0 +1,31 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Helper functions for the local executor
+"""
+
+from enum import Enum
+
+
+class Signals(Enum):
+    """
+    Signals to enable communication between the executors and Covalent dispatcher
+    """
+
+    GET = 0
+    PUT = 1
+    EXIT = 2
diff --git a/covalent/executor/utils/serialize.py b/covalent/executor/utils/serialize.py
new file mode 100644
index 00000000..a5c683b1
--- /dev/null
+++ b/covalent/executor/utils/serialize.py
@@ -0,0 +1,33 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Functions for serializing and deserializing assets
+"""
+
+from typing import Any
+
+from ..._serialize.common import deserialize_asset, serialize_asset
+from ..._serialize.electron import ASSET_TYPES
+
+
+# Convenience functions for executor plugins
+def serialize_node_asset(data: Any, key: str) -> bytes:
+    return serialize_asset(data, ASSET_TYPES[key])
+
+
+def deserialize_node_asset(data: bytes, key: str) -> Any:
+    return deserialize_asset(data, ASSET_TYPES[key])
diff --git a/covalent/executor/utils/wrappers.py b/covalent/executor/utils/wrappers.py
index a682d615..b2199b0d 100644
--- a/covalent/executor/utils/wrappers.py
+++ b/covalent/executor/utils/wrappers.py
@@ -19,22 +19,79 @@ Helper functions for the local executor
 """
 
 import io
+import json
 import os
+import sys
 import traceback
 from contextlib import redirect_stderr, redirect_stdout
-from enum import Enum
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Tuple
 
+import requests
+
+from covalent._workflow.depsbash import DepsBash
+from covalent._workflow.depscall import RESERVED_RETVAL_KEY__FILES, DepsCall
+from covalent._workflow.depspip import DepsPip
+from covalent._workflow.transport import TransportableObject
+from covalent.executor.utils.serialize import deserialize_node_asset, serialize_node_asset
+
+
+def wrapper_fn(
+    function: TransportableObject,
+    call_before: List[Tuple[TransportableObject, TransportableObject, TransportableObject]],
+    call_after: List[Tuple[TransportableObject, TransportableObject, TransportableObject]],
+    *args,
+    **kwargs,
+):
+    """Wrapper for serialized callable.
+
+    Execute preparatory shell commands before deserializing and
+    running the callable. This is the actual function to be sent to
+    the various executors.
 
-class Signals(Enum):
-    """
-    Signals to enable communication between the executors and Covalent dispatcher
     """
 
-    GET = 0
-    PUT = 1
-    EXIT = 2
+    cb_retvals = {}
+    for tup in call_before:
+        serialized_fn, serialized_args, serialized_kwargs, retval_key = tup
+        cb_fn = serialized_fn.get_deserialized()
+        cb_args = serialized_args.get_deserialized()
+        cb_kwargs = serialized_kwargs.get_deserialized()
+        retval = cb_fn(*cb_args, **cb_kwargs)
+
+        # we always store cb_kwargs dict values as arrays to factor in non-unique values
+        if retval_key and retval_key in cb_retvals:
+            cb_retvals[retval_key].append(retval)
+        elif retval_key:
+            cb_retvals[retval_key] = [retval]
+
+    # if cb_retvals key only contains one item this means it is a unique (non-repeated) retval key
+    # so we only return the first element however if it is a 'files' kwarg we always return as a list
+    cb_retvals = {
+        key: value[0] if len(value) == 1 and key != RESERVED_RETVAL_KEY__FILES else value
+        for key, value in cb_retvals.items()
+    }
+
+    fn = function.get_deserialized()
+
+    new_args = [arg.get_deserialized() for arg in args]
+
+    new_kwargs = {k: v.get_deserialized() for k, v in kwargs.items()}
+
+    # Inject return values into kwargs
+    for key, val in cb_retvals.items():
+        new_kwargs[key] = val
+
+    output = fn(*new_args, **new_kwargs)
+
+    for tup in call_after:
+        serialized_fn, serialized_args, serialized_kwargs, retval_key = tup
+        ca_fn = serialized_fn.get_deserialized()
+        ca_args = serialized_args.get_deserialized()
+        ca_kwargs = serialized_kwargs.get_deserialized()
+        ca_fn(*ca_args, **ca_kwargs)
+
+    return TransportableObject(output)
 
 
 def io_wrapper(
@@ -58,3 +115,361 @@ def io_wrapper(
         finally:
             os.chdir(current_dir)
     return output, stdout.getvalue(), stderr.getvalue(), tb
+
+
+# Copied from runner.py
+def _gather_deps(deps, call_before_objs_json, call_after_objs_json) -> Tuple[List, List]:
+    """Assemble deps for a node into the final call_before and call_after"""
+
+    call_before = []
+    call_after = []
+
+    # Rehydrate deps from JSON
+    if "bash" in deps:
+        dep = DepsBash()
+        dep.from_dict(deps["bash"])
+        call_before.append(dep.apply())
+
+    if "pip" in deps:
+        dep = DepsPip()
+        dep.from_dict(deps["pip"])
+        call_before.append(dep.apply())
+
+    for dep_json in call_before_objs_json:
+        dep = DepsCall()
+        dep.from_dict(dep_json)
+        call_before.append(dep.apply())
+
+    for dep_json in call_after_objs_json:
+        dep = DepsCall()
+        dep.from_dict(dep_json)
+        call_after.append(dep.apply())
+
+    return call_before, call_after
+
+
+# Basic wrapper for executing a topologically sorted sequence of
+# tasks. For the `task_specs` and `resources` schema see the comments
+# for `AsyncBaseExecutor.send()`.
+
+
+# URIs are just file paths
+def run_task_from_uris(
+    task_specs: List[Dict],
+    resources: dict,
+    output_uris: List[Tuple[str, str, str]],
+    results_dir: str,
+    task_group_metadata: dict,
+    server_url: str,
+):
+    prefix = "file://"
+    prefix_len = len(prefix)
+
+    outputs = {}
+    results = []
+    dispatch_id = task_group_metadata["dispatch_id"]
+    task_ids = task_group_metadata["node_ids"]
+    gid = task_group_metadata["task_group_id"]
+
+    os.environ["COVALENT_DISPATCH_ID"] = dispatch_id
+    os.environ["COVALENT_DISPATCHER_URL"] = server_url
+
+    for i, task in enumerate(task_specs):
+        result_uri, stdout_uri, stderr_uri = output_uris[i]
+
+        with open(stdout_uri, "w") as stdout, open(stderr_uri, "w") as stderr:
+            with redirect_stdout(stdout), redirect_stderr(stderr):
+                try:
+                    task_id = task["function_id"]
+                    args_ids = task["args_ids"]
+                    kwargs_ids = task["kwargs_ids"]
+
+                    function_uri = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/function"
+
+                    # Download function
+                    resp = requests.get(function_uri, stream=True)
+                    resp.raise_for_status()
+                    serialized_fn = deserialize_node_asset(resp.content, "function")
+
+                    ser_args = []
+                    ser_kwargs = {}
+
+                    # Download args and kwargs
+                    for node_id in args_ids:
+                        url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{node_id}/assets/output"
+                        resp = requests.get(url, stream=True)
+                        resp.raise_for_status()
+                        ser_args.append(deserialize_node_asset(resp.content, "output"))
+
+                    for k, node_id in kwargs_ids.items():
+                        url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{node_id}/assets/output"
+                        resp = requests.get(url, stream=True)
+                        resp.raise_for_status()
+                        ser_kwargs[k] = deserialize_node_asset(resp.content, "output")
+
+                    # Download deps
+                    deps_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/deps"
+                    resp = requests.get(deps_url, stream=True)
+                    resp.raise_for_status()
+                    deps_json = deserialize_node_asset(resp.content, "deps")
+
+                    cb_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/call_before"
+                    resp = requests.get(cb_url, stream=True)
+                    resp.raise_for_status()
+                    call_before_json = deserialize_node_asset(resp.content, "call_before")
+
+                    ca_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/call_after"
+                    resp = requests.get(ca_url, stream=True)
+                    resp.raise_for_status()
+                    call_after_json = deserialize_node_asset(resp.content, "call_after")
+
+                    # Assemble and run the task
+                    call_before, call_after = _gather_deps(
+                        deps_json, call_before_json, call_after_json
+                    )
+                    exception_occurred = False
+
+                    transportable_output = wrapper_fn(
+                        serialized_fn, call_before, call_after, *ser_args, **ser_kwargs
+                    )
+                    ser_output = serialize_node_asset(transportable_output, "output")
+                    with open(result_uri, "wb") as f:
+                        f.write(ser_output)
+
+                    outputs[task_id] = result_uri
+
+                    result_summary = {
+                        "node_id": task_id,
+                        "output_uri": result_uri,
+                        "stdout_uri": stdout_uri,
+                        "stderr_uri": stderr_uri,
+                        "exception_occurred": exception_occurred,
+                    }
+
+                except Exception as ex:
+                    exception_occurred = True
+                    tb = "".join(traceback.TracebackException.from_exception(ex).format())
+                    print(tb, file=sys.stderr)
+                    result_uri = None
+                    result_summary = {
+                        "node_id": task_id,
+                        "output_uri": result_uri,
+                        "stdout_uri": stdout_uri,
+                        "stderr_uri": stderr_uri,
+                        "exception_occurred": exception_occurred,
+                    }
+
+                    break
+
+                finally:
+                    # POST task artifacts
+                    if result_uri:
+                        upload_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/output"
+                        with open(result_uri, "rb") as f:
+                            requests.put(upload_url, data=f)
+
+                    sys.stdout.flush()
+                    if stdout_uri:
+                        upload_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/stdout"
+                        with open(stdout_uri, "rb") as f:
+                            headers = {"Content-Length": os.path.getsize(stdout_uri)}
+                            requests.put(upload_url, data=f)
+
+                    sys.stderr.flush()
+                    if stderr_uri:
+                        upload_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/assets/stderr"
+                        with open(stderr_uri, "rb") as f:
+                            headers = {"Content-Length": os.path.getsize(stderr_uri)}
+                            requests.put(upload_url, data=f)
+
+                    result_path = os.path.join(results_dir, f"result-{dispatch_id}:{task_id}.json")
+
+                    with open(result_path, "w") as f:
+                        json.dump(result_summary, f)
+
+                    results.append(result_summary)
+
+                    # Notify Covalent that the task has terminated
+                    terminal_status = "FAILED" if exception_occurred else "COMPLETED"
+                    url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/job"
+                    data = {"status": terminal_status}
+                    requests.put(url, json=data)
+
+    # Deal with any tasks that did not run
+    n = len(results)
+    if n < len(task_ids):
+        for i in range(n, len(task_ids)):
+            result_summary = {
+                "node_id": task_ids[i],
+                "output_uri": "",
+                "stdout_uri": "",
+                "stderr_uri": "",
+                "exception_occurred": True,
+            }
+
+            results.append(result_summary)
+
+            result_path = os.path.join(results_dir, f"result-{dispatch_id}:{task_id}.json")
+
+            with open(result_path, "w") as f:
+                json.dump(result_summary, f)
+
+            url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/{task_id}/job"
+            requests.put(url)
+
+
+# URIs are just file paths
+def run_task_from_uris_alt(
+    task_specs: List[Dict],
+    resources: dict,
+    output_uris: List[Tuple[str, str, str]],
+    results_dir: str,
+    task_group_metadata: dict,
+    server_url: str,
+):
+    """Alternate form of run_task_from_uris for sync executors.
+
+    This is appropriate for backends that cannot reach the Covalent
+    server. Covalent will push input assets to the executor's
+    persistent storage before invoking `Executor.send()` and pull output
+    artifacts after `Executor.receive()`.
+
+    """
+
+    prefix = "file://"
+    prefix_len = len(prefix)
+
+    outputs = {}
+    results = []
+    dispatch_id = task_group_metadata["dispatch_id"]
+    task_ids = task_group_metadata["node_ids"]
+    gid = task_group_metadata["task_group_id"]
+
+    os.environ["COVALENT_DISPATCH_ID"] = dispatch_id
+    os.environ["COVALENT_DISPATCHER_URL"] = server_url
+
+    for i, task in enumerate(task_specs):
+        result_uri, stdout_uri, stderr_uri = output_uris[i]
+
+        with open(stdout_uri, "w") as stdout, open(stderr_uri, "w") as stderr:
+            with redirect_stdout(stdout), redirect_stderr(stderr):
+                try:
+                    task_id = task["function_id"]
+                    args_ids = task["args_ids"]
+                    kwargs_ids = task["kwargs_ids"]
+
+                    # Load function
+                    function_uri = resources["functions"][task_id]
+                    if function_uri.startswith(prefix):
+                        function_uri = function_uri[prefix_len:]
+
+                    with open(function_uri, "rb") as f:
+                        serialized_fn = deserialize_node_asset(f.read(), "function")
+
+                    # Load args and kwargs
+                    ser_args = []
+                    ser_kwargs = {}
+
+                    args_uris = [resources["inputs"][index] for index in args_ids]
+                    for uri in args_uris:
+                        if uri.startswith(prefix):
+                            uri = uri[prefix_len:]
+                        with open(uri, "rb") as f:
+                            ser_args.append(deserialize_node_asset(f.read(), "output"))
+
+                    kwargs_uris = {k: resources["inputs"][v] for k, v in kwargs_ids.items()}
+                    for key, uri in kwargs_uris.items():
+                        if uri.startswith(prefix):
+                            uri = uri[prefix_len:]
+                        with open(uri, "rb") as f:
+                            ser_kwargs[key] = deserialize_node_asset(f.read(), "output")
+
+                    # Load deps
+                    deps_id = task["deps_id"]
+                    deps_uri = resources["deps"][deps_id]
+                    if deps_uri.startswith(prefix):
+                        deps_uri = deps_uri[prefix_len:]
+                    with open(deps_uri, "rb") as f:
+                        deps_json = deserialize_node_asset(f.read(), "deps")
+
+                    call_before_id = task["call_before_id"]
+                    call_before_uri = resources["deps"][call_before_id]
+                    if call_before_uri.startswith(prefix):
+                        call_before_uri = call_before_uri[prefix_len:]
+                    with open(call_before_uri, "rb") as f:
+                        call_before_json = deserialize_node_asset(f.read(), "call_before")
+
+                    call_after_id = task["call_after_id"]
+                    call_after_uri = resources["deps"][call_after_id]
+                    if call_after_uri.startswith(prefix):
+                        call_after_uri = call_after_uri[prefix_len:]
+                    with open(call_after_uri, "rb") as f:
+                        call_after_json = deserialize_node_asset(f.read(), "call_after")
+
+                    # Assemble and invoke the task
+                    call_before, call_after = _gather_deps(
+                        deps_json, call_before_json, call_after_json
+                    )
+                    exception_occurred = False
+
+                    transportable_output = wrapper_fn(
+                        serialized_fn, call_before, call_after, *ser_args, **ser_kwargs
+                    )
+                    ser_output = serialize_node_asset(transportable_output, "output")
+
+                    # Save output
+                    with open(result_uri, "wb") as f:
+                        f.write(ser_output)
+
+                    resources["inputs"][task_id] = result_uri
+
+                    result_summary = {
+                        "node_id": task_id,
+                        "output_uri": result_uri,
+                        "stdout_uri": stdout_uri,
+                        "stderr_uri": stderr_uri,
+                        "exception_occurred": exception_occurred,
+                    }
+
+                except Exception as ex:
+                    exception_occurred = True
+                    tb = "".join(traceback.TracebackException.from_exception(ex).format())
+                    print(tb, file=sys.stderr)
+                    result_uri = None
+                    result_summary = {
+                        "node_id": task_id,
+                        "output_uri": result_uri,
+                        "stdout_uri": stdout_uri,
+                        "stderr_uri": stderr_uri,
+                        "exception_occurred": exception_occurred,
+                    }
+
+                    break
+
+                finally:
+                    results.append(result_summary)
+                    result_path = os.path.join(results_dir, f"result-{dispatch_id}:{task_id}.json")
+
+                    # Write the summary file containing the URIs for
+                    # the serialized result, stdout, and stderr
+                    with open(result_path, "w") as f:
+                        json.dump(result_summary, f)
+
+    # Deal with any tasks that did not run
+    n = len(results)
+    if n < len(task_ids):
+        for i in range(n, len(task_ids)):
+            result_summary = {
+                "node_id": task_ids[i],
+                "output_uri": "",
+                "stdout_uri": "",
+                "stderr_uri": "",
+                "exception_occurred": True,
+            }
+
+            results.append(result_summary)
+
+            result_path = os.path.join(results_dir, f"result-{dispatch_id}:{task_id}.json")
+
+            with open(result_path, "w") as f:
+                json.dump(result_summary, f)
diff --git a/covalent_dispatcher/_cli/service.py b/covalent_dispatcher/_cli/service.py
index bcd7a555..faa14617 100644
--- a/covalent_dispatcher/_cli/service.py
+++ b/covalent_dispatcher/_cli/service.py
@@ -51,7 +51,7 @@ from rich.syntax import Syntax
 from rich.table import Table
 from rich.text import Text
 
-from covalent._shared_files.config import ConfigManager, get_config, set_config
+from covalent._shared_files.config import ConfigManager, get_config, reload_config, set_config
 
 from .._db.datastore import DataStore
 from .migrate import migrate_pickled_result_object
@@ -241,6 +241,11 @@ def _graceful_start(
         except requests.exceptions.ConnectionError:
             time.sleep(1)
 
+    # Since the dispatcher process might update the config file with the Dask cluster's state,
+    # we need to sync those changes with the CLI's ConfigManager instance. Otherwise the next
+    # call to `set_config()` from this module would obliterate the Dask cluster state.
+    reload_config()
+
     Path(get_config("dispatcher.cache_dir")).mkdir(parents=True, exist_ok=True)
     Path(get_config("dispatcher.results_dir")).mkdir(parents=True, exist_ok=True)
     Path(get_config("dispatcher.log_dir")).mkdir(parents=True, exist_ok=True)
diff --git a/covalent_dispatcher/_core/data_manager.py b/covalent_dispatcher/_core/data_manager.py
index 7e1967f0..c41c48c8 100644
--- a/covalent_dispatcher/_core/data_manager.py
+++ b/covalent_dispatcher/_core/data_manager.py
@@ -45,6 +45,7 @@ app_log = logger.app_log
 log_stack_info = logger.log_stack_info
 
 
+# TODO: Remove dispatch_id from the signature once qelectron_db becomes an Asset (PR #1690)
 def generate_node_result(
     dispatch_id: str,
     node_id: int,
diff --git a/covalent_dispatcher/_core/dispatcher.py b/covalent_dispatcher/_core/dispatcher.py
index 7041dde8..e1754796 100644
--- a/covalent_dispatcher/_core/dispatcher.py
+++ b/covalent_dispatcher/_core/dispatcher.py
@@ -31,7 +31,7 @@ from covalent._shared_files.defaults import WAIT_EDGE_NAME, parameter_prefix
 from covalent._shared_files.util_classes import RESULT_STATUS
 
 from . import data_manager as datasvc
-from . import runner
+from . import runner_ng
 from .data_modules import graph as tg_utils
 from .data_modules import job_manager as jbmgr
 from .dispatcher_modules.caches import _pending_parents, _sorted_task_groups, _unresolved_tasks
@@ -224,19 +224,11 @@ async def _submit_task_group(dispatch_id: str, sorted_nodes: List[int], task_gro
             app_log.debug(f"Using new runner for task group {task_group_id}")
 
             known_nodes = list(set(known_nodes))
-
-            task_spec = task_specs[0]
-            abstract_inputs = {"args": task_spec["args_ids"], "kwargs": task_spec["kwargs_ids"]}
-
-            # Temporarily redirect to in-memory runner (this is incompatible with task packing)
-            if len(task_specs) > 1:
-                raise RuntimeError("Task packing is not supported yet.")
-
-            coro = runner.run_abstract_task(
+            coro = runner_ng.run_abstract_task_group(
                 dispatch_id=dispatch_id,
-                node_id=task_group_id,
-                node_name=node_name,
-                abstract_inputs=abstract_inputs,
+                task_group_id=task_group_id,
+                task_seq=task_specs,
+                known_nodes=known_nodes,
                 selected_executor=[selected_executor, selected_executor_data],
             )
 
diff --git a/covalent_dispatcher/_core/runner.py b/covalent_dispatcher/_core/runner.py
index c945dfa8..08b990cd 100644
--- a/covalent_dispatcher/_core/runner.py
+++ b/covalent_dispatcher/_core/runner.py
@@ -15,7 +15,7 @@
 # limitations under the License.
 
 """
-Defines the core functionality of the runner
+Defines the core functionality of the legacy runner
 """
 
 import asyncio
@@ -30,8 +30,8 @@ from covalent._shared_files.config import get_config
 from covalent._shared_files.util_classes import RESULT_STATUS
 from covalent._workflow import DepsBash, DepsCall, DepsPip
 from covalent._workflow.transport import TransportableObject
-from covalent.executor.base import wrapper_fn
 from covalent.executor.utils import set_context
+from covalent.executor.utils.wrappers import wrapper_fn
 
 from . import data_manager as datasvc
 from .runner_modules import executor_proxy
diff --git a/covalent_dispatcher/_core/runner_ng.py b/covalent_dispatcher/_core/runner_ng.py
new file mode 100644
index 00000000..f4157929
--- /dev/null
+++ b/covalent_dispatcher/_core/runner_ng.py
@@ -0,0 +1,467 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Defines the core functionality of the new improved runner
+"""
+
+import asyncio
+import traceback
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+from typing import Any, Dict
+
+from covalent._shared_files import logger
+from covalent._shared_files.config import get_config
+from covalent._shared_files.exceptions import TaskCancelledError
+from covalent._shared_files.util_classes import RESULT_STATUS
+from covalent.executor.base import AsyncBaseExecutor
+from covalent.executor.schemas import ResourceMap, TaskSpec
+from covalent.executor.utils import Signals
+
+from . import data_manager as datamgr
+from . import runner as runner_legacy
+from .data_modules import asset_manager as am
+from .runner_modules import executor_proxy, jobs
+from .runner_modules.cancel import cancel_tasks  # nopycln: import
+from .runner_modules.utils import get_executor
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+debug_mode = get_config("sdk.log_level") == "debug"
+
+# Dedicated thread pool for invoking non-async Executor.cancel()
+_cancel_threadpool = ThreadPoolExecutor()
+
+# Asyncio Queue
+_job_events = None
+
+_job_event_listener = None
+
+_futures = set()
+
+# message format:
+# Ready for retrieve result
+# {"task_group_metadata": dict, "event": "READY"}
+#
+# Unable to retrieve result (e.g. credentials expired)
+#
+# {"task_group_metadata": dict, "event": "FAILED", "detail": str}
+
+
+# Domain: runner
+async def _submit_abstract_task_group(
+    dispatch_id: str,
+    task_group_id: int,
+    task_seq: list,
+    known_nodes: list,
+    executor: AsyncBaseExecutor,
+) -> None:
+    # Task sequence of the form {"function_id": task_id, "args_ids":
+    # [node_ids], "kwargs_ids": {key: node_id}}
+
+    task_ids = [task["function_id"] for task in task_seq]
+    task_specs = []
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "task_group_id": task_group_id,
+        "node_ids": task_ids,
+    }
+
+    try:
+        if not type(executor).SUPPORTS_MANAGED_EXECUTION:
+            raise NotImplementedError("Executor does not support managed execution")
+
+        resources = {"functions": {}, "inputs": {}, "deps": {}}
+
+        # Get upload URIs
+        for task_spec in task_seq:
+            task_id = task_spec["function_id"]
+
+            function_uri = executor.get_upload_uri(task_group_metadata, f"function-{task_id}")
+            deps_uri = executor.get_upload_uri(task_group_metadata, f"deps-{task_id}")
+            call_before_uri = executor.get_upload_uri(
+                task_group_metadata, f"call_before-{task_id}"
+            )
+            call_after_uri = executor.get_upload_uri(task_group_metadata, f"call_after-{task_id}")
+
+            await am.upload_asset_for_nodes(dispatch_id, "function", {task_id: function_uri})
+            await am.upload_asset_for_nodes(dispatch_id, "deps", {task_id: deps_uri})
+            await am.upload_asset_for_nodes(dispatch_id, "call_before", {task_id: call_before_uri})
+            await am.upload_asset_for_nodes(dispatch_id, "call_after", {task_id: call_after_uri})
+
+            deps_id = f"deps-{task_id}"
+            call_before_id = f"call_before-{task_id}"
+            call_after_id = f"call_after-{task_id}"
+            task_spec["deps_id"] = deps_id
+            task_spec["call_before_id"] = call_before_id
+            task_spec["call_after_id"] = call_after_id
+
+            resources["functions"][task_id] = function_uri
+            resources["deps"][deps_id] = deps_uri
+            resources["deps"][call_before_id] = call_before_uri
+            resources["deps"][call_after_id] = call_after_uri
+
+            task_specs.append(TaskSpec(**task_spec))
+
+        node_upload_uris = {
+            node_id: executor.get_upload_uri(task_group_metadata, f"node_{node_id}")
+            for node_id in known_nodes
+        }
+        resources["inputs"] = node_upload_uris
+
+        app_log.debug(
+            f"Uploading known nodes {known_nodes} for task group {dispatch_id}:{task_group_id}"
+        )
+        await am.upload_asset_for_nodes(dispatch_id, "output", node_upload_uris)
+
+        ts = datetime.now(timezone.utc)
+        node_results = [
+            datamgr.generate_node_result(
+                dispatch_id=dispatch_id,
+                node_id=task_id,
+                start_time=ts,
+                status=RESULT_STATUS.RUNNING,
+            )
+            for task_id in task_ids
+        ]
+
+        # Use one proxy for the task group; handles the following requests:
+        # - check if the job has a pending cancel request
+        # - set the job handle
+        # - set job status
+
+        # Watch the task group
+        fut = asyncio.create_task(executor_proxy.watch(dispatch_id, task_ids[0], executor))
+        _futures.add(fut)
+        fut.add_done_callback(_futures.discard)
+
+        send_retval = await executor.send(
+            task_specs,
+            ResourceMap(**resources),
+            task_group_metadata,
+        )
+
+        app_log.debug(f"Submitted task group {dispatch_id}:{task_group_id}")
+
+    except TaskCancelledError:
+        app_log.debug(f"Task group {dispatch_id}:{task_group_id} cancelled")
+
+        send_retval = None
+
+        node_results = [
+            datamgr.generate_node_result(
+                dispatch_id=dispatch_id,
+                node_id=task_id,
+                end_time=datetime.now(timezone.utc),
+                status=RESULT_STATUS.CANCELLED,
+            )
+            for task_id in task_ids
+        ]
+
+    except Exception as ex:
+        tb = "".join(traceback.TracebackException.from_exception(ex).format())
+        app_log.debug(f"Exception occurred when running task group {task_group_id}:")
+        app_log.debug(tb)
+        error_msg = tb if debug_mode else str(ex)
+        ts = datetime.now(timezone.utc)
+
+        send_retval = None
+
+        node_results = [
+            datamgr.generate_node_result(
+                dispatch_id=dispatch_id,
+                node_id=task_id,
+                end_time=datetime.now(timezone.utc),
+                status=RESULT_STATUS.FAILED,
+                error=error_msg,
+            )
+            for task_id in task_ids
+        ]
+
+    return node_results, send_retval
+
+
+async def _get_task_result(task_group_metadata: Dict, data: Any):
+    """Retrieve task results from executor.
+
+    Parameters:
+        task_group_metadata: metadata about the task group
+        data: task execution information (such as status)
+
+    Both `task_group_metadata` and `data` will be passed directly to
+    Executor.receive().
+
+    """
+
+    dispatch_id = task_group_metadata["dispatch_id"]
+    task_ids = task_group_metadata["node_ids"]
+    gid = task_group_metadata["task_group_id"]
+    app_log.debug(f"Pulling job artifacts for task group {dispatch_id}:{gid}")
+    try:
+        executor_attrs = await datamgr.electron.get(
+            dispatch_id, gid, ["executor", "executor_data"]
+        )
+        executor_name = executor_attrs["executor"]
+        executor_data = executor_attrs["executor_data"]
+
+        executor = get_executor(
+            node_id=gid,
+            selected_executor=[executor_name, executor_data],
+            loop=asyncio.get_running_loop(),
+            pool=None,
+        )
+
+        # Expects a list of TaskUpdates
+        task_group_results = await executor.receive(task_group_metadata, data)
+
+        node_results = []
+        for task_result in task_group_results:
+            task_id = task_result.node_id
+            status = task_result.status
+            await am.download_assets_for_node(dispatch_id, task_id, task_result.assets)
+
+            node_result = datamgr.generate_node_result(
+                dispatch_id=dispatch_id,
+                node_id=task_id,
+                end_time=datetime.now(timezone.utc),
+                status=status,
+            )
+            node_results.append(node_result)
+
+    except Exception as ex:
+        tb = "".join(traceback.TracebackException.from_exception(ex).format())
+        app_log.debug(f"Exception occurred when receiving task group {gid}:")
+        app_log.debug(tb)
+        error_msg = tb if debug_mode else str(ex)
+        ts = datetime.now(timezone.utc)
+
+        node_results = [
+            datamgr.generate_node_result(
+                dispatch_id=dispatch_id,
+                node_id=node_id,
+                end_time=ts,
+                status=RESULT_STATUS.FAILED,
+                error=error_msg,
+            )
+            for node_id in task_ids
+        ]
+
+    for node_result in node_results:
+        await datamgr.update_node_result(dispatch_id, node_result)
+
+
+async def run_abstract_task_group(
+    dispatch_id: str,
+    task_group_id: int,
+    task_seq: list,
+    known_nodes: list,
+    selected_executor: Any,
+) -> None:
+    executor = None
+
+    try:
+        app_log.debug(f"Attempting to instantiate executor {selected_executor}")
+        task_ids = [task["function_id"] for task in task_seq]
+        app_log.debug(f"Running task group {dispatch_id}:{task_group_id}")
+        executor = get_executor(
+            node_id=task_group_id,
+            selected_executor=selected_executor,
+            loop=asyncio.get_running_loop(),
+            pool=None,
+        )
+
+        # Check if the job should be cancelled
+        if await jobs.get_cancel_requested(dispatch_id, task_ids[0]):
+            await jobs.put_job_status(dispatch_id, task_ids[0], RESULT_STATUS.CANCELLED)
+
+            for task_id in task_ids:
+                task_metadata = {"dispatch_id": dispatch_id, "node_id": task_id}
+                app_log.debug(f"Refusing to execute cancelled task {dispatch_id}:{task_id}")
+                await mark_task_ready(task_metadata, None)
+
+            return
+
+        # Legacy runner doesn't yet support task packing
+        if not type(executor).SUPPORTS_MANAGED_EXECUTION:
+            if len(task_seq) != 1:
+                raise RuntimeError("Task packing not supported by executor plugin")
+
+            task_spec = task_seq[0]
+            node_id = task_spec["function_id"]
+            name = task_spec["name"]
+            abstract_inputs = {
+                "args": task_spec["args_ids"],
+                "kwargs": task_spec["kwargs_ids"],
+            }
+            app_log.debug(f"Reverting to legacy runner for task {task_group_id}")
+            coro = runner_legacy.run_abstract_task(
+                dispatch_id,
+                node_id,
+                name,
+                abstract_inputs,
+                selected_executor,
+            )
+            fut = asyncio.create_task(coro)
+            _futures.add(fut)
+            fut.add_done_callback(_futures.discard)
+            return
+
+        node_results, send_retval = await _submit_abstract_task_group(
+            dispatch_id,
+            task_group_id,
+            task_seq,
+            known_nodes,
+            executor,
+        )
+
+    except Exception as ex:
+        tb = "".join(traceback.TracebackException.from_exception(ex).format())
+        app_log.debug("Exception when trying to instantiate executor:")
+        app_log.debug(tb)
+        error_msg = tb if debug_mode else str(ex)
+        ts = datetime.now(timezone.utc)
+
+        send_retval = None
+        node_results = [
+            datamgr.generate_node_result(
+                dispatch_id=dispatch_id,
+                node_id=node_id,
+                start_time=ts,
+                end_time=ts,
+                status=RESULT_STATUS.FAILED,
+                error=error_msg,
+            )
+            for node_id in task_ids
+        ]
+
+    for node_result in node_results:
+        await datamgr.update_node_result(dispatch_id, node_result)
+
+    if node_results[0]["status"] == RESULT_STATUS.RUNNING:
+        task_group_metadata = {
+            "dispatch_id": dispatch_id,
+            "node_ids": task_ids,
+            "task_group_id": task_group_id,
+        }
+        await _poll_task_status(task_group_metadata, executor, send_retval)
+
+    # Terminate proxy
+    if executor:
+        executor._notify(Signals.EXIT)
+        app_log.debug(f"Stopping proxy for task group {dispatch_id}:{task_group_id}")
+
+
+async def _listen_for_job_events():
+    app_log.debug("Starting event listener")
+    while True:
+        msg = await _job_events.get()
+        try:
+            event = msg["event"]
+            app_log.debug(f"Received job event {event}")
+            if event == "BYE":
+                app_log.debug("Terminating job event listener")
+                break
+
+            # job has reached a terminal state
+            if event == "READY":
+                task_group_metadata = msg["task_group_metadata"]
+                detail = msg["detail"]
+                fut = asyncio.create_task(_get_task_result(task_group_metadata, detail))
+                _futures.add(fut)
+                fut.add_done_callback(_futures.discard)
+                continue
+
+            if event == "FAILED":
+                task_group_metadata = msg["task_group_metadata"]
+                dispatch_id = task_group_metadata["dispatch_id"]
+                gid = task_group_metadata["task_group_id"]
+                task_ids = task_group_metadata["node_ids"]
+                detail = msg["detail"]
+                ts = datetime.now(timezone.utc)
+                for task_id in task_ids:
+                    node_result = datamgr.generate_node_result(
+                        dispatch_id=dispatch_id,
+                        node_id=task_id,
+                        end_time=ts,
+                        status=RESULT_STATUS.FAILED,
+                        error=detail,
+                    )
+                    await datamgr.update_node_result(dispatch_id, node_result)
+
+        except Exception as ex:
+            app_log.exception("Error reading message: {ex}")
+
+
+async def _mark_ready(task_group_metadata: dict, detail: Any):
+    await _job_events.put(
+        {"task_group_metadata": task_group_metadata, "event": "READY", "detail": detail}
+    )
+
+
+async def _mark_failed(task_group_metadata: dict, detail: str):
+    await _job_events.put(
+        {"task_group_metadata": task_group_metadata, "event": "FAILED", "detail": detail}
+    )
+
+
+async def _poll_task_status(
+    task_group_metadata: Dict, executor: AsyncBaseExecutor, poll_data: Any
+):
+    """Polls a group of tasks until it terminates.
+
+    `poll_data` is the return value of `Executor.send()` and will be
+    passed directly to `Executor.poll()`.
+
+    """
+    # Return immediately if no polling logic (default return value is -1)
+
+    dispatch_id = task_group_metadata["dispatch_id"]
+    task_group_id = task_group_metadata["task_group_id"]
+    task_ids = task_group_metadata["node_ids"]
+
+    try:
+        app_log.debug(f"Polling status for task group {dispatch_id}:{task_group_id}")
+        receive_data = await executor.poll(task_group_metadata, poll_data)
+        await _mark_ready(task_group_metadata, receive_data)
+
+    except NotImplementedError:
+        app_log.debug(f"Executor {executor.short_name()} is async.")
+
+    except TaskCancelledError:
+        app_log.debug(f"Task group {dispatch_id}:{task_group_id} cancelled")
+        await _mark_ready(task_group_metadata, None)
+
+    except Exception as ex:
+        task_group_id = task_group_metadata["task_group_id"]
+        tb = "".join(traceback.TracebackException.from_exception(ex).format())
+        app_log.debug(f"Exception occurred when polling task {task_group_id}:")
+        app_log.debug(tb)
+        error_msg = tb if debug_mode else str(ex)
+        await _mark_failed(task_group_metadata, error_msg)
+
+
+async def mark_task_ready(task_metadata: dict, detail: Any):
+    dispatch_id = task_metadata["dispatch_id"]
+    node_id = task_metadata["node_id"]
+    gid = (await datamgr.electron.get(dispatch_id, node_id, ["task_group_id"]))["task_group_id"]
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": gid,
+    }
+
+    await _mark_ready(task_group_metadata, detail)
diff --git a/covalent_dispatcher/_dal/controller.py b/covalent_dispatcher/_dal/controller.py
index dba7969f..3e682b97 100644
--- a/covalent_dispatcher/_dal/controller.py
+++ b/covalent_dispatcher/_dal/controller.py
@@ -163,6 +163,8 @@ class Record(Generic[T]):
             equality_filters={"id": self.primary_key},
             membership_filters={},
         )
+        for k, v in values.items():
+            self._attrs[k] = v
 
     def incr(self, session: Session, *, increments: dict):
         """Increment the fields of the corresponding record."""
diff --git a/covalent_dispatcher/_service/app.py b/covalent_dispatcher/_service/app.py
index 924d4174..9a9c7d46 100644
--- a/covalent_dispatcher/_service/app.py
+++ b/covalent_dispatcher/_service/app.py
@@ -31,6 +31,7 @@ from covalent._shared_files import logger
 from covalent._shared_files.schemas.result import ResultSchema
 from covalent._shared_files.util_classes import RESULT_STATUS
 from covalent_dispatcher._core import dispatcher as core_dispatcher
+from covalent_dispatcher._core import runner_ng as core_runner
 
 from .._dal.exporters.result import export_result_manifest
 from .._dal.result import Result, get_result_object
@@ -39,9 +40,6 @@ from .._db.dispatchdb import DispatchDB
 from .heartbeat import Heartbeat
 from .models import DispatchStatusSetSchema, ExportResponseSchema, TargetDispatchStatus
 
-# from covalent_dispatcher._core import runner_ng as core_runner
-
-
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
 
@@ -59,9 +57,9 @@ async def lifespan(app: FastAPI):
     _background_tasks.add(fut)
     fut.add_done_callback(_background_tasks.discard)
 
-    # # Runner event queue and listener
-    # core_runner._job_events = asyncio.Queue()
-    # core_runner._job_event_listener = asyncio.create_task(core_runner._listen_for_job_events())
+    # Runner event queue and listener
+    core_runner._job_events = asyncio.Queue()
+    core_runner._job_event_listener = asyncio.create_task(core_runner._listen_for_job_events())
 
     # Dispatcher event queue and listener
     core_dispatcher._global_status_queue = asyncio.Queue()
@@ -79,7 +77,7 @@ async def lifespan(app: FastAPI):
         await cancel_all_with_status(status)
 
     core_dispatcher._global_event_listener.cancel()
-    # core_runner._job_event_listener.cancel()
+    core_runner._job_event_listener.cancel()
 
     Heartbeat.stop()
 
diff --git a/covalent_dispatcher/_service/app_dask.py b/covalent_dispatcher/_service/app_dask.py
index 58a4bd0b..0254ada3 100644
--- a/covalent_dispatcher/_service/app_dask.py
+++ b/covalent_dispatcher/_service/app_dask.py
@@ -20,6 +20,7 @@ import asyncio
 import os
 from logging import Logger
 from multiprocessing import Process, current_process
+from multiprocessing.connection import Connection
 from threading import Thread
 
 import dask.config
@@ -27,7 +28,7 @@ from dask.distributed import LocalCluster
 from distributed.core import Server, rpc
 
 from covalent._shared_files import logger
-from covalent._shared_files.config import get_config, update_config
+from covalent._shared_files.config import get_config
 from covalent._shared_files.utils import get_random_available_port
 
 app_log = logger.app_log
@@ -170,12 +171,15 @@ class DaskCluster(Process):
     randomly selected TCP port that is available
     """
 
-    def __init__(self, name: str, logger: Logger):
+    def __init__(self, name: str, logger: Logger, conn: Connection):
         super(DaskCluster, self).__init__()
         self.name = name
         self.logger = logger
         self.cluster = None
 
+        # For sending cluster state back to main covalent process
+        self.conn = conn
+
         # Cluster configuration
         self.num_workers = None
         self.mem_per_worker = None
@@ -219,18 +223,18 @@ class DaskCluster(Process):
         dashboard_link = self.cluster.dashboard_link
 
         try:
-            update_config(
-                {
-                    "dask": {
-                        "scheduler_address": scheduler_address,
-                        "dashboard_link": dashboard_link,
-                        "process_info": current_process(),
-                        "pid": os.getpid(),
-                        "admin_host": self.admin_host,
-                        "admin_port": self.admin_port,
-                    }
+            dask_config = {
+                "dask": {
+                    "scheduler_address": scheduler_address,
+                    "dashboard_link": dashboard_link,
+                    "process_info": str(current_process()),
+                    "pid": os.getpid(),
+                    "admin_host": self.admin_host,
+                    "admin_port": self.admin_port,
                 }
-            )
+            }
+
+            self.conn.send(dask_config)
 
             admin = DaskAdminWorker(self.cluster, self.admin_host, self.admin_port, self.logger)
             admin.start()
diff --git a/covalent_dispatcher/_service/runnersvc.py b/covalent_dispatcher/_service/runnersvc.py
new file mode 100644
index 00000000..ce0b8af9
--- /dev/null
+++ b/covalent_dispatcher/_service/runnersvc.py
@@ -0,0 +1,53 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Endpoints to update status of running tasks."""
+
+
+from fastapi import APIRouter, Request
+
+from covalent._shared_files import logger
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+
+router: APIRouter = APIRouter()
+
+
+@router.put("/dispatches/{dispatch_id}/electrons/{node_id}/job")
+async def update_task_status(dispatch_id: str, node_id: int, request: Request):
+    """Updates the status of a running task.
+
+    The request JSON will be passed to the task executor plugin's
+    `receive()` method together with `dispatch_id` and `node_id`.
+    """
+
+    from .._core import runner_ng
+
+    task_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_id": node_id,
+    }
+    try:
+        detail = await request.json()
+        f"Task {task_metadata} marked ready with detail {detail}"
+        # detail = {"status": Status(status.value.upper())}
+        await runner_ng.mark_task_ready(task_metadata, detail)
+        # app_log.debug(f"Marked task {dispatch_id}:{node_id} with status {status}")
+        return f"Task {task_metadata} marked ready with detail {detail}"
+    except Exception as e:
+        app_log.debug(f"Exception in update_task_status: {e}")
+        raise
diff --git a/covalent_ui/api/v1/routes/routes.py b/covalent_ui/api/v1/routes/routes.py
index 3aa5ae70..9b6c50b4 100644
--- a/covalent_ui/api/v1/routes/routes.py
+++ b/covalent_ui/api/v1/routes/routes.py
@@ -18,7 +18,7 @@
 
 from fastapi import APIRouter
 
-from covalent_dispatcher._service import app, assets
+from covalent_dispatcher._service import app, assets, runnersvc
 from covalent_dispatcher._triggers_app.app import router as tr_router
 from covalent_ui.api.v1.routes.end_points import (
     electron_routes,
@@ -42,5 +42,4 @@ routes.include_router(logs_route.routes, prefix="/api/v1/logs", tags=["Logs"])
 routes.include_router(tr_router, prefix="/api", tags=["Triggers"])
 routes.include_router(app.router, prefix="/api/v2", tags=["Dispatcher"])
 routes.include_router(assets.router, prefix="/api/v2", tags=["Assets"])
-# This will be enabled in the next patch
-# routes.include_router(runnersvc.router, prefix="/api/v1", tags=["Runner"])
+routes.include_router(runnersvc.router, prefix="/api/v2", tags=["Runner"])
diff --git a/covalent_ui/app.py b/covalent_ui/app.py
index aa2c830a..bf1d473e 100644
--- a/covalent_ui/app.py
+++ b/covalent_ui/app.py
@@ -16,6 +16,7 @@
 
 import argparse
 import os
+from multiprocessing import Pipe
 
 import socketio
 import uvicorn
@@ -23,7 +24,7 @@ from fastapi import Request
 from fastapi.templating import Jinja2Templates
 
 from covalent._shared_files import logger
-from covalent._shared_files.config import get_config
+from covalent._shared_files.config import get_config, update_config
 from covalent_dispatcher._service.app_dask import DaskCluster
 from covalent_dispatcher._triggers_app import triggers_only_app  # nopycln: import
 from covalent_ui.api.main import app as fastapi_app
@@ -110,8 +111,11 @@ if __name__ == "__main__":
 
     # Start dask if no-cluster flag is not specified (covalent stop auto terminates all child processes of this)
     if args.cluster:
-        dask_cluster = DaskCluster(name="LocalDaskCluster", logger=app_log)
+        parent_conn, child_conn = Pipe()
+        dask_cluster = DaskCluster(name="LocalDaskCluster", logger=app_log, conn=child_conn)
         dask_cluster.start()
+        dask_config = parent_conn.recv()
+        update_config(dask_config)
 
     app_name = "app:fastapi_app"
     if args.triggers_only:


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 99c5d607..cf30e1f1 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -211,16 +211,25 @@ jobs:
         run: |
           covalent db migrate
           if [ "${{ matrix.backend }}" = 'dask' ] ; then
-            covalent start -d
+            COVALENT_ENABLE_TASK_PACKING=1 covalent start -d
           elif [ "${{ matrix.backend }}" = 'local' ] ; then
             covalent start --no-cluster -d
           else
             echo "Invalid backend specified in test matrix."
             exit 1
           fi
+          cat $HOME/.config/covalent/covalent.conf
         env:
           COVALENT_EXECUTOR_DIR: doc/source/how_to/execution/custom_executors
 
+      - name: Print Covalent status
+        if: env.BUILD_AND_RUN_ALL
+        id: covalent_status
+        run: |
+          covalent status
+          covalent cluster --info
+          covalent cluster --logs
+
       - name: Run functional tests and measure coverage
         id: functional-tests
         if: env.BUILD_AND_RUN_ALL
diff --git a/tests/covalent_dispatcher_tests/_core/dispatcher_test.py b/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
index 5b5b7941..6f373b6d 100644
--- a/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
+++ b/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
@@ -512,11 +512,10 @@ async def test_submit_initial_tasks(mocker):
 
 
 @pytest.mark.asyncio
-async def test_submit_task_group_single(mocker):
-    """Test submitting a singleton task groups"""
+async def test_submit_task_group(mocker):
     dispatch_id = "dispatch_1"
     gid = 2
-    nodes = [2]
+    nodes = [4, 3, 2]
 
     mock_get_abs_input = mocker.patch(
         "covalent_dispatcher._core.dispatcher._get_abstract_task_inputs",
@@ -553,9 +552,8 @@ async def test_submit_task_group_single(mocker):
         "covalent_dispatcher._core.dispatcher.datasvc.update_node_result",
     )
 
-    # This will be removed in the next patch
     mock_run_abs_task = mocker.patch(
-        "covalent_dispatcher._core.dispatcher.runner.run_abstract_task",
+        "covalent_dispatcher._core.dispatcher.runner_ng.run_abstract_task_group",
     )
 
     await _submit_task_group(dispatch_id, nodes, gid)
@@ -563,59 +561,6 @@ async def test_submit_task_group_single(mocker):
     assert mock_get_abs_input.await_count == len(nodes)
 
 
-# Temporary only because the current runner does not support
-# nontrivial task groups.
-@pytest.mark.asyncio
-async def test_submit_task_group_multiple(mocker):
-    """Check that submitting multiple tasks errors out"""
-    dispatch_id = "dispatch_1"
-    gid = 2
-    nodes = [4, 3, 2]
-
-    mock_get_abs_input = mocker.patch(
-        "covalent_dispatcher._core.dispatcher._get_abstract_task_inputs",
-        return_value={"args": [], "kwargs": {}},
-    )
-
-    mock_attrs = {
-        "name": "task",
-        "value": 5,
-        "executor": "local",
-        "executor_data": {},
-    }
-
-    mock_statuses = [
-        {"status": Result.NEW_OBJ},
-        {"status": Result.NEW_OBJ},
-        {"status": Result.NEW_OBJ},
-    ]
-
-    async def get_electron_attrs(dispatch_id, node_id, keys):
-        return {key: mock_attrs[key] for key in keys}
-
-    mocker.patch(
-        "covalent_dispatcher._core.dispatcher.datasvc.electron.get",
-        get_electron_attrs,
-    )
-
-    mocker.patch(
-        "covalent_dispatcher._core.dispatcher.datasvc.electron.get_bulk",
-        return_value=mock_statuses,
-    )
-
-    mocker.patch(
-        "covalent_dispatcher._core.dispatcher.datasvc.update_node_result",
-    )
-
-    # This will be removed in the next patch
-    mock_run_abs_task = mocker.patch(
-        "covalent_dispatcher._core.dispatcher.runner.run_abstract_task",
-    )
-
-    with pytest.raises(RuntimeError):
-        await _submit_task_group(dispatch_id, nodes, gid)
-
-
 @pytest.mark.asyncio
 async def test_submit_task_group_skips_reusable(mocker):
     """Check that submit_task_group skips reusable groups"""
@@ -658,9 +603,8 @@ async def test_submit_task_group_skips_reusable(mocker):
         "covalent_dispatcher._core.dispatcher.datasvc.update_node_result",
     )
 
-    # Will be removed next patch
     mock_run_abs_task = mocker.patch(
-        "covalent_dispatcher._core.dispatcher.runner.run_abstract_task",
+        "covalent_dispatcher._core.dispatcher.runner_ng.run_abstract_task_group",
     )
 
     await _submit_task_group(dispatch_id, nodes, gid)
@@ -695,9 +639,8 @@ async def test_submit_parameter(mocker):
         "covalent_dispatcher._core.dispatcher.datasvc.update_node_result",
     )
 
-    # Will be removed next patch
     mock_run_abs_task = mocker.patch(
-        "covalent_dispatcher._core.dispatcher.runner.run_abstract_task",
+        "covalent_dispatcher._core.dispatcher.runner_ng.run_abstract_task_group",
     )
 
     await _submit_task_group(dispatch_id, [node_id], node_id)
diff --git a/tests/covalent_dispatcher_tests/_core/execution_test.py b/tests/covalent_dispatcher_tests/_core/execution_test.py
index 5dc5712f..6d521691 100644
--- a/tests/covalent_dispatcher_tests/_core/execution_test.py
+++ b/tests/covalent_dispatcher_tests/_core/execution_test.py
@@ -230,6 +230,7 @@ async def test_get_task_inputs(mocker, test_db):
     assert input_args == [1, 2]
 
 
+@pytest.mark.skip(reason="Needs to be rewritten for the new improved dispatcher")
 @pytest.mark.asyncio
 async def test_run_workflow_does_not_deserialize(test_db, mocker):
     """Check that dispatcher does not deserialize user data when using
diff --git a/tests/covalent_dispatcher_tests/_core/runner_ng_test.py b/tests/covalent_dispatcher_tests/_core/runner_ng_test.py
new file mode 100644
index 00000000..fd88b083
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_core/runner_ng_test.py
@@ -0,0 +1,714 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Tests for the core functionality of the runner.
+"""
+
+
+import asyncio
+import datetime
+import sys
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlalchemy.pool import StaticPool
+
+import covalent as ct
+from covalent._results_manager import Result
+from covalent._shared_files.exceptions import TaskCancelledError
+from covalent._shared_files.util_classes import RESULT_STATUS
+from covalent._workflow.lattice import Lattice
+from covalent.executor.base import AsyncBaseExecutor
+from covalent.executor.schemas import ResourceMap, TaskSpec, TaskUpdate
+from covalent_dispatcher._core.runner_ng import (
+    _get_task_result,
+    _listen_for_job_events,
+    _mark_failed,
+    _mark_ready,
+    _poll_task_status,
+    _submit_abstract_task_group,
+    run_abstract_task_group,
+)
+from covalent_dispatcher._dal.result import Result as SRVResult
+from covalent_dispatcher._dal.result import get_result_object
+from covalent_dispatcher._db import update
+from covalent_dispatcher._db.datastore import DataStore
+
+TEST_RESULTS_DIR = "/tmp/results"
+
+
+class MockExecutor(AsyncBaseExecutor):
+    async def run(self, function, args, kwargs, task_metadata):
+        pass
+
+
+class MockManagedExecutor(AsyncBaseExecutor):
+    SUPPORTS_MANAGED_EXECUTION = True
+
+    async def run(self, function, args, kwargs, task_metadata):
+        pass
+
+    def get_upload_uri(self, task_metadata, object_key):
+        return f"file:///tmp/{object_key}"
+
+
+@pytest.fixture
+def test_db():
+    """Instantiate and return an in-memory database."""
+
+    return DataStore(
+        db_URL="sqlite+pysqlite:///:memory:",
+        initialize_db=True,
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+
+
+def get_mock_result() -> Result:
+    """Construct a mock result object corresponding to a lattice."""
+
+    @ct.electron(executor="local")
+    def task(x):
+        print(f"stdout: {x}")
+        print("Error!", file=sys.stderr)
+        return x
+
+    @ct.lattice(deps_bash=ct.DepsBash(["ls"]))
+    def pipeline(x):
+        res1 = task(x)
+        res2 = task(res1)
+        return res2
+
+    pipeline.build_graph(x="absolute")
+    received_workflow = Lattice.deserialize_from_json(pipeline.serialize_to_json())
+    result_object = Result(received_workflow, "pipeline_workflow")
+
+    return result_object
+
+
+def get_mock_srvresult(sdkres, test_db) -> SRVResult:
+    sdkres._initialize_nodes()
+
+    update.persist(sdkres)
+
+    return get_result_object(sdkres.dispatch_id)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "task_cancelled",
+    [False, True],
+)
+async def test_submit_abstract_task_group(mocker, task_cancelled):
+    me = MockManagedExecutor()
+
+    if task_cancelled:
+        me.send = AsyncMock(side_effect=TaskCancelledError())
+    else:
+        me.send = AsyncMock(return_value="42")
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.electron.get",
+        return_value={"executor": "managed_dask", "executor_data": {}},
+    )
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.get_executor",
+        return_value=me,
+    )
+
+    ts = datetime.datetime.now()
+
+    node_result = {
+        "node_id": 0,
+        "start_time": ts,
+        "status": "RUNNING",
+    }
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.generate_node_result",
+        return_value=node_result,
+    )
+
+    mock_upload = mocker.patch(
+        "covalent_dispatcher._core.data_modules.asset_manager.upload_asset_for_nodes",
+    )
+
+    dispatch_id = "dispatch"
+    name = "task"
+    abstract_inputs = {"args": [1], "kwargs": {"key": 2}}
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [0, 3],
+        "task_group_id": 0,
+    }
+
+    mock_function_uri_0 = me.get_upload_uri(task_group_metadata, "function-0")
+    mock_deps_uri_0 = me.get_upload_uri(task_group_metadata, "deps-0")
+    mock_cb_uri_0 = me.get_upload_uri(task_group_metadata, "call_before-0")
+    mock_ca_uri_0 = me.get_upload_uri(task_group_metadata, "call_after-0")
+
+    mock_function_uri_3 = me.get_upload_uri(task_group_metadata, "function-3")
+    mock_deps_uri_3 = me.get_upload_uri(task_group_metadata, "deps-3")
+    mock_cb_uri_3 = me.get_upload_uri(task_group_metadata, "call_before-3")
+    mock_ca_uri_3 = me.get_upload_uri(task_group_metadata, "call_after-3")
+
+    mock_node_upload_uri_1 = me.get_upload_uri(task_group_metadata, "node_1")
+    mock_node_upload_uri_2 = me.get_upload_uri(task_group_metadata, "node_2")
+
+    mock_function_id_0 = 0
+    mock_args_ids = abstract_inputs["args"]
+    mock_kwargs_ids = abstract_inputs["kwargs"]
+    mock_deps_id_0 = "deps-0"
+    mock_cb_id_0 = "call_before-0"
+    mock_ca_id_0 = "call_after-0"
+
+    mock_function_id_3 = 3
+    mock_deps_id_3 = "deps-3"
+    mock_cb_id_3 = "call_before-3"
+    mock_ca_id_3 = "call_after-3"
+
+    resources = {
+        "functions": {
+            0: mock_function_uri_0,
+            3: mock_function_uri_3,
+        },
+        "inputs": {
+            1: mock_node_upload_uri_1,
+            2: mock_node_upload_uri_2,
+        },
+        "deps": {
+            mock_deps_id_0: mock_deps_uri_0,
+            mock_cb_id_0: mock_cb_uri_0,
+            mock_ca_id_0: mock_ca_uri_0,
+            mock_deps_id_3: mock_deps_uri_3,
+            mock_cb_id_3: mock_cb_uri_3,
+            mock_ca_id_3: mock_ca_uri_3,
+        },
+    }
+
+    mock_task_spec_0 = {
+        "function_id": mock_function_id_0,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+        "deps_id": mock_deps_id_0,
+        "call_before_id": mock_cb_id_0,
+        "call_after_id": mock_ca_id_0,
+    }
+
+    mock_task_spec_3 = {
+        "function_id": mock_function_id_3,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+        "deps_id": mock_deps_id_3,
+        "call_before_id": mock_cb_id_3,
+        "call_after_id": mock_ca_id_3,
+    }
+
+    mock_task_0 = {
+        "function_id": mock_function_id_0,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+
+    mock_task_3 = {
+        "function_id": mock_function_id_3,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+
+    known_nodes = [1, 2]
+
+    node_result, send_retval = await _submit_abstract_task_group(
+        dispatch_id=dispatch_id,
+        task_group_id=0,
+        task_seq=[mock_task_0, mock_task_3],
+        known_nodes=known_nodes,
+        executor=me,
+    )
+
+    mock_upload.assert_awaited()
+
+    me.send.assert_awaited_with(
+        [TaskSpec(**mock_task_spec_0), TaskSpec(**mock_task_spec_3)],
+        ResourceMap(**resources),
+        task_group_metadata,
+    )
+
+    if task_cancelled:
+        assert send_retval is None
+    else:
+        assert send_retval == "42"
+
+
+@pytest.mark.asyncio
+async def test_submit_requires_opt_in(mocker):
+    """Checks submit rejects old-style executors"""
+
+    import datetime
+
+    me = MockExecutor()
+    me.send = AsyncMock(return_value="42")
+    ts = datetime.datetime.now()
+    dispatch_id = "dispatch"
+    task_id = 0
+    name = "task"
+    abstract_inputs = {"args": [1], "kwargs": {"key": 2}}
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.electron.get",
+        return_value={"executor": "managed_dask", "executor_data": {}},
+    )
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.get_executor",
+        return_value=me,
+    )
+
+    error_msg = str(NotImplementedError("Executor does not support managed execution"))
+
+    node_result = {
+        "node_id": 0,
+        "end_time": ts,
+        "status": RESULT_STATUS.FAILED,
+        "error": error_msg,
+    }
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.generate_node_result",
+        return_value=node_result,
+    )
+    mock_function_id = task_id
+    mock_args_ids = abstract_inputs["args"]
+    mock_kwargs_ids = abstract_inputs["kwargs"]
+
+    mock_task = {
+        "function_id": mock_function_id,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+    known_nodes = [1, 2]
+
+    assert ([node_result], None) == await _submit_abstract_task_group(
+        dispatch_id, task_id, [mock_task], known_nodes, me
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_task_result(mocker):
+    me = MockManagedExecutor()
+    asset_uri = "file:///tmp/asset.pkl"
+    mock_task_result = {
+        "dispatch_id": "dispatch",
+        "node_id": 0,
+        "assets": {
+            "output": {
+                "remote_uri": asset_uri,
+            },
+            "stdout": {
+                "remote_uri": asset_uri,
+            },
+            "stderr": {
+                "remote_uri": asset_uri,
+            },
+        },
+        "status": RESULT_STATUS.COMPLETED,
+    }
+    me.receive = AsyncMock(return_value=[TaskUpdate(**mock_task_result)])
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.electron.get",
+        return_value={"executor": "managed_dask", "executor_data": {}},
+    )
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.get_executor",
+        return_value=me,
+    )
+
+    ts = datetime.datetime.now()
+
+    node_result = {
+        "node_id": 0,
+        "start_time": ts,
+        "end_time": ts,
+        "status": RESULT_STATUS.COMPLETED,
+    }
+
+    expected_node_result = {
+        "node_id": 0,
+        "start_time": ts,
+        "end_time": ts,
+        "status": RESULT_STATUS.COMPLETED,
+    }
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.generate_node_result",
+        return_value=node_result,
+    )
+
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.update_node_result",
+    )
+    mock_download = mocker.patch(
+        "covalent_dispatcher._core.data_modules.asset_manager.download_assets_for_node",
+    )
+
+    dispatch_id = "dispatch"
+    task_id = 0
+    name = "task"
+
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [task_id],
+        "task_group_id": task_id,
+    }
+    job_meta = [{"job_handle": "42", "status": "COMPLETED"}]
+
+    await _get_task_result(task_group_metadata, job_meta)
+
+    me.receive.assert_awaited_with(task_group_metadata, job_meta)
+
+    mock_update.assert_awaited_with(dispatch_id, expected_node_result)
+    mock_download.assert_awaited()
+    # Test exception during get
+    me.receive = AsyncMock(side_effect=RuntimeError())
+    mock_update.reset_mock()
+
+    await _get_task_result(task_group_metadata, job_meta)
+    mock_update.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_poll_status(mocker):
+    me = MockManagedExecutor()
+    me.poll = AsyncMock(return_value=0)
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.get_executor",
+        return_value=me,
+    )
+    mock_mark_ready = mocker.patch(
+        "covalent_dispatcher._core.runner_ng._mark_ready",
+    )
+
+    dispatch_id = "dispatch"
+    task_id = 1
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [task_id],
+        "task_group_id": task_id,
+    }
+
+    await _poll_task_status(task_group_metadata, me, "42")
+
+    mock_mark_ready.assert_awaited_with(task_group_metadata, 0)
+
+    me.poll = AsyncMock(side_effect=NotImplementedError())
+    mock_mark_ready.reset_mock()
+
+    await _poll_task_status(task_group_metadata, me, "42")
+    mock_mark_ready.assert_not_awaited()
+
+    me.poll = AsyncMock(side_effect=RuntimeError())
+    mock_mark_ready.reset_mock()
+    mock_mark_failed = mocker.patch(
+        "covalent_dispatcher._core.runner_ng._mark_failed",
+    )
+
+    await _poll_task_status(task_group_metadata, me, "42")
+    mock_mark_ready.assert_not_awaited()
+    mock_mark_failed.assert_awaited()
+    mock_mark_ready.reset_mock()
+    mock_mark_failed.reset_mock()
+
+    me.poll = AsyncMock(side_effect=TaskCancelledError())
+
+    await _poll_task_status(task_group_metadata, me, "42")
+    mock_mark_ready.assert_awaited_with(task_group_metadata, None)
+    mock_mark_failed.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_event_listener(mocker):
+    ts = datetime.datetime.now()
+    node_result = {
+        "node_id": 0,
+        "start_time": ts,
+        "end_time": ts,
+        "status": RESULT_STATUS.FAILED,
+        "error": "error",
+    }
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.generate_node_result",
+        return_value=node_result,
+    )
+
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.update_node_result",
+    )
+
+    mock_get = mocker.patch("covalent_dispatcher._core.runner_ng._get_task_result")
+
+    task_group_metadata = {"dispatch_id": "dispatch", "task_group_id": 1, "node_ids": [1]}
+
+    job_events = [{"event": "READY", "task_group_metadata": task_group_metadata}, {"event": "BYE"}]
+
+    mock_event_queue = asyncio.Queue()
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng._job_events",
+        mock_event_queue,
+    )
+    fut = asyncio.create_task(_listen_for_job_events())
+    await _mark_ready(task_group_metadata, "RUNNING")
+    await _mark_ready(task_group_metadata, "COMPLETED")
+    await mock_event_queue.put({"event": "BYE"})
+
+    await asyncio.wait_for(fut, 1)
+
+    assert mock_get.call_count == 2
+
+    mock_get.reset_mock()
+
+    fut = asyncio.create_task(_listen_for_job_events())
+
+    await _mark_failed(task_group_metadata, "error")
+    await mock_event_queue.put({"event": "BYE"})
+
+    await asyncio.wait_for(fut, 1)
+
+    mock_update.assert_awaited_with(task_group_metadata["dispatch_id"], node_result)
+
+    await mock_event_queue.put({"BAD_EVENT": "asdf"})
+    await mock_event_queue.put({"event": "BYE"})
+    mock_log = mocker.patch("covalent_dispatcher._core.runner_ng.app_log.exception")
+
+    fut = asyncio.create_task(_listen_for_job_events())
+
+    await _mark_failed(task_group_metadata, "error")
+    await mock_event_queue.put({"event": "BYE"})
+
+    await asyncio.wait_for(fut, 1)
+
+
+@pytest.mark.asyncio
+async def test_run_abstract_task_group(mocker):
+    mock_listen = AsyncMock()
+    me = MockManagedExecutor()
+    me._init_runtime()
+
+    me.poll = AsyncMock(return_value=0)
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.get_executor",
+        return_value=me,
+    )
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_modules.jobs.get_cancel_requested", return_value=False
+    )
+
+    mock_poll = mocker.patch(
+        "covalent_dispatcher._core.runner_ng._poll_task_status",
+    )
+
+    node_result = {"node_id": 0, "status": RESULT_STATUS.RUNNING}
+
+    mock_submit = mocker.patch(
+        "covalent_dispatcher._core.runner_ng._submit_abstract_task_group",
+        return_value=([node_result], 42),
+    )
+
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.update_node_result",
+    )
+
+    dispatch_id = "dispatch"
+    node_id = 0
+    node_name = "task"
+    abstract_inputs = {"args": [], "kwargs": {}}
+    selected_executor = ["local", {}]
+    mock_function_id = node_id
+    mock_args_ids = abstract_inputs["args"]
+    mock_kwargs_ids = abstract_inputs["kwargs"]
+
+    mock_task = {
+        "function_id": mock_function_id,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+    known_nodes = [1, 2]
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": node_id,
+    }
+
+    await run_abstract_task_group(
+        dispatch_id,
+        node_id,
+        [mock_task],
+        known_nodes,
+        selected_executor,
+    )
+
+    mock_submit.assert_awaited()
+    mock_update.assert_awaited()
+    mock_poll.assert_awaited_with(task_group_metadata, me, 42)
+
+
+@pytest.mark.asyncio
+async def test_run_abstract_task_group_handles_old_execs(mocker):
+    mock_listen = AsyncMock()
+    me = MockExecutor()
+    me._init_runtime()
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_ng.get_executor",
+        return_value=me,
+    )
+    mocker.patch(
+        "covalent_dispatcher._core.runner_modules.jobs.get_cancel_requested", return_value=False
+    )
+
+    mock_legacy_run = mocker.patch("covalent_dispatcher._core.runner.run_abstract_task")
+
+    mock_submit = mocker.patch("covalent_dispatcher._core.runner_ng._submit_abstract_task_group")
+
+    dispatch_id = "dispatch"
+    node_id = 0
+    node_name = "task"
+    abstract_inputs = {"args": [], "kwargs": {}}
+    selected_executor = ["local", {}]
+    mock_function_id = node_id
+    mock_args_ids = abstract_inputs["args"]
+    mock_kwargs_ids = abstract_inputs["kwargs"]
+
+    mock_task = {
+        "function_id": mock_function_id,
+        "name": node_name,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+    known_nodes = [1, 2]
+
+    await run_abstract_task_group(
+        dispatch_id,
+        node_id,
+        [mock_task],
+        known_nodes,
+        selected_executor,
+    )
+
+    mock_legacy_run.assert_called()
+    mock_submit.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_abstract_task_group_handles_bad_executors(mocker):
+    """Check handling of executors during get_executor"""
+
+    from covalent._shared_files.defaults import sublattice_prefix
+
+    mocker.patch("covalent_dispatcher._core.runner_ng.get_executor", side_effect=RuntimeError())
+
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.update_node_result",
+    )
+    dispatch_id = "dispatch"
+    node_id = 0
+    node_name = sublattice_prefix
+    abstract_inputs = {"args": [], "kwargs": {}}
+    selected_executor = ["local", {}]
+    mock_function_id = node_id
+    mock_args_ids = abstract_inputs["args"]
+    mock_kwargs_ids = abstract_inputs["kwargs"]
+
+    mock_task = {
+        "function_id": mock_function_id,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+    known_nodes = [1, 2]
+
+    await run_abstract_task_group(
+        dispatch_id,
+        node_id,
+        [mock_task],
+        known_nodes,
+        selected_executor,
+    )
+
+    mock_update.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_abstract_task_group_handles_cancelled_tasks(mocker):
+    """Check handling of cancelled tasks"""
+
+    mock_listen = AsyncMock()
+    me = MockManagedExecutor()
+    me._init_runtime()
+
+    me.poll = AsyncMock(return_value=0)
+
+    mocker.patch(
+        "covalent_dispatcher._core.runner_modules.jobs.get_cancel_requested", return_value=True
+    )
+
+    mock_jobs_put = mocker.patch(
+        "covalent_dispatcher._core.runner_modules.jobs.put_job_status", return_value=True
+    )
+
+    mock_submit = mocker.patch(
+        "covalent_dispatcher._core.runner_ng._submit_abstract_task_group",
+    )
+
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.datamgr.update_node_result",
+    )
+    mock_mark_ready = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.mark_task_ready",
+    )
+
+    dispatch_id = "dispatch"
+    node_id = 0
+    node_name = "task"
+    abstract_inputs = {"args": [], "kwargs": {}}
+    selected_executor = ["local", {}]
+    mock_function_id = node_id
+    mock_args_ids = abstract_inputs["args"]
+    mock_kwargs_ids = abstract_inputs["kwargs"]
+
+    mock_task = {
+        "function_id": mock_function_id,
+        "args_ids": mock_args_ids,
+        "kwargs_ids": mock_kwargs_ids,
+    }
+    known_nodes = [1, 2]
+
+    await run_abstract_task_group(
+        dispatch_id,
+        node_id,
+        [mock_task],
+        known_nodes,
+        selected_executor,
+    )
+
+    mock_submit.assert_not_awaited()
+    mock_update.assert_not_awaited()
+    mock_mark_ready.assert_awaited()
diff --git a/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py b/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py
index 84d647db..e5558fc0 100644
--- a/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py
+++ b/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py
@@ -70,8 +70,8 @@ TRANSPORT_GRAPH_FILENAME = "transport_graph.pkl"
 DEPS_FILENAME = "deps.pkl"
 CALL_BEFORE_FILENAME = "call_before.pkl"
 CALL_AFTER_FILENAME = "call_after.pkl"
-COVA_IMPORTS_FILENAME = "cova_imports.pkl"
-LATTICE_IMPORTS_FILENAME = "lattice_imports.pkl"
+COVA_IMPORTS_FILENAME = "cova_imports.json"
+LATTICE_IMPORTS_FILENAME = "lattice_imports.txt"
 RESULTS_DIR = "/tmp/results"
 
 
diff --git a/tests/covalent_dispatcher_tests/_service/runnersvc_test.py b/tests/covalent_dispatcher_tests/_service/runnersvc_test.py
new file mode 100644
index 00000000..61675e2e
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_service/runnersvc_test.py
@@ -0,0 +1,69 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Unit tests for the FastAPI runner endpoints"""
+
+
+import pytest
+from fastapi.testclient import TestClient
+
+from covalent_ui.app import fastapi_app as fast_app
+
+DISPATCH_ID = "f34671d1-48f2-41ce-89d9-9a8cb5c60e5d"
+
+
+@pytest.fixture
+def app():
+    yield fast_app
+
+
+@pytest.fixture
+def client():
+    with TestClient(fast_app) as c:
+        yield c
+
+
+def test_update_node_asset(mocker, client):
+    """
+    Test update task status
+    """
+    mocker.patch("covalent_dispatcher._service.app.cancel_all_with_status")
+
+    node_id = 0
+    dispatch_id = "test_update_task_status"
+    body = {"status": "COMPLETED"}
+
+    mock_mark_task_ready = mocker.patch("covalent_dispatcher._core.runner_ng.mark_task_ready")
+    resp = client.put(f"/api/v2/dispatches/{dispatch_id}/electrons/{node_id}/job", json=body)
+
+    task_metadata = {"dispatch_id": dispatch_id, "node_id": node_id}
+    mock_mark_task_ready.assert_called_with(task_metadata, body)
+
+
+def test_update_node_asset_exception(mocker, client):
+    """
+    Test update task status
+    """
+    mocker.patch("covalent_dispatcher._service.app.cancel_all_with_status")
+    node_id = 0
+    dispatch_id = "test_update_task_status"
+    body = {"status": "COMPLETED"}
+
+    mock_mark_task_ready = mocker.patch(
+        "covalent_dispatcher._core.runner_ng.mark_task_ready", side_effect=KeyError()
+    )
+    with pytest.raises(KeyError):
+        client.put(f"/api/v2/dispatches/{dispatch_id}/electrons/{node_id}/job", json=body)
diff --git a/tests/covalent_tests/executor/__init__.py b/tests/covalent_tests/executor/__init__.py
index e69de29b..cfc23bfd 100644
--- a/tests/covalent_tests/executor/__init__.py
+++ b/tests/covalent_tests/executor/__init__.py
@@ -0,0 +1,15 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
diff --git a/tests/covalent_tests/executor/base_test.py b/tests/covalent_tests/executor/base_test.py
index ae45bd1d..5c6e5856 100644
--- a/tests/covalent_tests/executor/base_test.py
+++ b/tests/covalent_tests/executor/base_test.py
@@ -26,9 +26,10 @@ import pytest
 from covalent import DepsCall, TransportableObject
 from covalent._results_manager import Result
 from covalent._shared_files.exceptions import TaskCancelledError, TaskRuntimeError
-from covalent.executor import BaseExecutor, wrapper_fn
+from covalent.executor import BaseExecutor
 from covalent.executor.base import AsyncBaseExecutor
-from covalent.executor.utils.wrappers import Signals
+from covalent.executor.utils.enums import Signals
+from covalent.executor.utils.wrappers import wrapper_fn
 
 
 class MockExecutor(BaseExecutor):
@@ -608,6 +609,21 @@ def test_base_executor_get_cancel_requested(mocker):
     mock_notify.assert_called_once()
 
 
+def test_base_executor_get_version_info(mocker):
+    """
+    Test executor invoking get cancel requested
+    """
+    me = MockExecutor()
+    me._init_runtime()
+    recv_queue = me._recv_queue
+    mock_version_info = {"python": "3.8", "covalent": "1.0"}
+    recv_queue.put_nowait((True, mock_version_info))
+    mock_notify = mocker.patch("covalent.executor.base.BaseExecutor._notify")
+
+    assert me.get_version_info() == mock_version_info
+    mock_notify.assert_called_once()
+
+
 @pytest.mark.asyncio
 async def test_async_base_executor_get_cancel_requested(mocker):
     me = MockAsyncExecutor()
@@ -618,6 +634,17 @@ async def test_async_base_executor_get_cancel_requested(mocker):
     assert await me.get_cancel_requested() is True
 
 
+@pytest.mark.asyncio
+async def test_async_base_executor_get_version_info(mocker):
+    me = MockAsyncExecutor()
+    me._init_runtime()
+    send_queue = me._send_queue
+    recv_queue = me._recv_queue
+    mock_version_info = {"python": "3.8", "covalent": "1.0"}
+    recv_queue.put_nowait((True, mock_version_info))
+    assert await me.get_version_info() == mock_version_info
+
+
 def test_base_executor_set_job_handle(mocker):
     me = MockExecutor()
     me._init_runtime()
@@ -799,5 +826,5 @@ async def test_base_async_executor_private_cancel(mocker):
 
     cancel_result = await me._cancel(task_metadata=task_metadata, job_handle=job_handle)
     mock_app_log.assert_called_with(f"Cancel not implemented for executor {type(me)}")
-    me.teardown.assert_awaited_with(task_metadata)
+    # me.teardown.assert_awaited_with(task_metadata)
     assert cancel_result is False
diff --git a/tests/covalent_tests/executor/executor_plugins/__init__.py b/tests/covalent_tests/executor/executor_plugins/__init__.py
index e69de29b..cfc23bfd 100644
--- a/tests/covalent_tests/executor/executor_plugins/__init__.py
+++ b/tests/covalent_tests/executor/executor_plugins/__init__.py
@@ -0,0 +1,15 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the Apache License 2.0 (the "License"). A copy of the
+# License may be obtained with this software package or at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Use of this file is prohibited except in compliance with the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
diff --git a/tests/covalent_tests/executor/executor_plugins/dask_test.py b/tests/covalent_tests/executor/executor_plugins/dask_test.py
index 81b3b855..c28bc52f 100644
--- a/tests/covalent_tests/executor/executor_plugins/dask_test.py
+++ b/tests/covalent_tests/executor/executor_plugins/dask_test.py
@@ -17,16 +17,29 @@
 """Tests for Covalent dask executor."""
 
 import asyncio
+import io
+import json
 import os
+import sys
 import tempfile
 from unittest.mock import AsyncMock
 
 import pytest
+from dask.distributed import LocalCluster
 
 import covalent as ct
 from covalent._shared_files import TaskRuntimeError
 from covalent._shared_files.exceptions import TaskCancelledError
-from covalent.executor.executor_plugins.dask import _EXECUTOR_PLUGIN_DEFAULTS, DaskExecutor
+from covalent._workflow.transportable_object import TransportableObject
+from covalent.executor.executor_plugins.dask import (
+    _EXECUTOR_PLUGIN_DEFAULTS,
+    DaskExecutor,
+    ResourceMap,
+    TaskSpec,
+    dask_wrapper,
+    run_task_from_uris_alt,
+)
+from covalent.executor.utils.serialize import serialize_node_asset
 
 
 def test_dask_executor_init(mocker):
@@ -49,8 +62,6 @@ def test_dask_executor_init(mocker):
 
 
 def test_dask_executor_with_workdir(mocker):
-    from dask.distributed import LocalCluster
-
     with tempfile.TemporaryDirectory() as tmp_dir:
         lc = LocalCluster()
         de = ct.executor.DaskExecutor(
@@ -87,10 +98,6 @@ def test_dask_executor_with_workdir(mocker):
 
 
 def test_dask_wrapper_fn(mocker):
-    import sys
-
-    from covalent.executor.executor_plugins.dask import dask_wrapper
-
     def f(x):
         print("Hello", file=sys.stdout)
         print("Bye", file=sys.stderr)
@@ -107,10 +114,6 @@ def test_dask_wrapper_fn(mocker):
 
 
 def test_dask_wrapper_fn_exception_handling(mocker):
-    import sys
-
-    from covalent.executor.executor_plugins.dask import dask_wrapper
-
     def f(x):
         raise RuntimeError("error")
 
@@ -126,13 +129,6 @@ def test_dask_wrapper_fn_exception_handling(mocker):
 def test_dask_executor_run(mocker):
     """Test run method for Dask executor"""
 
-    import io
-    import sys
-
-    from dask.distributed import LocalCluster
-
-    from covalent.executor import DaskExecutor
-
     cluster = LocalCluster()
 
     dask_exec = DaskExecutor(cluster.scheduler_address)
@@ -164,12 +160,6 @@ def test_dask_executor_run_cancel_requested(mocker):
     """
     Test dask executor cancel request
     """
-    import io
-    import sys
-
-    from dask.distributed import LocalCluster
-
-    from covalent.executor import DaskExecutor
 
     cluster = LocalCluster()
 
@@ -198,13 +188,6 @@ def test_dask_executor_run_cancel_requested(mocker):
 def test_dask_executor_run_exception_handling(mocker):
     """Test run method for Dask executor"""
 
-    import io
-    import sys
-
-    from dask.distributed import LocalCluster
-
-    from covalent.executor import DaskExecutor
-
     cluster = LocalCluster()
 
     dask_exec = DaskExecutor(cluster.scheduler_address)
@@ -235,9 +218,6 @@ def test_dask_app_log_debug_when_cancel_requested(mocker):
     """
     Test logging when task cancellation is requested
     """
-    from dask.distributed import LocalCluster
-
-    from covalent.executor import DaskExecutor
 
     cluster = LocalCluster()
 
@@ -268,9 +248,6 @@ def test_dask_task_cancel(mocker):
     """
     Test dask task cancellation method
     """
-    from dask.distributed import LocalCluster
-
-    from covalent.executor import DaskExecutor
 
     cluster = LocalCluster()
 
@@ -284,3 +261,351 @@ def test_dask_task_cancel(mocker):
     result = asyncio.run(dask_exec.cancel(task_metadata, job_handle))
     mock_app_log.assert_called_with(f"Cancelled future with key {job_handle}")
     assert result is True
+
+
+def test_dask_send_poll_receive(mocker):
+    """Test running a task using send + poll + receive."""
+
+    cluster = LocalCluster()
+    dask_exec = DaskExecutor()
+
+    mock_get_config = mocker.patch(
+        "covalent.executor.executor_plugins.dask.get_config",
+        return_value=cluster.scheduler_address,
+    )
+
+    mock_get_cancel_requested = mocker.patch.object(
+        dask_exec, "get_cancel_requested", AsyncMock(return_value=False)
+    )
+    mock_set_job_handle = mocker.patch.object(dask_exec, "set_job_handle", AsyncMock())
+
+    def task(x, y):
+        return x + y
+
+    dispatch_id = "test_dask_send_receive"
+    node_id = 0
+    task_group_id = 0
+
+    x = TransportableObject(1)
+    y = TransportableObject(2)
+    deps = {}
+    call_before = []
+    call_after = []
+
+    ser_task = serialize_node_asset(TransportableObject(task), "function")
+    ser_deps = serialize_node_asset(deps, "deps")
+    ser_cb = serialize_node_asset(deps, "call_before")
+    ser_ca = serialize_node_asset(deps, "call_after")
+    ser_x = serialize_node_asset(x, "output")
+    ser_y = serialize_node_asset(y, "output")
+
+    node_0_file = tempfile.NamedTemporaryFile("wb")
+    node_0_file.write(ser_task)
+    node_0_file.flush()
+
+    deps_file = tempfile.NamedTemporaryFile("wb")
+    deps_file.write(ser_deps)
+    deps_file.flush()
+
+    cb_file = tempfile.NamedTemporaryFile("wb")
+    cb_file.write(ser_cb)
+    cb_file.flush()
+
+    ca_file = tempfile.NamedTemporaryFile("wb")
+    ca_file.write(ser_ca)
+    ca_file.flush()
+
+    node_1_file = tempfile.NamedTemporaryFile("wb")
+    node_1_file.write(ser_x)
+    node_1_file.flush()
+
+    node_2_file = tempfile.NamedTemporaryFile("wb")
+    node_2_file.write(ser_y)
+    node_2_file.flush()
+
+    task_spec = TaskSpec(
+        function_id=0,
+        args_ids=[1, 2],
+        kwargs_ids={},
+        deps_id="deps",
+        call_before_id="call_before",
+        call_after_id="call_after",
+    )
+
+    resources = ResourceMap(
+        functions={
+            0: node_0_file.name,
+        },
+        inputs={
+            1: node_1_file.name,
+            2: node_2_file.name,
+        },
+        deps={
+            "deps": deps_file.name,
+            "call_before": cb_file.name,
+            "call_after": ca_file.name,
+        },
+    )
+
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": task_group_id,
+    }
+
+    async def run_dask_job(task_specs, resources, task_group_metadata):
+        job_id = await dask_exec.send(task_specs, resources, task_group_metadata)
+        job_status = await dask_exec.poll(task_group_metadata, job_id)
+        task_updates = await dask_exec.receive(task_group_metadata, job_status)
+        return job_status, task_updates
+
+    job_status, task_updates = asyncio.run(
+        run_dask_job([task_spec], resources, task_group_metadata)
+    )
+
+    mock_get_config.assert_called_once()
+
+    assert job_status["status"] == "READY"
+    assert len(task_updates) == 1
+    task_update = task_updates[0]
+    assert str(task_update.status) == (ct.status.COMPLETED)
+    output_uri = task_update.assets["output"].remote_uri
+
+    with open(output_uri, "rb") as f:
+        output = TransportableObject.deserialize(f.read())
+    assert output.get_deserialized() == 3
+
+
+def test_run_task_from_uris_alt():
+    """Test the wrapper submitted to dask"""
+
+    def task(x, y):
+        return x + y
+
+    dispatch_id = "test_dask_send_receive"
+    node_id = 0
+    task_group_id = 0
+
+    x = TransportableObject(1)
+    y = TransportableObject(2)
+    deps = {}
+
+    cb_tmpfile = tempfile.NamedTemporaryFile()
+    ca_tmpfile = tempfile.NamedTemporaryFile()
+
+    call_before = [ct.DepsBash([f"echo Hello > {cb_tmpfile.name}"]).to_dict()]
+    call_after = [ct.DepsBash(f"echo Bye > {ca_tmpfile.name}").to_dict()]
+
+    ser_task = serialize_node_asset(TransportableObject(task), "function")
+    ser_deps = serialize_node_asset(deps, "deps")
+    ser_cb = serialize_node_asset(call_before, "call_before")
+    ser_ca = serialize_node_asset(call_after, "call_after")
+    ser_x = serialize_node_asset(x, "output")
+    ser_y = serialize_node_asset(y, "output")
+
+    node_0_file = tempfile.NamedTemporaryFile("wb")
+    node_0_file.write(ser_task)
+    node_0_file.flush()
+
+    deps_file = tempfile.NamedTemporaryFile("wb")
+    deps_file.write(ser_deps)
+    deps_file.flush()
+
+    cb_file = tempfile.NamedTemporaryFile("wb")
+    cb_file.write(ser_cb)
+    cb_file.flush()
+
+    ca_file = tempfile.NamedTemporaryFile("wb")
+    ca_file.write(ser_ca)
+    ca_file.flush()
+
+    node_1_file = tempfile.NamedTemporaryFile("wb")
+    node_1_file.write(ser_x)
+    node_1_file.flush()
+
+    node_2_file = tempfile.NamedTemporaryFile("wb")
+    node_2_file.write(ser_y)
+    node_2_file.flush()
+
+    task_spec = TaskSpec(
+        function_id=0,
+        args_ids=[1, 2],
+        kwargs_ids={},
+        deps_id="deps",
+        call_before_id="call_before",
+        call_after_id="call_after",
+    )
+
+    resources = ResourceMap(
+        functions={
+            0: node_0_file.name,
+        },
+        inputs={
+            1: node_1_file.name,
+            2: node_2_file.name,
+        },
+        deps={
+            "deps": deps_file.name,
+            "call_before": cb_file.name,
+            "call_after": ca_file.name,
+        },
+    )
+
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": task_group_id,
+    }
+
+    result_file = tempfile.NamedTemporaryFile()
+    stdout_file = tempfile.NamedTemporaryFile()
+    stderr_file = tempfile.NamedTemporaryFile()
+
+    results_dir = tempfile.TemporaryDirectory()
+
+    run_task_from_uris_alt(
+        task_specs=[task_spec.dict()],
+        resources=resources.dict(),
+        output_uris=[(result_file.name, stdout_file.name, stderr_file.name)],
+        results_dir=results_dir.name,
+        task_group_metadata=task_group_metadata,
+        server_url="http://localhost:48008",
+    )
+
+    with open(result_file.name, "rb") as f:
+        output = TransportableObject.deserialize(f.read())
+    assert output.get_deserialized() == 3
+
+    with open(cb_tmpfile.name, "r") as f:
+        assert f.read() == "Hello\n"
+
+    with open(ca_tmpfile.name, "r") as f:
+        assert f.read() == "Bye\n"
+
+
+def test_run_task_from_uris_alt_exception():
+    """Test the wrapper submitted to dask"""
+
+    def task(x, y):
+        assert False
+
+    dispatch_id = "test_dask_send_receive"
+    node_id = 0
+    task_group_id = 0
+
+    x = TransportableObject(1)
+    y = TransportableObject(2)
+    deps = {}
+
+    cb_tmpfile = tempfile.NamedTemporaryFile()
+    ca_tmpfile = tempfile.NamedTemporaryFile()
+
+    call_before = [ct.DepsBash([f"echo Hello > {cb_tmpfile.name}"]).to_dict()]
+    call_after = [ct.DepsBash(f"echo Bye > {ca_tmpfile.name}").to_dict()]
+
+    ser_task = serialize_node_asset(TransportableObject(task), "function")
+    ser_deps = serialize_node_asset(deps, "deps")
+    ser_cb = serialize_node_asset(call_before, "call_before")
+    ser_ca = serialize_node_asset(call_after, "call_after")
+    ser_x = serialize_node_asset(x, "output")
+    ser_y = serialize_node_asset(y, "output")
+
+    node_0_file = tempfile.NamedTemporaryFile("wb")
+    node_0_file.write(ser_task)
+    node_0_file.flush()
+
+    deps_file = tempfile.NamedTemporaryFile("wb")
+    deps_file.write(ser_deps)
+    deps_file.flush()
+
+    cb_file = tempfile.NamedTemporaryFile("wb")
+    cb_file.write(ser_cb)
+    cb_file.flush()
+
+    ca_file = tempfile.NamedTemporaryFile("wb")
+    ca_file.write(ser_ca)
+    ca_file.flush()
+
+    node_1_file = tempfile.NamedTemporaryFile("wb")
+    node_1_file.write(ser_x)
+    node_1_file.flush()
+
+    node_2_file = tempfile.NamedTemporaryFile("wb")
+    node_2_file.write(ser_y)
+    node_2_file.flush()
+
+    task_spec = TaskSpec(
+        function_id=0,
+        args_ids=[1],
+        kwargs_ids={"y": 2},
+        deps_id="deps",
+        call_before_id="call_before",
+        call_after_id="call_after",
+    )
+
+    resources = ResourceMap(
+        functions={
+            0: f"file://{node_0_file.name}",
+        },
+        inputs={
+            1: f"file://{node_1_file.name}",
+            2: f"file://{node_2_file.name}",
+        },
+        deps={
+            "deps": f"file://{deps_file.name}",
+            "call_before": f"file://{cb_file.name}",
+            "call_after": f"file://{ca_file.name}",
+        },
+    )
+
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": task_group_id,
+    }
+
+    result_file = tempfile.NamedTemporaryFile()
+    stdout_file = tempfile.NamedTemporaryFile()
+    stderr_file = tempfile.NamedTemporaryFile()
+
+    results_dir = tempfile.TemporaryDirectory()
+
+    run_task_from_uris_alt(
+        task_specs=[task_spec.dict()],
+        resources=resources.dict(),
+        output_uris=[(result_file.name, stdout_file.name, stderr_file.name)],
+        results_dir=results_dir.name,
+        task_group_metadata=task_group_metadata,
+        server_url="http://localhost:48008",
+    )
+    summary_file_path = f"{results_dir.name}/result-{dispatch_id}:{node_id}.json"
+
+    with open(summary_file_path, "r") as f:
+        summary = json.load(f)
+        assert summary["exception_occurred"] is True
+
+
+def test_get_upload_uri():
+    """
+    Test the get_upload_uri method
+    """
+
+    dispatch_id = "test_dask_send_receive"
+    node_id = 0
+    task_group_id = 0
+
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": task_group_id,
+    }
+
+    object_key = "test_object_key"
+
+    dask_exec = DaskExecutor()
+
+    path_string = str(dask_exec.get_upload_uri(task_group_metadata, object_key))
+
+    assert dispatch_id in path_string
+    assert str(task_group_id) in path_string
+    assert object_key in path_string
diff --git a/tests/covalent_tests/executor/executor_plugins/local_test.py b/tests/covalent_tests/executor/executor_plugins/local_test.py
index 93fa8a04..c5c34ae5 100644
--- a/tests/covalent_tests/executor/executor_plugins/local_test.py
+++ b/tests/covalent_tests/executor/executor_plugins/local_test.py
@@ -17,10 +17,11 @@
 """Tests for Covalent local executor."""
 
 import io
+import json
 import os
 import tempfile
 from functools import partial
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 
@@ -28,8 +29,17 @@ import covalent as ct
 from covalent._shared_files import TaskRuntimeError
 from covalent._shared_files.exceptions import TaskCancelledError
 from covalent._workflow.transport import TransportableObject
-from covalent.executor.base import wrapper_fn
-from covalent.executor.executor_plugins.local import _EXECUTOR_PLUGIN_DEFAULTS, LocalExecutor
+from covalent.executor.executor_plugins.local import (
+    _EXECUTOR_PLUGIN_DEFAULTS,
+    RESULT_STATUS,
+    LocalExecutor,
+    StatusEnum,
+    TaskSpec,
+    run_task_from_uris,
+)
+from covalent.executor.schemas import ResourceMap
+from covalent.executor.utils.serialize import serialize_node_asset
+from covalent.executor.utils.wrappers import wrapper_fn
 
 
 def test_local_executor_init(mocker):
@@ -227,3 +237,503 @@ def test_local_executor_get_cancel_requested(mocker):
         le.run(local_executor_run__mock_task, args, kwargs, task_metadata)
         le.get_cancel_requested.assert_called_once()
         assert mock_app_log.call_count == 2
+
+
+def test_run_task_from_uris(mocker):
+    """Test the wrapper submitted to local"""
+
+    def task(x, y):
+        return x + y
+
+    dispatch_id = "test_local_send_receive"
+    node_id = 0
+    task_group_id = 0
+    server_url = "http://localhost:48008"
+
+    x = TransportableObject(1)
+    y = TransportableObject(2)
+    deps = {}
+
+    cb_tmpfile = tempfile.NamedTemporaryFile()
+    ca_tmpfile = tempfile.NamedTemporaryFile()
+
+    deps = {
+        "bash": ct.DepsBash([f"echo Hello > {cb_tmpfile.name}"]).to_dict(),
+        "pip": ct.DepsBash(f"echo Bye > {ca_tmpfile.name}").to_dict(),
+    }
+
+    call_before = []
+    call_after = []
+
+    ser_task = serialize_node_asset(TransportableObject(task), "function")
+    ser_deps = serialize_node_asset(deps, "deps")
+    ser_cb = serialize_node_asset(call_before, "call_before")
+    ser_ca = serialize_node_asset(call_after, "call_after")
+    ser_x = serialize_node_asset(x, "output")
+    ser_y = serialize_node_asset(y, "output")
+
+    node_0_file = tempfile.NamedTemporaryFile("wb")
+    node_0_file.write(ser_task)
+    node_0_file.flush()
+    node_0_function_url = (
+        f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/function"
+    )
+
+    deps_file = tempfile.NamedTemporaryFile("wb")
+    deps_file.write(ser_deps)
+    deps_file.flush()
+    deps_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/deps"
+
+    cb_file = tempfile.NamedTemporaryFile("wb")
+    cb_file.write(ser_cb)
+    cb_file.flush()
+    cb_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/call_before"
+
+    ca_file = tempfile.NamedTemporaryFile("wb")
+    ca_file.write(ser_ca)
+    ca_file.flush()
+    ca_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/call_after"
+
+    node_1_file = tempfile.NamedTemporaryFile("wb")
+    node_1_file.write(ser_x)
+    node_1_file.flush()
+    node_1_output_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/1/assets/output"
+
+    node_2_file = tempfile.NamedTemporaryFile("wb")
+    node_2_file.write(ser_y)
+    node_2_file.flush()
+    node_2_output_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/2/assets/output"
+
+    task_spec = TaskSpec(
+        function_id=0,
+        args_ids=[1, 2],
+        kwargs_ids={},
+        deps_id="deps",
+        call_before_id="call_before",
+        call_after_id="call_after",
+    )
+
+    resources = {
+        node_0_function_url: ser_task,
+        node_1_output_url: ser_x,
+        node_2_output_url: ser_y,
+        deps_url: ser_deps,
+        cb_url: ser_cb,
+        ca_url: ser_ca,
+    }
+
+    def mock_req_get(url, stream):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.content = resources[url]
+        return mock_resp
+
+    def mock_req_post(url, files):
+        resources[url] = files["asset_file"].read()
+
+    mocker.patch("requests.get", mock_req_get)
+    mocker.patch("requests.post", mock_req_post)
+    mock_put = mocker.patch("requests.put")
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": task_group_id,
+    }
+
+    result_file = tempfile.NamedTemporaryFile()
+    stdout_file = tempfile.NamedTemporaryFile()
+    stderr_file = tempfile.NamedTemporaryFile()
+
+    results_dir = tempfile.TemporaryDirectory()
+
+    run_task_from_uris(
+        task_specs=[task_spec.dict()],
+        resources={},
+        output_uris=[(result_file.name, stdout_file.name, stderr_file.name)],
+        results_dir=results_dir.name,
+        task_group_metadata=task_group_metadata,
+        server_url=server_url,
+    )
+
+    with open(result_file.name, "rb") as f:
+        output = TransportableObject.deserialize(f.read())
+    assert output.get_deserialized() == 3
+
+    with open(cb_tmpfile.name, "r") as f:
+        assert f.read() == "Hello\n"
+
+    with open(ca_tmpfile.name, "r") as f:
+        assert f.read() == "Bye\n"
+
+    mock_put.assert_called()
+
+
+def test_run_task_from_uris_exception(mocker):
+    """Test the wrapper submitted to local"""
+
+    def task(x, y):
+        assert False
+
+    dispatch_id = "test_local_send_receive"
+    node_id = 0
+    task_group_id = 0
+    server_url = "http://localhost:48008"
+
+    x = TransportableObject(1)
+    y = TransportableObject(2)
+    deps = {}
+
+    cb_tmpfile = tempfile.NamedTemporaryFile()
+    ca_tmpfile = tempfile.NamedTemporaryFile()
+
+    deps = {
+        "bash": ct.DepsBash([f"echo Hello > {cb_tmpfile.name}"]).to_dict(),
+        "pip": ct.DepsBash(f"echo Bye > {ca_tmpfile.name}").to_dict(),
+    }
+
+    call_before = []
+    call_after = []
+
+    ser_task = serialize_node_asset(TransportableObject(task), "function")
+    ser_deps = serialize_node_asset(deps, "deps")
+    ser_cb = serialize_node_asset(call_before, "call_before")
+    ser_ca = serialize_node_asset(call_after, "call_after")
+    ser_x = serialize_node_asset(x, "output")
+    ser_y = serialize_node_asset(y, "output")
+
+    node_0_file = tempfile.NamedTemporaryFile("wb")
+    node_0_file.write(ser_task)
+    node_0_file.flush()
+    node_0_function_url = (
+        f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/function"
+    )
+
+    deps_file = tempfile.NamedTemporaryFile("wb")
+    deps_file.write(ser_deps)
+    deps_file.flush()
+    deps_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/deps"
+
+    cb_file = tempfile.NamedTemporaryFile("wb")
+    cb_file.write(ser_cb)
+    cb_file.flush()
+    cb_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/call_before"
+
+    ca_file = tempfile.NamedTemporaryFile("wb")
+    ca_file.write(ser_ca)
+    ca_file.flush()
+    ca_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/0/assets/call_after"
+
+    node_1_file = tempfile.NamedTemporaryFile("wb")
+    node_1_file.write(ser_x)
+    node_1_file.flush()
+    node_1_output_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/1/assets/output"
+
+    node_2_file = tempfile.NamedTemporaryFile("wb")
+    node_2_file.write(ser_y)
+    node_2_file.flush()
+    node_2_output_url = f"{server_url}/api/v2/dispatches/{dispatch_id}/electrons/2/assets/output"
+
+    task_spec = TaskSpec(
+        function_id=0,
+        args_ids=[1],
+        kwargs_ids={"y": 2},
+        deps_id="deps",
+        call_before_id="call_before",
+        call_after_id="call_after",
+    )
+
+    resources = {
+        node_0_function_url: ser_task,
+        node_1_output_url: ser_x,
+        node_2_output_url: ser_y,
+        deps_url: ser_deps,
+        cb_url: ser_cb,
+        ca_url: ser_ca,
+    }
+
+    def mock_req_get(url, stream):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.content = resources[url]
+        return mock_resp
+
+    def mock_req_post(url, files):
+        resources[url] = files["asset_file"].read()
+
+    mocker.patch("requests.get", mock_req_get)
+    mocker.patch("requests.post", mock_req_post)
+    mocker.patch("requests.put")
+    task_group_metadata = {
+        "dispatch_id": dispatch_id,
+        "node_ids": [node_id],
+        "task_group_id": task_group_id,
+    }
+
+    result_file = tempfile.NamedTemporaryFile()
+    stdout_file = tempfile.NamedTemporaryFile()
+    stderr_file = tempfile.NamedTemporaryFile()
+
+    results_dir = tempfile.TemporaryDirectory()
+
+    run_task_from_uris(
+        task_specs=[task_spec.dict()],
+        resources={},
+        output_uris=[(result_file.name, stdout_file.name, stderr_file.name)],
+        results_dir=results_dir.name,
+        task_group_metadata=task_group_metadata,
+        server_url=server_url,
+    )
+
+    summary_file_path = f"{results_dir.name}/result-{dispatch_id}:{node_id}.json"
+
+    with open(summary_file_path, "r") as f:
+        summary = json.load(f)
+        assert summary["exception_occurred"] is True
+
+
+# Mocks for external dependencies
+@pytest.fixture
+def mock_os_path_join():
+    with patch("os.path.join", return_value="mock_path") as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_format_server_url():
+    with patch(
+        "covalent.executor.executor_plugins.local.format_server_url",
+        return_value="mock_server_url",
+    ) as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_future():
+    mock = Mock()
+    mock.cancelled.return_value = False
+    return mock
+
+
+@pytest.fixture
+def mock_proc_pool_submit(mock_future):
+    with patch(
+        "covalent.executor.executor_plugins.local.proc_pool.submit", return_value=mock_future
+    ) as mock:
+        yield mock
+
+
+# Test cases
+test_cases = [
+    # Happy path
+    {
+        "id": "happy_path",
+        "task_specs": [
+            TaskSpec(
+                function_id=0,
+                args_ids=[1],
+                kwargs_ids={"y": 2},
+                deps_id="deps",
+                call_before_id="call_before",
+                call_after_id="call_after",
+            )
+        ],
+        "resources": ResourceMap(
+            functions={0: "mock_function_uri"},
+            inputs={1: "mock_input_uri"},
+            deps={"deps": "mock_deps_uri"},
+        ),
+        "task_group_metadata": {"dispatch_id": "1", "node_ids": ["1"], "task_group_id": "1"},
+        "expected_output_uris": [("mock_path", "mock_path", "mock_path")],
+        "expected_server_url": "mock_server_url",
+        "expected_future_cancelled": False,
+    },
+    {
+        "id": "future_cancelled",
+        "task_specs": [
+            TaskSpec(
+                function_id=0,
+                args_ids=[1],
+                kwargs_ids={"y": 2},
+                deps_id="deps",
+                call_before_id="call_before",
+                call_after_id="call_after",
+            )
+        ],
+        "resources": ResourceMap(
+            functions={0: "mock_function_uri"},
+            inputs={1: "mock_input_uri"},
+            deps={"deps": "mock_deps_uri"},
+        ),
+        "task_group_metadata": {"dispatch_id": "1", "node_ids": ["1"], "task_group_id": "1"},
+        "expected_output_uris": [("mock_path", "mock_path", "mock_path")],
+        "expected_server_url": "mock_server_url",
+        "expected_future_cancelled": True,
+    },
+]
+
+
+@pytest.mark.parametrize("test_case", test_cases, ids=[tc["id"] for tc in test_cases])
+def test_send_internal(
+    test_case,
+    mock_os_path_join,
+    mock_format_server_url,
+    mock_future,
+    mock_proc_pool_submit,
+):
+    """Test the internal _send function of LocalExecutor"""
+
+    local_exec = LocalExecutor()
+
+    # Arrange
+    local_exec.cache_dir = "mock_cache_dir"
+    mock_future.cancelled.return_value = test_case["expected_future_cancelled"]
+
+    # Act
+    local_exec._send(
+        test_case["task_specs"],
+        test_case["resources"],
+        test_case["task_group_metadata"],
+    )
+
+    # Assert
+    mock_os_path_join.assert_called()
+    mock_format_server_url.assert_called_once_with()
+    mock_proc_pool_submit.assert_called_once_with(
+        run_task_from_uris,
+        list(map(lambda t: t.dict(), test_case["task_specs"])),
+        test_case["resources"].dict(),
+        test_case["expected_output_uris"],
+        "mock_cache_dir",
+        test_case["task_group_metadata"],
+        test_case["expected_server_url"],
+    )
+
+
+@pytest.mark.asyncio
+async def test_send(mocker):
+    """Test the send function of LocalExecutor"""
+
+    local_exec = LocalExecutor()
+
+    # Arrange
+    task_group_metadata = {"dispatch_id": "1", "node_ids": ["1", "2"]}
+    task_spec = TaskSpec(
+        function_id=0,
+        args_ids=[1],
+        kwargs_ids={"y": 2},
+        deps_id="deps",
+        call_before_id="call_before",
+        call_after_id="call_after",
+    )
+    resource = ResourceMap(
+        functions={0: "mock_function_uri"},
+        inputs={1: "mock_input_uri"},
+        deps={"deps": "mock_deps_uri"},
+    )
+
+    mock_loop = mocker.Mock()
+
+    mock_get_running_loop = mocker.patch(
+        "covalent.executor.executor_plugins.local.asyncio.get_running_loop",
+        return_value=mock_loop,
+    )
+    mock_get_running_loop.return_value.run_in_executor = mocker.AsyncMock()
+
+    await local_exec.send(
+        [task_spec],
+        resource,
+        task_group_metadata,
+    )
+
+    mock_get_running_loop.assert_called_once()
+
+    mock_get_running_loop.return_value.run_in_executor.assert_awaited_once_with(
+        None,
+        local_exec._send,
+        [task_spec],
+        resource,
+        task_group_metadata,
+    )
+
+
+# Test data
+test_data = [
+    # Happy path tests
+    {
+        "id": "HP1",
+        "task_group_metadata": {"dispatch_id": "1", "node_ids": ["1", "2"]},
+        "data": {"status": StatusEnum.COMPLETED},
+        "expected_status": StatusEnum.COMPLETED,
+    },
+    {
+        "id": "HP2",
+        "task_group_metadata": {"dispatch_id": "2", "node_ids": ["3", "4"]},
+        "data": {"status": StatusEnum.FAILED},
+        "expected_status": StatusEnum.FAILED,
+    },
+    # Edge case tests
+    {
+        "id": "EC1",
+        "task_group_metadata": {"dispatch_id": "3", "node_ids": []},
+        "data": {"status": StatusEnum.COMPLETED},
+        "expected_status": StatusEnum.COMPLETED,
+    },
+    {
+        "id": "EC2",
+        "task_group_metadata": {"dispatch_id": "4", "node_ids": ["5"]},
+        "data": None,
+        "expected_status": RESULT_STATUS.CANCELLED,
+    },
+]
+
+
+@pytest.mark.parametrize("test_case", test_data, ids=[tc["id"] for tc in test_data])
+def test_receive_internal(test_case):
+    """Test the internal _receive function of LocalExecutor"""
+
+    local_exec = LocalExecutor()
+
+    # Arrange
+    task_group_metadata = test_case["task_group_metadata"]
+    data = test_case["data"]
+    expected_status = test_case["expected_status"]
+
+    # Act
+    task_results = local_exec._receive(task_group_metadata, data)
+
+    # Assert
+    for task_result in task_results:
+        assert task_result.status == expected_status
+
+
+@pytest.mark.asyncio
+async def test_receive(mocker):
+    """Test the receive function of LocalExecutor"""
+
+    local_exec = LocalExecutor()
+
+    # Arrange
+    task_group_metadata = {"dispatch_id": "1", "node_ids": ["1", "2"]}
+    test_data = {"status": StatusEnum.COMPLETED}
+
+    mock_loop = mocker.Mock()
+
+    mock_get_running_loop = mocker.patch(
+        "covalent.executor.executor_plugins.local.asyncio.get_running_loop",
+        return_value=mock_loop,
+    )
+    mock_get_running_loop.return_value.run_in_executor = mocker.AsyncMock()
+
+    await local_exec.receive(
+        task_group_metadata,
+        test_data,
+    )
+
+    mock_get_running_loop.assert_called_once()
+
+    mock_get_running_loop.return_value.run_in_executor.assert_awaited_once_with(
+        None,
+        local_exec._receive,
+        task_group_metadata,
+        test_data,
+    )
diff --git a/tests/covalent_tests/triggers/database_trigger_test.py b/tests/covalent_tests/triggers/database_trigger_test.py
index 660a78ab..734eb2e0 100644
--- a/tests/covalent_tests/triggers/database_trigger_test.py
+++ b/tests/covalent_tests/triggers/database_trigger_test.py
@@ -15,6 +15,7 @@
 # limitations under the License.
 
 import pytest
+import sqlalchemy
 
 from covalent.triggers.database_trigger import DatabaseTrigger
 
@@ -77,11 +78,14 @@ def test_database_trigger_observe(mocker, where_clauses, database_trigger):
     mock_db_engine.assert_called_once_with("test_db_path")
     mock_session.assert_called_once_with(mock_db_engine("test_db_path"))
     mock_event.assert_called_once()
-    mock_sql_execute = mocker.patch.object(mock_session, "execute", autospec=True)
+    mock_sql_execute = mock_session.return_value.__enter__.return_value.execute
     mock_sql_execute.assert_called_once_with(sql_poll_cmd)
     mock_sleep.assert_called_once_with(1)
 
 
+@pytest.mark.skip(
+    reason="Not sure what the purpose of this test is since no specific exception is raised or checked for."
+)
 @pytest.mark.parametrize(
     "where_clauses",
     [
@@ -116,21 +120,13 @@ def test_database_trigger_exception(mocker, where_clauses, database_trigger):
     mock_sleep.assert_called_once_with(1)
 
 
-@pytest.mark.parametrize(
-    "where_clauses",
-    [
-        ["id > 2", "status = FAILED"],
-        None,
-    ],
-)
-def test_database_trigger_exception_session(mocker, where_clauses, database_trigger):
+def test_database_trigger_exception_session(mocker, database_trigger):
     """
-    Test the observe method of Database trigger when an OperationalError is raised
+    Test the observe method of Database trigger when an ArgumentError is raised
     """
     database_trigger.trigger = mocker.MagicMock()
     mock_event = mocker.patch("covalent.triggers.database_trigger.Event")
     mock_event.return_value.is_set.side_effect = [False, True]
-    import sqlalchemy
 
     # Call the 'observer' method
     try:
diff --git a/tests/covalent_tests/workflow/lattice_serialization_test.py b/tests/covalent_tests/workflow/lattice_serialization_test.py
index 8cb83345..72d962d0 100644
--- a/tests/covalent_tests/workflow/lattice_serialization_test.py
+++ b/tests/covalent_tests/workflow/lattice_serialization_test.py
@@ -55,7 +55,7 @@ def test_lattice_json_serialization():
         return f(x)
 
     workflow.build_graph(5)
-    workflow.cova_imports = {"dummy_module"}
+    workflow.cova_imports = ["dummy_module"]
 
     json_workflow = workflow.serialize_to_json()
 
diff --git a/tests/covalent_ui_backend_tests/utils/data/lattices.json b/tests/covalent_ui_backend_tests/utils/data/lattices.json
index c166bad0..bbe1fa9d 100644
--- a/tests/covalent_ui_backend_tests/utils/data/lattices.json
+++ b/tests/covalent_ui_backend_tests/utils/data/lattices.json
@@ -4,7 +4,7 @@
     "call_before_filename": "call_before.pkl",
     "completed_at": "2022-09-23 10:01:11.717064",
     "completed_electron_num": 6,
-    "cova_imports_filename": "cova_imports.pkl",
+    "cova_imports_filename": "cova_imports.json",
     "created_at": "2022-09-23 10:01:11.044857",
     "deps_filename": "deps.pkl",
     "dispatch_id": "78525234-72ec-42dc-94a0-f4751707f9cd",
@@ -20,7 +20,7 @@
     "id": 1,
     "inputs_filename": "inputs.pkl",
     "is_active": 1,
-    "lattice_imports_filename": "lattice_imports.pkl",
+    "lattice_imports_filename": "lattice_imports.txt",
     "name": "workflow",
     "named_args_filename": "named_args.pkl",
     "named_kwargs_filename": "named_kwargs.pkl",
@@ -41,7 +41,7 @@
     "call_before_filename": "call_before.pkl",
     "completed_at": "2022-10-27 10:08:43.991108",
     "completed_electron_num": 4,
-    "cova_imports_filename": "cova_imports.pkl",
+    "cova_imports_filename": "cova_imports.json",
     "created_at": "2022-10-27 10:08:33.702548",
     "deps_filename": "deps.pkl",
     "dispatch_id": "a95d84ad-c441-446d-83ae-46380dcdf38e",
@@ -57,7 +57,7 @@
     "id": 2,
     "inputs_filename": "inputs.pkl",
     "is_active": 1,
-    "lattice_imports_filename": "lattice_imports.pkl",
+    "lattice_imports_filename": "lattice_imports.txt",
     "name": "workflow",
     "named_args_filename": "named_args.pkl",
     "named_kwargs_filename": "named_kwargs.pkl",
@@ -78,7 +78,7 @@
     "call_before_filename": "call_before.pkl",
     "completed_at": "2022-10-27 10:08:35.997225",
     "completed_electron_num": 20,
-    "cova_imports_filename": "cova_imports.pkl",
+    "cova_imports_filename": "cova_imports.json",
     "created_at": "2022-10-27 10:08:34.103977",
     "deps_filename": "deps.pkl",
     "dispatch_id": "89be0bcf-95dd-40a6-947e-6af6c56f147d",
@@ -94,7 +94,7 @@
     "id": 3,
     "inputs_filename": "inputs.pkl",
     "is_active": 1,
-    "lattice_imports_filename": "lattice_imports.pkl",
+    "lattice_imports_filename": "lattice_imports.txt",
     "name": "sub",
     "named_args_filename": "named_args.pkl",
     "named_kwargs_filename": "named_kwargs.pkl",
@@ -115,7 +115,7 @@
     "call_before_filename": "call_before.pkl",
     "completed_at": "2022-10-27 10:08:43.877056",
     "completed_electron_num": 120,
-    "cova_imports_filename": "cova_imports.pkl",
+    "cova_imports_filename": "cova_imports.json",
     "created_at": "2022-10-27 10:08:36.287047",
     "deps_filename": "deps.pkl",
     "dispatch_id": "69dec597-79d9-4c99-96de-8d5f06f3d4dd",
@@ -131,7 +131,7 @@
     "id": 4,
     "inputs_filename": "inputs.pkl",
     "is_active": 1,
-    "lattice_imports_filename": "lattice_imports.pkl",
+    "lattice_imports_filename": "lattice_imports.txt",
     "name": "sub",
     "named_args_filename": "named_args.pkl",
     "named_kwargs_filename": "named_kwargs.pkl",
@@ -152,7 +152,7 @@
     "call_before_filename": "call_before.pkl",
     "completed_at": "2023-08-10 10:08:55.902257",
     "completed_electron_num": 2,
-    "cova_imports_filename": "cova_imports.pkl",
+    "cova_imports_filename": "cova_imports.json",
     "created_at": "2023-08-10 10:08:55.387554",
     "deps_filename": "deps.pkl",
     "dispatch_id": "e8fd09c9-1406-4686-9e77-c8d4d64a76ee",
@@ -168,7 +168,7 @@
     "id": 5,
     "inputs_filename": "inputs.pkl",
     "is_active": 1,
-    "lattice_imports_filename": "lattice_imports.pkl",
+    "lattice_imports_filename": "lattice_imports.txt",
     "name": "workflow",
     "named_args_filename": "named_args.pkl",
     "named_kwargs_filename": "named_kwargs.pkl",
diff --git a/tests/covalent_ui_backend_tests/utils/data/mock_files.py b/tests/covalent_ui_backend_tests/utils/data/mock_files.py
index eb7bc817..613192fd 100644
--- a/tests/covalent_ui_backend_tests/utils/data/mock_files.py
+++ b/tests/covalent_ui_backend_tests/utils/data/mock_files.py
@@ -16,6 +16,7 @@
 
 """Mock files data"""
 
+import json
 import os
 import pickle
 
@@ -50,12 +51,12 @@ def mock_files_data():
         "function": "function.pkl",
         "executor": "executor_data.pkl",
         "deps": "deps.pkl",
-        "cova_imports": "cova_imports.pkl",
+        "cova_imports": "cova_imports.json",
         "error": "error.log",
         "function_docstring": "function_docstring.txt",
         "function_string": "function_string.txt",
         "inputs": "inputs.pkl",
-        "lattice_imports": "lattice_imports.pkl",
+        "lattice_imports": "lattice_imports.txt",
     }
 
     _object_id = "gAWVNwAAAAAAAACMG2NvdmFsZW50Ll93b3JrZmxvdy5kZXBzYmFzaJSME2FwcGx5X2Jhc2hfY29tbWFuZHOUk5Qu"  # pragma: allowlist secret
@@ -78,7 +79,7 @@ def mock_files_data():
             "files": [
                 {"file_name": file_name["call_after"], "data": []},
                 {"file_name": file_name["call_before"], "data": []},
-                {"file_name": file_name["cova_imports"], "data": {"electron", "ct"}},
+                {"file_name": file_name["cova_imports"], "data": json.dumps(["electron", "ct"])},
                 {"file_name": file_name["deps"], "data": {}},
                 {"file_name": file_name["error"], "data": ""},
                 {"file_name": file_name["executor"], "data": {}},