Repository: AgnostiqHQ/covalent
Pull request #1447 (2022-12-12 06:51:58)

Execution.py refactor into components
- [ ] I have added the tests to cover my changes.
- [ ] I have updated the documentation and CHANGELOG accordingly.
- [ ] I have read the CONTRIBUTING document.


github-actions[bot]: Hello. You may have forgotten to update the changelog!
Please edit [`CHANGELOG.md`](/AgnostiqHQ/covalent/blob/master/CHANGELOG.md) with a one-to-two sentence description of the change in the UNRELEASED section. You may include a small working example for new features.
codecov[bot]: # [Codecov](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1447?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) Report
> Merging [#1447](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1447?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (b0e27af) into [develop](https://codecov.io/gh/AgnostiqHQ/covalent/commit/3cf994648504f8124a395d83ed8472262c212885?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (3cf9946) will **increase** coverage by `0.74%`.
> The diff coverage is `99.78%`.

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


[![Impacted file tree graph](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1447/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://codecov.io/gh/AgnostiqHQ/covalent/pull/1447?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)

```diff
@@             Coverage Diff             @@
##           develop    #1447      +/-   ##
===========================================
+ Coverage    76.21%   76.96%   +0.74%     
===========================================
  Files          141      144       +3     
  Lines         5601     5700      +99     
  Branches       145      145              
===========================================
+ Hits          4269     4387     +118     
+ Misses        1210     1191      -19     
  Partials       122      122              
```

| Flag | Coverage Δ | |
|---|---|---|
| Dispatcher | `88.19% <100.00%> (+1.97%)` | :arrow_up: |
| Functional_Tests | `61.25% <49.57%> (+1.41%)` | :arrow_up: |
| SDK | `81.40% <90.47%> (+0.52%)` | :arrow_up: |
| UI_Frontend | `47.00% <ø> (ø)` | |

</details>

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index fa1c58b7..49ad231b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 - Updated `requirements.txt` to include `pyscopg2`
+- Refactored execution.py into loosely coupled modular pieces
 
 ### Fixed
 
diff --git a/covalent/_results_manager/result.py b/covalent/_results_manager/result.py
index 0aa32a12..0f8c065c 100644
--- a/covalent/_results_manager/result.py
+++ b/covalent/_results_manager/result.py
@@ -85,8 +85,11 @@ class Result:
         self._dispatch_id = dispatch_id
 
         self._root_dispatch_id = dispatch_id
+        self._electron_id = None
 
         self._status = Result.NEW_OBJ
+        self._task_failed = False
+        self._task_cancelled = False
 
         self._result = TransportableObject(None)
 
@@ -243,6 +246,8 @@ Node Outputs
 
             self.lattice.transport_graph.set_node_value(node_id, "error", None)
 
+            self.lattice.transport_graph.set_node_value(node_id, "sub_dispatch_id", None)
+
             self.lattice.transport_graph.set_node_value(node_id, "sublattice_result", None)
 
             self.lattice.transport_graph.set_node_value(node_id, "stdout", None)
@@ -332,8 +337,10 @@ Node Outputs
         with active_lattice_manager.claim(lattice):
             lattice.post_processing = True
             lattice.electron_outputs = ordered_node_outputs
+            args = [arg.get_deserialized() for arg in lattice.args]
+            kwargs = {k: v.get_deserialized() for k, v in lattice.kwargs.items()}
             workflow_function = lattice.workflow_function.get_deserialized()
-            result = workflow_function(*lattice.args, **lattice.kwargs)
+            result = workflow_function(*args, **kwargs)
             lattice.post_processing = False
         return result
 
@@ -405,6 +412,7 @@ Node Outputs
         status: "Status" = None,
         output: Any = None,
         error: Exception = None,
+        sub_dispatch_id: str = None,
         sublattice_result: "Result" = None,
         stdout: str = None,
         stderr: str = None,
@@ -449,6 +457,11 @@ Node Outputs
         if error is not None:
             self.lattice.transport_graph.set_node_value(node_id, "error", error)
 
+        if sub_dispatch_id is not None:
+            self.lattice.transport_graph.set_node_value(
+                node_id, "sub_dispatch_id", sub_dispatch_id
+            )
+
         if sublattice_result is not None:
             self.lattice.transport_graph.set_node_value(
                 node_id, "sublattice_result", sublattice_result
diff --git a/covalent/_workflow/electron.py b/covalent/_workflow/electron.py
index 34f33dc0..a14d8e5d 100644
--- a/covalent/_workflow/electron.py
+++ b/covalent/_workflow/electron.py
@@ -34,8 +34,6 @@ from .._shared_files.context_managers import active_lattice_manager
 from .._shared_files.defaults import (
     WAIT_EDGE_NAME,
     DefaultMetadataValues,
-    electron_dict_prefix,
-    electron_list_prefix,
     parameter_prefix,
     prefix_separator,
     sublattice_prefix,
@@ -396,6 +394,11 @@ class Electron:
             None
         """
 
+        collection_metadata = encode_metadata(DEFAULT_METADATA_VALUES.copy())
+        if "executor" in self.metadata:
+            collection_metadata["executor"] = self.metadata["executor"]
+            collection_metadata["executor_data"] = self.metadata["executor_data"]
+
         if isinstance(param_value, Electron):
             transport_graph.add_edge(
                 param_value.node_id,
@@ -406,15 +409,10 @@ class Electron:
             )
 
         elif isinstance(param_value, list):
-            list_node = self.add_collection_node_to_graph(transport_graph, electron_list_prefix)
-
-            for index, v in enumerate(param_value):
-                self.connect_node_with_others(
-                    list_node, param_name, v, "kwarg", index, transport_graph
-                )
-
+            list_electron = Electron(function=_auto_list_node, metadata=collection_metadata)
+            list_electron(*param_value)
             transport_graph.add_edge(
-                list_node,
+                list_electron.node_id,
                 node_id,
                 edge_name=param_name,
                 param_type=param_type,
@@ -422,13 +420,10 @@ class Electron:
             )
 
         elif isinstance(param_value, dict):
-            dict_node = self.add_collection_node_to_graph(transport_graph, electron_dict_prefix)
-
-            for k, v in param_value.items():
-                self.connect_node_with_others(dict_node, k, v, "kwarg", None, transport_graph)
-
+            dict_electron = Electron(function=_auto_dict_node, metadata=collection_metadata)
+            dict_electron(**param_value)
             transport_graph.add_edge(
-                dict_node,
+                dict_electron.node_id,
                 node_id,
                 edge_name=param_name,
                 param_type=param_type,
@@ -655,3 +650,11 @@ def to_decoded_electron_collection(**x):
         return TransportableObject.deserialize_list(collection)
     elif isinstance(collection, dict):
         return TransportableObject.deserialize_dict(collection)
+
+
+def _auto_list_node(*args, **kwargs):
+    return list(args)
+
+
+def _auto_dict_node(*args, **kwargs):
+    return dict(kwargs)
diff --git a/covalent_dispatcher/_core/__init__.py b/covalent_dispatcher/_core/__init__.py
index cb43508c..810c5a49 100644
--- a/covalent_dispatcher/_core/__init__.py
+++ b/covalent_dispatcher/_core/__init__.py
@@ -18,4 +18,5 @@
 #
 # Relief from the License may be granted by purchasing a commercial license.
 
-from .execution import cancel_workflow, initialize_result_object, run_workflow
+from .data_manager import make_dispatch
+from .dispatcher import cancel_workflow, run_dispatch
diff --git a/covalent_dispatcher/_core/data_manager.py b/covalent_dispatcher/_core/data_manager.py
new file mode 100644
index 00000000..b9261e17
--- /dev/null
+++ b/covalent_dispatcher/_core/data_manager.py
@@ -0,0 +1,195 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""
+Defines the core functionality of the result service
+"""
+
+import asyncio
+import uuid
+
+from covalent._results_manager import Result
+from covalent._shared_files import logger
+from covalent._workflow.lattice import Lattice
+
+from .._db import update, upsert
+from .._db.write_result_to_db import resolve_electron_id
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+
+# References to result objects of live dispatches
+_registered_dispatches = {}
+
+# Map of dispatch_id -> message_queue for pushing node status updates
+# to dispatcher
+_dispatch_status_queues = {}
+
+
+def generate_node_result(
+    node_id,
+    start_time=None,
+    end_time=None,
+    status=None,
+    output=None,
+    error=None,
+    stdout=None,
+    stderr=None,
+    sub_dispatch_id=None,
+    sublattice_result=None,
+):
+
+    return {
+        "node_id": node_id,
+        "start_time": start_time,
+        "end_time": end_time,
+        "status": status,
+        "output": output,
+        "error": error,
+        "stdout": stdout,
+        "stderr": stderr,
+        "sub_dispatch_id": sub_dispatch_id,
+        "sublattice_result": sublattice_result,
+    }
+
+
+# Domain: result
+async def update_node_result(result_object, node_result):
+    app_log.warning("Updating node result (run_planned_workflow).")
+    try:
+        update._node(result_object, **node_result)
+    except Exception as ex:
+        app_log.exception("Error persisting node update: {ex}")
+        node_result["status"] = Result.FAILED
+    finally:
+        node_id = node_result["node_id"]
+        node_status = node_result["status"]
+        dispatch_id = result_object.dispatch_id
+        if node_status:
+            status_queue = get_status_queue(dispatch_id)
+            await status_queue.put((node_id, node_status))
+
+
+# Domain: result
+def initialize_result_object(
+    json_lattice: str, parent_result_object: Result = None, parent_electron_id: int = None
+) -> Result:
+    """Convenience function for constructing a result object from a json-serialized lattice.
+
+    Args:
+        json_lattice: a JSON-serialized lattice
+        parent_result_object: the parent result object if json_lattice is a sublattice
+        parent_electron_id: the DB id of the parent electron (for sublattices)
+
+    Returns:
+        Result: result object
+    """
+
+    dispatch_id = get_unique_id()
+    lattice = Lattice.deserialize_from_json(json_lattice)
+    result_object = Result(lattice, dispatch_id)
+    if parent_result_object:
+        result_object._root_dispatch_id = parent_result_object._root_dispatch_id
+
+    result_object._electron_id = parent_electron_id
+    result_object._initialize_nodes()
+    app_log.debug("2: Constructed result object and initialized nodes.")
+
+    update.persist(result_object, electron_id=parent_electron_id)
+    app_log.debug("Result object persisted.")
+
+    return result_object
+
+
+# Domain: result
+def get_unique_id() -> str:
+    """
+    Get a unique ID.
+
+    Args:
+        None
+
+    Returns:
+        str: Unique ID
+    """
+
+    return str(uuid.uuid4())
+
+
+def make_dispatch(
+    json_lattice: str, parent_result_object: Result = None, parent_electron_id: int = None
+) -> Result:
+
+    result_object = initialize_result_object(
+        json_lattice, parent_result_object, parent_electron_id
+    )
+    _register_result_object(result_object)
+    return result_object.dispatch_id
+
+
+def get_result_object(dispatch_id: str) -> Result:
+    return _registered_dispatches[dispatch_id]
+
+
+def _register_result_object(result_object: Result):
+    dispatch_id = result_object.dispatch_id
+    _registered_dispatches[dispatch_id] = result_object
+    _dispatch_status_queues[dispatch_id] = asyncio.Queue()
+
+
+def finalize_dispatch(dispatch_id: str):
+    del _dispatch_status_queues[dispatch_id]
+    del _registered_dispatches[dispatch_id]
+
+
+def get_status_queue(dispatch_id: str):
+    return _dispatch_status_queues[dispatch_id]
+
+
+async def persist_result(dispatch_id: str):
+    result_object = get_result_object(dispatch_id)
+    update.persist(result_object)
+    await _update_parent_electron(result_object)
+
+
+async def _update_parent_electron(result_object: Result):
+    parent_eid = result_object._electron_id
+
+    if parent_eid:
+        dispatch_id, node_id = resolve_electron_id(parent_eid)
+        status = result_object.status
+        if status == Result.POSTPROCESSING_FAILED:
+            status = Result.FAILED
+        node_result = generate_node_result(
+            node_id=node_id,
+            end_time=result_object.end_time,
+            status=status,
+            output=result_object._result,
+            error=result_object._error,
+            sublattice_result=result_object,
+        )
+        parent_result_obj = get_result_object(dispatch_id)
+        app_log.debug(f"Updating sublattice parent node {dispatch_id}:{node_id}")
+        await update_node_result(parent_result_obj, node_result)
+
+
+def upsert_lattice_data(dispatch_id: str):
+    result_object = get_result_object(dispatch_id)
+    upsert._lattice_data(result_object)
diff --git a/covalent_dispatcher/_core/dispatcher.py b/covalent_dispatcher/_core/dispatcher.py
new file mode 100644
index 00000000..ac19b319
--- /dev/null
+++ b/covalent_dispatcher/_core/dispatcher.py
@@ -0,0 +1,364 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""
+Defines the core functionality of the dispatcher
+"""
+
+import asyncio
+import traceback
+from datetime import datetime, timezone
+from typing import Dict, Tuple
+
+from covalent._results_manager import Result
+from covalent._shared_files import logger
+from covalent._shared_files.defaults import parameter_prefix
+from covalent_ui import result_webhook
+
+from . import data_manager as datasvc
+from . import runner
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+
+
+# Domain: dispatcher
+def _get_abstract_task_inputs(node_id: int, node_name: str, result_object: Result) -> dict:
+    """Return placeholders for the required inputs for a task execution.
+
+    Args:
+        node_id: Node id of this task in the transport graph.
+        node_name: Name of the node.
+        result_object: Result object to be used to update and store execution related
+                       info including the results.
+
+    Returns: inputs: Input dictionary to be passed to the task with
+        `node_id` placeholders for args, kwargs. These are to be
+        resolved to their values later.
+    """
+
+    abstract_task_input = {"args": [], "kwargs": {}}
+
+    for parent in result_object.lattice.transport_graph.get_dependencies(node_id):
+
+        edge_data = result_object.lattice.transport_graph.get_edge_data(parent, node_id)
+        # value = result_object.lattice.transport_graph.get_node_value(parent, "output")
+
+        for e_key, d in edge_data.items():
+            if not d.get("wait_for"):
+                if d["param_type"] == "arg":
+                    abstract_task_input["args"].append((parent, d["arg_index"]))
+                elif d["param_type"] == "kwarg":
+                    key = d["edge_name"]
+                    abstract_task_input["kwargs"][key] = parent
+
+    sorted_args = sorted(abstract_task_input["args"], key=lambda x: x[1])
+    abstract_task_input["args"] = [x[0] for x in sorted_args]
+
+    return abstract_task_input
+
+
+# Domain: dispatcher
+async def _handle_completed_node(result_object, node_id, pending_parents):
+    g = result_object.lattice.transport_graph._graph
+
+    ready_nodes = []
+    app_log.debug(f"Node {node_id} completed")
+    for child, edges in g.adj[node_id].items():
+        for edge in edges:
+            pending_parents[child] -= 1
+        if pending_parents[child] < 1:
+            app_log.debug(f"Queuing node {child} for execution")
+            ready_nodes.append(child)
+
+    return ready_nodes
+
+
+# Domain: dispatcher
+async def _handle_failed_node(result_object, node_id):
+    result_object._task_failed = True
+    result_object._end_time = datetime.now(timezone.utc)
+    app_log.debug(f"Node {result_object.dispatch_id}:{node_id} failed")
+    app_log.debug("8A: Failed node upsert statement (run_planned_workflow)")
+    datasvc.upsert_lattice_data(result_object.dispatch_id)
+    await result_webhook.send_update(result_object)
+
+
+# Domain: dispatcher
+async def _handle_cancelled_node(result_object, node_id):
+    result_object._task_cancelled = True
+    result_object._end_time = datetime.now(timezone.utc)
+    app_log.debug(f"Node {result_object.dispatch_id}:{node_id} cancelled")
+    app_log.debug("9: Cancelled node upsert statement (run_planned_workflow)")
+    datasvc.upsert_lattice_data(result_object.dispatch_id)
+    await result_webhook.send_update(result_object)
+
+
+# Domain: dispatcher
+async def _get_initial_tasks_and_deps(result_object: Result) -> Tuple[int, int, Dict]:
+    """Compute the initial batch of tasks to submit and initialize each task's dep count
+
+    Returns: (num_tasks, ready_nodes, pending_parents) where num_tasks is
+        the total number of tasks in the graph, ready_nodes is the
+        initial list of tasks to dispatch, and pending_parents is a map
+        from `node_id` to the number of parents that have yet to
+        complete.
+
+    """
+
+    num_tasks = 0
+    ready_nodes = []
+    pending_parents = {}
+
+    g = result_object.lattice.transport_graph._graph
+    for node_id, d in g.in_degree():
+        app_log.debug(f"Node {node_id} has {d} parents")
+
+        pending_parents[node_id] = d
+        num_tasks += 1
+        if d == 0:
+            ready_nodes.append(node_id)
+
+    return num_tasks, ready_nodes, pending_parents
+
+
+# Domain: dispatcher
+async def _submit_task(result_object, node_id):
+
+    # Get name of the node for the current task
+    node_name = result_object.lattice.transport_graph.get_node_value(node_id, "name")
+    app_log.debug(f"7A: Node name: {node_name} (run_planned_workflow).")
+
+    # Handle parameter nodes
+    if node_name.startswith(parameter_prefix):
+        app_log.debug("7C: Parameter if block (run_planned_workflow).")
+        output = result_object.lattice.transport_graph.get_node_value(node_id, "value")
+        app_log.debug(f"7C: Node output: {output} (run_planned_workflow).")
+        app_log.debug("8: Starting update node (run_planned_workflow).")
+
+        node_result = {
+            "node_id": node_id,
+            "start_time": datetime.now(timezone.utc),
+            "end_time": datetime.now(timezone.utc),
+            "status": Result.COMPLETED,
+            "output": output,
+        }
+        await datasvc.update_node_result(result_object, node_result)
+        app_log.debug("8A: Update node success (run_planned_workflow).")
+
+    else:
+
+        # Executor for post_processing and dispatching sublattices
+        pp_executor = result_object.lattice.get_metadata("workflow_executor")
+        pp_executor_data = result_object.lattice.get_metadata("workflow_executor_data")
+        post_processor = [pp_executor, pp_executor_data]
+
+        # Gather inputs and dispatch task
+        app_log.debug(f"Gathering inputs for task {node_id} (run_planned_workflow).")
+
+        abs_task_input = _get_abstract_task_inputs(node_id, node_name, result_object)
+
+        selected_executor = result_object.lattice.transport_graph.get_node_value(
+            node_id, "metadata"
+        )["executor"]
+
+        selected_executor_data = result_object.lattice.transport_graph.get_node_value(
+            node_id, "metadata"
+        )["executor_data"]
+
+        app_log.debug(f"Submitting task {node_id} to executor")
+
+        coro = runner.run_abstract_task(
+            dispatch_id=result_object.dispatch_id,
+            node_id=node_id,
+            selected_executor=[selected_executor, selected_executor_data],
+            node_name=node_name,
+            abstract_inputs=abs_task_input,
+            workflow_executor=post_processor,
+        )
+
+        asyncio.create_task(coro)
+
+
+# Domain: dispatcher
+async def _run_planned_workflow(result_object: Result, status_queue: asyncio.Queue) -> Result:
+    """
+    Run the workflow in the topological order of their position on the
+    transport graph. Does this in an asynchronous manner so that nodes
+    at the same level are executed in parallel. Also updates the status
+    of the whole workflow execution.
+
+    Args:
+        result_object: Result object being used for current dispatch
+        status_queue: message queue for notifying the main loop of status updates
+
+    Returns:
+        None
+    """
+
+    app_log.debug("3: Inside run_planned_workflow (run_planned_workflow).")
+    result_object._status = Result.RUNNING
+    result_object._start_time = datetime.now(timezone.utc)
+
+    app_log.debug(
+        f"4: Workflow status changed to running {result_object.dispatch_id} (run_planned_workflow)."
+    )
+    datasvc.upsert_lattice_data(result_object.dispatch_id)
+    app_log.debug("5: Wrote lattice status to DB (run_planned_workflow).")
+
+    tasks_left, initial_nodes, pending_parents = await _get_initial_tasks_and_deps(result_object)
+
+    unresolved_tasks = 0
+    resolved_tasks = 0
+
+    for node_id in initial_nodes:
+        unresolved_tasks += 1
+        await _submit_task(result_object, node_id)
+
+    while unresolved_tasks > 0:
+        app_log.debug(f"{tasks_left} tasks left to complete")
+        app_log.debug(f"Waiting to hear from {unresolved_tasks} tasks")
+        node_id, node_status = await status_queue.get()
+
+        app_log.debug(f"Received node status update {node_id}: {node_status}")
+
+        if node_status == Result.RUNNING:
+            continue
+
+        unresolved_tasks -= 1
+
+        if node_status == Result.COMPLETED:
+            tasks_left -= 1
+            ready_nodes = await _handle_completed_node(result_object, node_id, pending_parents)
+            for node_id in ready_nodes:
+                unresolved_tasks += 1
+                await _submit_task(result_object, node_id)
+
+        if node_status == Result.FAILED:
+            await _handle_failed_node(result_object, node_id)
+            continue
+
+        if node_status == Result.CANCELLED:
+            await _handle_cancelled_node(result_object, node_id)
+            continue
+
+    if result_object._task_failed or result_object._task_cancelled:
+        app_log.debug(f"Workflow {result_object.dispatch_id} cancelled or failed")
+        failed_nodes = result_object._get_failed_nodes()
+        failed_nodes = map(lambda x: f"{x[0]}: {x[1]}", failed_nodes)
+        failed_nodes_msg = "\n".join(failed_nodes)
+        result_object._error = "The following tasks failed:\n" + failed_nodes_msg
+        result_object._status = Result.FAILED if result_object._task_failed else Result.CANCELLED
+        return result_object
+
+    app_log.debug("8: All tasks finished running (run_planned_workflow)")
+
+    result_object = await runner.postprocess_workflow(result_object.dispatch_id)
+
+    app_log.debug(f"Status after PP: {result_object._status}")
+    await result_webhook.send_update(result_object)
+
+    return result_object
+
+
+def _plan_workflow(result_object: Result) -> None:
+    """
+    Function to plan a workflow according to a schedule.
+    Planning means to decide which executors (along with their arguments) will
+    be used by each node.
+
+    Args:
+        result_object: Result object being used for current dispatch
+
+    Returns:
+        None
+    """
+
+    if result_object.lattice.get_metadata("schedule"):
+        # Custom scheduling logic of the format:
+        # scheduled_executors = get_schedule(result_object)
+
+        # for node_id, executor in scheduled_executors.items():
+        #    result_object.lattice.transport_graph.set_node_value(node_id, "executor", executor)
+        pass
+
+
+async def run_workflow(result_object: Result) -> Result:
+    """
+    Plan and run the workflow by loading the result object corresponding to the
+    dispatch id and retrieving essential information from it.
+    Returns without changing anything if a redispatch is done of a (partially or fully)
+    completed workflow with the same dispatch id.
+
+    Args:
+        dispatch_id: Dispatch id of the workflow to be run
+        results_dir: Directory where the result object is stored
+
+    Returns:
+        The result object from the workflow execution
+    """
+
+    app_log.debug("Inside run_workflow.")
+
+    if result_object.status == Result.COMPLETED:
+        datasvc.finalize_dispatch(result_object.dispatch_id)
+        return result_object
+
+    try:
+        _plan_workflow(result_object)
+        status_queue = datasvc.get_status_queue(result_object.dispatch_id)
+        result_object = await _run_planned_workflow(result_object, status_queue)
+
+    except Exception as ex:
+        app_log.error(f"Exception during _run_planned_workflow: {ex}")
+
+        error_msg = "".join(traceback.TracebackException.from_exception(ex).format())
+        result_object._status = Result.FAILED
+        result_object._error = error_msg
+        result_object._end_time = datetime.now(timezone.utc)
+
+    finally:
+        await datasvc.persist_result(result_object.dispatch_id)
+        datasvc.finalize_dispatch(result_object.dispatch_id)
+
+    return result_object
+
+
+# Domain: dispatcher
+def cancel_workflow(dispatch_id: str) -> None:
+    """
+    Cancels a dispatched workflow using publish subscribe mechanism
+    provided by Dask.
+
+    Args:
+        dispatch_id: Dispatch id of the workflow to be cancelled
+
+    Returns:
+        None
+    """
+
+    # shared_var = Variable(dispatch_id)
+    # shared_var.set(str(Result.CANCELLED))
+    pass
+
+
+def run_dispatch(dispatch_id: str) -> asyncio.Future:
+    result_object = datasvc.get_result_object(dispatch_id)
+    return asyncio.create_task(run_workflow(result_object))
diff --git a/covalent_dispatcher/_core/execution.py b/covalent_dispatcher/_core/execution.py
index 92c952d9..d498f1e6 100644
--- a/covalent_dispatcher/_core/execution.py
+++ b/covalent_dispatcher/_core/execution.py
@@ -22,79 +22,41 @@
 Defines the core functionality of the dispatcher
 """
 
-import asyncio
-import json
-import os
-import traceback
-import uuid
-from asyncio import Queue
-from datetime import datetime, timezone
-from functools import partial
-from typing import Any, Dict, List, Tuple
+# Legacy imports
+# from .dispatcher import (
+#     _build_sublattice_graph,
+#     _dispatch_sync_sublattice,
+#     _get_abstract_task_inputs,
+#     _handle_cancelled_node,
+#     _handle_completed_node,
+#     _handle_failed_node,
+#     _initialize_deps_and_queue,
+#     _plan_workflow,
+#     _post_process,
+#     _postprocess_workflow,
+#     _run_planned_workflow,
+#     _submit_task,
+#     cancel_workflow,
+#     run_workflow,
+# )
+# from .result import (
+#     _update_node_result,
+#     generate_node_result,
+#     get_unique_id,
+#     initialize_result_object,
+# )
+# from .runner import (
+#     _gather_deps,
+#     _get_task_input_values,
+#     _get_task_inputs,
+#     _run_abstract_task,
+#     _run_task,
+#     _run_task_and_update,
+# )
 
 from covalent._results_manager import Result
-from covalent._shared_files import logger
-from covalent._shared_files.config import get_config
-from covalent._shared_files.context_managers import active_lattice_manager
-from covalent._shared_files.defaults import (
-    electron_dict_prefix,
-    electron_list_prefix,
-    parameter_prefix,
-    prefix_separator,
-    sublattice_prefix,
-)
-from covalent._workflow import DepsBash, DepsCall, DepsPip
-from covalent._workflow.lattice import Lattice
-from covalent._workflow.transport import TransportableObject
-from covalent.executor import _executor_manager
-from covalent.executor.base import wrapper_fn
-from covalent_ui import result_webhook
 
-from .._db import update, upsert
-from .._db.write_result_to_db import (
-    get_sublattice_electron_id,
-    update_lattices_data,
-    write_lattice_error,
-)
-
-app_log = logger.app_log
-log_stack_info = logger.log_stack_info
-debug_mode = get_config("sdk.log_level") == "debug"
-
-
-# This is to be run out-of-process
-def _build_sublattice_graph(sub: Lattice, parent_metadata: Dict, *args, **kwargs):
-    for k in sub.metadata.keys():
-        if k != "workflow_executor" and k != "workflow_executor_data" and not sub.metadata[k]:
-            sub.metadata[k] = parent_metadata[k]
-
-    sub.build_graph(*args, **kwargs)
-    return sub.serialize_to_json()
-
-
-def generate_node_result(
-    node_id,
-    start_time=None,
-    end_time=None,
-    status=None,
-    output=None,
-    error=None,
-    stdout=None,
-    stderr=None,
-    sublattice_result=None,
-):
-
-    return {
-        "node_id": node_id,
-        "start_time": start_time,
-        "end_time": end_time,
-        "status": status,
-        "output": output,
-        "error": error,
-        "stdout": stdout,
-        "stderr": stderr,
-        "sublattice_result": sublattice_result,
-    }
+from . import dispatcher, runner
 
 
 def _get_task_inputs(node_id: int, node_name: str, result_object: Result) -> dict:
@@ -114,779 +76,13 @@ def _get_task_inputs(node_id: int, node_name: str, result_object: Result) -> dic
                 and any parent node execution results if present.
     """
 
-    if node_name.startswith(electron_list_prefix):
-        values = [
-            result_object.lattice.transport_graph.get_node_value(parent, "output")
-            for parent in result_object.lattice.transport_graph.get_dependencies(node_id)
-        ]
-        task_input = {"args": [], "kwargs": {"x": TransportableObject.make_transportable(values)}}
-    elif node_name.startswith(electron_dict_prefix):
-        values = {}
-        for parent in result_object.lattice.transport_graph.get_dependencies(node_id):
-
-            edge_data = result_object.lattice.transport_graph.get_edge_data(parent, node_id)
-
-            value = result_object.lattice.transport_graph.get_node_value(parent, "output")
-            for e_key, d in edge_data.items():
-                key = d["edge_name"]
-                values[key] = value
-
-        task_input = {"args": [], "kwargs": {"x": TransportableObject.make_transportable(values)}}
-    else:
-        task_input = {"args": [], "kwargs": {}}
-
-        for parent in result_object.lattice.transport_graph.get_dependencies(node_id):
-
-            edge_data = result_object.lattice.transport_graph.get_edge_data(parent, node_id)
-            value = result_object.lattice.transport_graph.get_node_value(parent, "output")
-
-            for e_key, d in edge_data.items():
-                if not d.get("wait_for"):
-                    if d["param_type"] == "arg":
-                        task_input["args"].append((value, d["arg_index"]))
-                    elif d["param_type"] == "kwarg":
-                        key = d["edge_name"]
-                        task_input["kwargs"][key] = value
+    abstract_inputs = dispatcher._get_abstract_task_inputs(node_id, node_name, result_object)
+    input_values = runner._get_task_input_values(result_object, abstract_inputs)
 
-        sorted_args = sorted(task_input["args"], key=lambda x: x[1])
-        task_input["args"] = [x[0] for x in sorted_args]
+    abstract_args = abstract_inputs["args"]
+    abstract_kwargs = abstract_inputs["kwargs"]
+    args = [input_values[node_id] for node_id in abstract_args]
+    kwargs = {k: input_values[v] for k, v in abstract_kwargs.items()}
+    task_input = {"args": args, "kwargs": kwargs}
 
     return task_input
-
-
-# This is to be run out-of-process
-def _post_process(lattice: Lattice, node_outputs: Dict) -> Any:
-    """
-    Post processing function to be called after the lattice execution.
-    This takes care of executing statements that were not an electron
-    but were inside the lattice's function. It also replaces any calls
-    to an electron with the result of that electron execution, hence
-    preventing a local execution of electron's function.
-
-    Note: Here `node_outputs` is used instead of `electron_outputs`
-    since an electron can be called multiple times with possibly different
-    arguments, but every time it's called, it will be executed as a separate node.
-    Thus, output of every node is used.
-
-    Args:
-        lattice: Lattice object that was dispatched.
-        node_outputs: Dictionary containing the output of all the nodes.
-        execution_order: List of lists containing the order of execution of the nodes.
-
-    Reurns:
-        result: The result of the lattice function.
-    """
-
-    ordered_node_outputs = []
-    app_log.debug(f"node_outputs: {node_outputs}")
-    app_log.debug(f"node_outputs: {node_outputs.items()}")
-    for i, item in enumerate(node_outputs.items()):
-        key, val = item
-        app_log.debug(f"Here's the key: {key}")
-        if not key.startswith(prefix_separator) or key.startswith(sublattice_prefix):
-            ordered_node_outputs.append((i, val))
-
-    with active_lattice_manager.claim(lattice):
-        lattice.post_processing = True
-        lattice.electron_outputs = ordered_node_outputs
-        args = [arg.get_deserialized() for arg in lattice.args]
-        kwargs = {k: v.get_deserialized() for k, v in lattice.kwargs.items()}
-        workflow_function = lattice.workflow_function.get_deserialized()
-        result = workflow_function(*args, **kwargs)
-        lattice.post_processing = False
-        return result
-
-
-async def _dispatch_sync_sublattice(
-    parent_result_object: Result,
-    parent_electron_id: int,
-    inputs: Dict,
-    serialized_callable: Any,
-    workflow_executor: Any,
-) -> str:
-    """Dispatch a sublattice using the workflow_executor."""
-
-    app_log.debug("Inside _dispatch_sync_sublattice")
-
-    try:
-        short_name, object_dict = workflow_executor
-
-        if short_name == "client":
-            app_log.error("No executor selected for dispatching sublattices")
-            raise RuntimeError("No executor selected for dispatching sublattices")
-
-    except Exception as ex:
-        app_log.debug(f"Exception when trying to determine sublattice executor: {ex}")
-        return None
-
-    parent_metadata = TransportableObject.make_transportable(parent_result_object.lattice.metadata)
-    sub_dispatch_inputs = {
-        "args": [serialized_callable, parent_metadata],
-        "kwargs": inputs["kwargs"],
-    }
-    for arg in inputs["args"]:
-        sub_dispatch_inputs["args"].append(arg)
-
-    # Build the sublattice graph. This must be run
-    # externally since it involves deserializing the
-    # sublattice workflow function.
-    fut = asyncio.create_task(
-        _run_task(
-            result_object=parent_result_object,
-            node_id=-1,
-            serialized_callable=TransportableObject.make_transportable(_build_sublattice_graph),
-            selected_executor=workflow_executor,
-            node_name="build_sublattice_graph",
-            call_before=[],
-            call_after=[],
-            inputs=sub_dispatch_inputs,
-            workflow_executor=workflow_executor,
-        )
-    )
-
-    res = await fut
-    if res["status"] == Result.COMPLETED:
-        json_sublattice = json.loads(res["output"].json)
-
-        sub_result_object = initialize_result_object(
-            json_sublattice, parent_result_object, parent_electron_id
-        )
-        app_log.debug(f"Sublattice dispatch id: {sub_result_object.dispatch_id}")
-
-        return await run_workflow(sub_result_object)
-    else:
-        return None
-
-
-async def _run_task(
-    result_object: Result,
-    node_id: int,
-    inputs: Dict,
-    serialized_callable: Any,
-    selected_executor: Any,
-    call_before: List,
-    call_after: List,
-    node_name: str,
-    workflow_executor: Any,
-) -> None:
-    """
-    Run a task with given inputs on the selected executor.
-    Also updates the status of current node execution while
-    checking if a redispatch has occurred. Exclude those nodes
-    from execution which were completed.
-
-    Also verifies if execution of this dispatch has been cancelled.
-
-    Args:
-        inputs: Inputs for the task.
-        result_object: Result object being used for current dispatch
-        node_id: Node id of the task to be executed.
-
-    Returns:
-        None
-    """
-
-    dispatch_id = result_object.dispatch_id
-    results_dir = os.environ.get("COVALENT_DATA_DIR") or get_config("dispatcher.results_dir")
-
-    # Instantiate the executor from JSON
-    try:
-        short_name, object_dict = selected_executor
-
-        app_log.debug(f"Running task {node_name} using executor {short_name}, {object_dict}")
-
-        # the executor is determined during scheduling and provided in the execution metadata
-        executor = _executor_manager.get_executor(short_name)
-        executor.from_dict(object_dict)
-    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)
-        node_result = generate_node_result(
-            node_id=node_id,
-            end_time=datetime.now(timezone.utc),
-            status=Result.FAILED,
-            error=error_msg,
-        )
-        return node_result
-
-    # run the task on the executor and register any failures
-    try:
-
-        if node_name.startswith(sublattice_prefix):
-            sub_electron_id = get_sublattice_electron_id(
-                parent_dispatch_id=dispatch_id, sublattice_node_id=node_id
-            )
-
-            # Read the result object directly from the server
-
-            sublattice_result = await _dispatch_sync_sublattice(
-                parent_result_object=result_object,
-                parent_electron_id=sub_electron_id,
-                inputs=inputs,
-                serialized_callable=serialized_callable,
-                workflow_executor=workflow_executor,
-            )
-
-            if not sublattice_result:
-                raise RuntimeError("Sublattice execution failed")
-
-            output = sublattice_result.encoded_result
-            end_time = datetime.now(timezone.utc)
-            node_result = generate_node_result(
-                node_id=node_id,
-                end_time=end_time,
-                status=Result.COMPLETED,
-                output=output,
-                sublattice_result=sublattice_result,
-            )
-
-            app_log.debug("Sublattice dispatched (run_task)")
-            # Don't continue unless sublattice finishes
-            if sublattice_result.status != Result.COMPLETED:
-                node_result["status"] = Result.FAILED
-                node_result["error"] = "Sublattice workflow failed to complete"
-
-                upsert._lattice_data(sublattice_result)
-
-        else:
-            app_log.debug(f"Executing task {node_name}")
-            assembled_callable = partial(wrapper_fn, serialized_callable, call_before, call_after)
-            execute_callable = partial(
-                executor.execute,
-                function=assembled_callable,
-                args=inputs["args"],
-                kwargs=inputs["kwargs"],
-                dispatch_id=dispatch_id,
-                results_dir=results_dir,
-                node_id=node_id,
-            )
-            output, stdout, stderr, exception_raised = await executor._execute(
-                function=assembled_callable,
-                args=inputs["args"],
-                kwargs=inputs["kwargs"],
-                dispatch_id=dispatch_id,
-                results_dir=results_dir,
-                node_id=node_id,
-            )
-            if exception_raised:
-                status = Result.FAILED
-            else:
-                status = Result.COMPLETED
-
-            node_result = generate_node_result(
-                node_id=node_id,
-                end_time=datetime.now(timezone.utc),
-                status=status,
-                output=output,
-                stdout=stdout,
-                stderr=stderr,
-            )
-
-    except Exception as ex:
-        tb = "".join(traceback.TracebackException.from_exception(ex).format())
-        app_log.debug(f"Exception occurred when running task {node_id}:")
-        app_log.debug(tb)
-        error_msg = tb if debug_mode else str(ex)
-        node_result = generate_node_result(
-            node_id=node_id,
-            end_time=datetime.now(timezone.utc),
-            status=Result.FAILED,
-            error=error_msg,
-        )
-    app_log.debug(f"Node result: {node_result}")
-    return node_result
-
-
-def _gather_deps(result_object: Result, node_id: int) -> Tuple[List, List]:
-    """Assemble deps for a node into the final call_before and call_after"""
-
-    deps = result_object.lattice.transport_graph.get_node_value(node_id, "metadata")["deps"]
-
-    # Assemble call_before and call_after from all the deps
-
-    call_before_objs_json = result_object.lattice.transport_graph.get_node_value(
-        node_id, "metadata"
-    )["call_before"]
-    call_after_objs_json = result_object.lattice.transport_graph.get_node_value(
-        node_id, "metadata"
-    )["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
-
-
-async def _handle_completed_node(result_object, node_result, pending_deps, tasks_queue):
-    g = result_object.lattice.transport_graph._graph
-
-    for child, edges in g.adj[node_result["node_id"]].items():
-        for edge in edges:
-            pending_deps[child] -= 1
-        if pending_deps[child] < 1:
-            app_log.debug(f"Queuing node {child} for execution")
-            await tasks_queue.put(child)
-
-
-async def _handle_failed_node(result_object, node_result, pending_deps, tasks_queue):
-    node_id = node_result["node_id"]
-    result_object._status = Result.FAILED
-    result_object._end_time = datetime.now(timezone.utc)
-    app_log.debug("8A: Failed node upsert statement (run_planned_workflow)")
-    upsert._lattice_data(result_object)
-    await result_webhook.send_update(result_object)
-    await tasks_queue.put(-1)
-
-
-async def _handle_cancelled_node(result_object, node_result, pending_deps, tasks_queue):
-    result_object._status = Result.CANCELLED
-    result_object._end_time = datetime.now(timezone.utc)
-    app_log.debug("9: Failed node upsert statement (run_planned_workflow)")
-    upsert._lattice_data(result_object)
-    await result_webhook.send_update(result_object)
-    await tasks_queue.put(-1)
-
-
-async def _update_node_result(result_object, node_result, pending_deps, tasks_queue):
-    app_log.debug("Updating node result (run_planned_workflow).")
-    update._node(result_object, **node_result)
-    await result_webhook.send_update(result_object)
-
-    node_status = node_result["status"]
-    if node_status == Result.COMPLETED:
-        await _handle_completed_node(result_object, node_result, pending_deps, tasks_queue)
-        return
-
-    if node_status == Result.FAILED:
-        await _handle_failed_node(result_object, node_result, pending_deps, tasks_queue)
-        return
-
-    if node_status == Result.CANCELLED:
-        await _handle_cancelled_node(result_object, node_result, pending_deps, tasks_queue)
-        return
-
-    if node_status == Result.RUNNING:
-        return
-
-
-async def _run_task_and_update(run_task_callable, result_object, pending_deps, tasks_queue):
-    node_result = await run_task_callable()
-
-    # NOTE: This is a blocking operation because of db writes and needs special handling when
-    # we switch to an event loop for processing tasks
-    await _update_node_result(result_object, node_result, pending_deps, tasks_queue)
-    return node_result
-
-
-async def _initialize_deps_and_queue(
-    result_object: Result, tasks_queue: Queue, pending_deps: dict
-) -> int:
-    """Initialize the data structures controlling when tasks are queued for execution.
-
-    Returns the total number of nodes in the transport graph."""
-
-    num_tasks = 0
-    g = result_object.lattice.transport_graph._graph
-    for node_id, d in g.in_degree():
-        app_log.debug(f"Node {node_id} has {d} parents")
-
-        pending_deps[node_id] = d
-        num_tasks += 1
-        if d == 0:
-            await tasks_queue.put(node_id)
-
-    return num_tasks
-
-
-async def _postprocess_workflow(result_object: Result) -> Result:
-    """
-    Postprocesses a workflow with a completed computational graph
-
-    Args:
-        result_object: Result object being used for current dispatch
-
-    Returns:
-        The postprocessed result object
-    """
-
-    # Executor for post_processing
-    pp_executor = result_object.lattice.get_metadata("workflow_executor")
-    pp_executor_data = result_object.lattice.get_metadata("workflow_executor_data")
-    post_processor = [pp_executor, pp_executor_data]
-
-    result_object._status = Result.POSTPROCESSING
-    upsert._lattice_data(result_object)
-
-    app_log.debug(f"Preparing to post-process workflow {result_object.dispatch_id}")
-
-    if pp_executor == "client":
-        app_log.debug("Workflow to be postprocessed client side")
-        result_object._status = Result.PENDING_POSTPROCESSING
-        result_object._end_time = datetime.now(timezone.utc)
-        upsert._lattice_data(result_object)
-        await result_webhook.send_update(result_object)
-        return result_object
-
-    post_processing_inputs = {}
-    post_processing_inputs["args"] = [
-        TransportableObject.make_transportable(result_object.lattice),
-        TransportableObject.make_transportable(result_object.get_all_node_outputs()),
-    ]
-    post_processing_inputs["kwargs"] = {}
-
-    try:
-        future = asyncio.create_task(
-            _run_task(
-                result_object=result_object,
-                node_id=-1,
-                serialized_callable=TransportableObject(_post_process),
-                selected_executor=post_processor,
-                node_name="post_process",
-                call_before=[],
-                call_after=[],
-                inputs=post_processing_inputs,
-                workflow_executor=post_processor,
-            )
-        )
-        pp_start_time = datetime.now(timezone.utc)
-        app_log.debug(
-            f"Submitted post-processing job to executor {post_processor} at {pp_start_time}"
-        )
-
-        post_process_result = await future
-    except Exception as ex:
-        app_log.debug(f"Exception during post-processing: {ex}")
-        result_object._status = Result.POSTPROCESSING_FAILED
-        result_object._error = "Post-processing failed"
-        result_object._end_time = datetime.now(timezone.utc)
-        upsert._lattice_data(result_object)
-        await result_webhook.send_update(result_object)
-
-        return result_object
-
-    if post_process_result["status"] != Result.COMPLETED:
-        err = post_process_result["stderr"]
-        app_log.debug(f"Post-processing failed: {err}")
-        result_object._status = Result.POSTPROCESSING_FAILED
-        result_object._error = f"Post-processing failed: {err}"
-        result_object._end_time = datetime.now(timezone.utc)
-        upsert._lattice_data(result_object)
-        await result_webhook.send_update(result_object)
-
-        return result_object
-
-    pp_end_time = post_process_result["end_time"]
-    app_log.debug(f"Post-processing completed at {pp_end_time}")
-    result_object._result = post_process_result["output"]
-    result_object._status = Result.COMPLETED
-    result_object._end_time = datetime.now(timezone.utc)
-
-    app_log.debug(
-        f"10: Successfully post-processed result {result_object.dispatch_id} (run_planned_workflow)"
-    )
-
-    return result_object
-
-
-async def _run_planned_workflow(result_object: Result) -> Result:
-    """
-    Run the workflow in the topological order of their position on the
-    transport graph. Does this in an asynchronous manner so that nodes
-    at the same level are executed in parallel. Also updates the status
-    of the whole workflow execution.
-
-    Args:
-        result_object: Result object being used for current dispatch
-
-    Returns:
-        None
-    """
-
-    app_log.debug("3: Inside run_planned_workflow (run_planned_workflow).")
-
-    tasks_queue = Queue()
-    pending_deps = {}
-    task_futures: list = []
-
-    app_log.debug(
-        f"4: Workflow status changed to running {result_object.dispatch_id} (run_planned_workflow)."
-    )
-
-    result_object._status = Result.RUNNING
-    result_object._start_time = datetime.now(timezone.utc)
-
-    upsert._lattice_data(result_object)
-    app_log.debug("5: Wrote lattice status to DB (run_planned_workflow).")
-
-    # Executor for post_processing and dispatching sublattices
-    pp_executor = result_object.lattice.get_metadata("workflow_executor")
-    pp_executor_data = result_object.lattice.get_metadata("workflow_executor_data")
-    post_processor = [pp_executor, pp_executor_data]
-
-    tasks_left = await _initialize_deps_and_queue(result_object, tasks_queue, pending_deps)
-
-    while tasks_left > 0:
-        app_log.debug(f"{tasks_left} tasks left")
-
-        tasks_left -= 1
-        node_id = await tasks_queue.get()
-        app_log.debug(f"Processing node {node_id}")
-
-        # Tasks queue has been poisoned, don't process the graph further
-        if node_id < 0:
-            break
-
-        # Get name of the node for the current task
-        node_name = result_object.lattice.transport_graph.get_node_value(node_id, "name")
-        app_log.debug(f"7A: Node name: {node_name} (run_planned_workflow).")
-
-        # Handle parameter nodes
-        if node_name.startswith(parameter_prefix):
-            app_log.debug("7C: Parameter if block (run_planned_workflow).")
-            output = result_object.lattice.transport_graph.get_node_value(node_id, "value")
-            app_log.debug(f"7C: Node output: {output} (run_planned_workflow).")
-            app_log.debug("8: Starting update node (run_planned_workflow).")
-
-            node_result = {
-                "node_id": node_id,
-                "start_time": datetime.now(timezone.utc),
-                "end_time": datetime.now(timezone.utc),
-                "status": Result.COMPLETED,
-                "output": output,
-            }
-            await _update_node_result(result_object, node_result, pending_deps, tasks_queue)
-            app_log.debug("8A: Update node success (run_planned_workflow).")
-
-            continue
-
-        # Gather inputs and dispatch task
-        app_log.debug(f"Gathering inputs for task {node_id} (run_planned_workflow).")
-        task_input = _get_task_inputs(node_id, node_name, result_object)
-
-        start_time = datetime.now(timezone.utc)
-        serialized_callable = result_object.lattice.transport_graph.get_node_value(
-            node_id, "function"
-        )
-
-        selected_executor = result_object.lattice.transport_graph.get_node_value(
-            node_id, "metadata"
-        )["executor"]
-
-        selected_executor_data = result_object.lattice.transport_graph.get_node_value(
-            node_id, "metadata"
-        )["executor_data"]
-
-        app_log.debug(f"Collecting deps for task {node_id}")
-        try:
-            call_before, call_after = _gather_deps(result_object, node_id)
-
-        except Exception as ex:
-            app_log.error(f"Exception when trying to collect deps: {ex}")
-            raise ex
-
-        node_result = generate_node_result(
-            node_id=node_id,
-            start_time=start_time,
-            status=Result.RUNNING,
-        )
-        await _update_node_result(result_object, node_result, pending_deps, tasks_queue)
-        app_log.debug("7: Updating nodes after deps (run_planned_workflow)")
-
-        app_log.debug(f"Submitting task {node_id} to executor")
-
-        run_task_callable = partial(
-            _run_task,
-            result_object=result_object,
-            node_id=node_id,
-            serialized_callable=serialized_callable,
-            selected_executor=[selected_executor, selected_executor_data],
-            node_name=node_name,
-            call_before=call_before,
-            call_after=call_after,
-            inputs=task_input,
-            workflow_executor=post_processor,
-        )
-
-        # Add the task generated for the node to the list of tasks
-        future = asyncio.create_task(
-            _run_task_and_update(
-                run_task_callable=run_task_callable,
-                result_object=result_object,
-                pending_deps=pending_deps,
-                tasks_queue=tasks_queue,
-            )
-        )
-
-        task_futures.append(future)
-
-    await asyncio.gather(*task_futures)
-
-    if result_object._status in [Result.FAILED, Result.CANCELLED]:
-        app_log.debug(f"Workflow {result_object.dispatch_id} cancelled or failed")
-
-        failed_nodes = result_object._get_failed_nodes()
-        failed_nodes = map(lambda x: f"{x[0]}: {x[1]}", failed_nodes)
-        failed_nodes_msg = "\n".join(failed_nodes)
-        result_object._error = "The following tasks failed:\n" + failed_nodes_msg
-        upsert._lattice_data(result_object)
-        return result_object
-
-    app_log.debug("8: All tasks finished running (run_planned_workflow)")
-
-    result_object = await _postprocess_workflow(result_object)
-
-    update.persist(result_object)
-    await result_webhook.send_update(result_object)
-
-    return result_object
-
-
-def _plan_workflow(result_object: Result) -> None:
-    """
-    Function to plan a workflow according to a schedule.
-    Planning means to decide which executors (along with their arguments) will
-    be used by each node.
-
-    Args:
-        result_object: Result object being used for current dispatch
-
-    Returns:
-        None
-    """
-
-    if result_object.lattice.get_metadata("schedule"):
-        # Custom scheduling logic of the format:
-        # scheduled_executors = get_schedule(result_object)
-
-        # for node_id, executor in scheduled_executors.items():
-        #    result_object.lattice.transport_graph.set_node_value(node_id, "executor", executor)
-        pass
-
-
-async def run_workflow(result_object: Result) -> Result:
-    """
-    Plan and run the workflow by loading the result object corresponding to the
-    dispatch id and retrieving essential information from it.
-    Returns without changing anything if a redispatch is done of a (partially or fully)
-    completed workflow with the same dispatch id.
-
-    Args:
-        dispatch_id: Dispatch id of the workflow to be run
-        results_dir: Directory where the result object is stored
-
-    Returns:
-        The result object from the workflow execution
-    """
-
-    app_log.debug("Inside run_workflow.")
-
-    if result_object.status == Result.COMPLETED:
-        return result_object
-
-    try:
-        _plan_workflow(result_object)
-        result_object = await _run_planned_workflow(result_object)
-
-    except Exception as ex:
-        app_log.error(f"Exception during _run_planned_workflow: {ex}")
-        update_lattices_data(
-            result_object.dispatch_id,
-            status=str(Result.FAILED),
-            completed_at=datetime.now(timezone.utc),
-            updated_at=datetime.now(timezone.utc),
-        )
-
-        error_msg = "".join(traceback.TracebackException.from_exception(ex).format())
-        result_object._status = Result.FAILED
-        result_object._error = error_msg
-        result_object._end_time = datetime.now(timezone.utc)
-        upsert._lattice_data(result_object)
-        write_lattice_error(
-            result_object.dispatch_id,
-            "".join(traceback.TracebackException.from_exception(ex).format()),
-        )
-
-    return result_object
-
-
-def cancel_workflow(dispatch_id: str) -> None:
-    """
-    Cancels a dispatched workflow using publish subscribe mechanism
-    provided by Dask.
-
-    Args:
-        dispatch_id: Dispatch id of the workflow to be cancelled
-
-    Returns:
-        None
-    """
-
-    # shared_var = Variable(dispatch_id)
-    # shared_var.set(str(Result.CANCELLED))
-    pass
-
-
-def initialize_result_object(
-    json_lattice: str, parent_result_object: Result = None, parent_electron_id: int = None
-) -> Result:
-    """Convenience function for constructing a result object from a json-serialized lattice.
-
-    Args:
-        json_lattice: a JSON-serialized lattice
-        parent_result_object: the parent result object if json_lattice is a sublattice
-        parent_electron_id: the DB id of the parent electron (for sublattices)
-
-    Returns:
-        Result: result object
-    """
-
-    dispatch_id = get_unique_id()
-    lattice = Lattice.deserialize_from_json(json_lattice)
-    result_object = Result(lattice, dispatch_id)
-    if parent_result_object:
-        result_object._root_dispatch_id = parent_result_object._root_dispatch_id
-
-    result_object._initialize_nodes()
-    app_log.debug("2: Constructed result object and initialized nodes.")
-
-    update.persist(result_object, electron_id=parent_electron_id)
-    app_log.debug("Result object persisted.")
-
-    return result_object
-
-
-def get_unique_id() -> str:
-    """
-    Get a unique ID.
-
-    Args:
-        None
-
-    Returns:
-        str: Unique ID
-    """
-
-    return str(uuid.uuid4())
diff --git a/covalent_dispatcher/_core/runner.py b/covalent_dispatcher/_core/runner.py
new file mode 100644
index 00000000..d1bbc7b6
--- /dev/null
+++ b/covalent_dispatcher/_core/runner.py
@@ -0,0 +1,514 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""
+Defines the core functionality of the runner
+"""
+
+import json
+import traceback
+from datetime import datetime, timezone
+from functools import partial
+from typing import Any, Dict, List, Tuple
+
+from covalent._results_manager import Result
+from covalent._shared_files import logger
+from covalent._shared_files.config import get_config
+from covalent._shared_files.context_managers import active_lattice_manager
+from covalent._shared_files.defaults import prefix_separator, sublattice_prefix
+from covalent._workflow import DepsBash, DepsCall, DepsPip
+from covalent._workflow.lattice import Lattice
+from covalent._workflow.transport import TransportableObject
+from covalent.executor import _executor_manager
+from covalent.executor.base import wrapper_fn
+
+from .._db import upsert
+from .._db.write_result_to_db import get_sublattice_electron_id
+from . import data_manager as datasvc
+from . import dispatcher
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+debug_mode = get_config("sdk.log_level") == "debug"
+
+
+# This is to be run out-of-process
+def _build_sublattice_graph(sub: Lattice, parent_metadata: Dict, *args, **kwargs):
+    for k in sub.metadata.keys():
+        if not sub.metadata[k]:
+            sub.metadata[k] = parent_metadata[k]
+
+    sub.build_graph(*args, **kwargs)
+    return sub.serialize_to_json()
+
+
+async def _dispatch_sublattice(
+    parent_result_object: Result,
+    parent_node_id: int,
+    parent_electron_id: int,
+    inputs: Dict,
+    serialized_callable: Any,
+    workflow_executor: Any,
+) -> str:
+    """Dispatch a sublattice using the workflow_executor."""
+
+    app_log.debug("Inside _dispatch_sublattice")
+
+    try:
+        short_name, object_dict = workflow_executor
+
+        if short_name == "client":
+            app_log.error("No executor selected for dispatching sublattices")
+            raise RuntimeError("No executor selected for dispatching sublattices")
+
+    except Exception as ex:
+        app_log.debug(f"Exception when trying to determine sublattice executor: {ex}")
+        raise ex
+
+    parent_metadata = TransportableObject.make_transportable(parent_result_object.lattice.metadata)
+    sub_dispatch_inputs = {
+        "args": [serialized_callable, parent_metadata],
+        "kwargs": inputs["kwargs"],
+    }
+    for arg in inputs["args"]:
+        sub_dispatch_inputs["args"].append(arg)
+
+    # Build the sublattice graph. This must be run
+    # externally since it involves deserializing the
+    # sublattice workflow function.
+    res = await _run_task(
+        result_object=parent_result_object,
+        node_id=parent_node_id,
+        serialized_callable=TransportableObject.make_transportable(_build_sublattice_graph),
+        selected_executor=workflow_executor,
+        node_name="build_sublattice_graph",
+        call_before=[],
+        call_after=[],
+        inputs=sub_dispatch_inputs,
+        workflow_executor=workflow_executor,
+    )
+
+    if res["status"] == Result.COMPLETED:
+        json_sublattice = json.loads(res["output"].json)
+
+        sub_dispatch_id = datasvc.make_dispatch(
+            json_sublattice, parent_result_object, parent_electron_id
+        )
+        app_log.debug(f"Sublattice dispatch id: {sub_dispatch_id}")
+        return sub_dispatch_id
+    else:
+        app_log.debug("Error building sublattice graph")
+        stderr = res["stderr"]
+        raise RuntimeError("Error building sublattice graph: {stderr}")
+
+
+# Domain: runner
+# to be called by _run_abstract_task
+def _get_task_input_values(result_object: Result, abs_task_inputs: dict) -> dict:
+    node_values = {}
+    args = abs_task_inputs["args"]
+    for node_id in args:
+        value = result_object.lattice.transport_graph.get_node_value(node_id, "output")
+        node_values[node_id] = value
+
+    kwargs = abs_task_inputs["kwargs"]
+    for key, node_id in kwargs.items():
+        value = result_object.lattice.transport_graph.get_node_value(node_id, "output")
+        node_values[node_id] = value
+
+    return node_values
+
+
+# Domain: runner
+async def _run_abstract_task(
+    dispatch_id: str,
+    node_id: int,
+    node_name: str,
+    abstract_inputs: Dict,
+    selected_executor: Any,
+    workflow_executor: Any,
+) -> None:
+
+    # Resolve abstract task and inputs to their concrete (serialized) values
+    result_object = datasvc.get_result_object(dispatch_id)
+    timestamp = datetime.now(timezone.utc)
+
+    try:
+        serialized_callable = result_object.lattice.transport_graph.get_node_value(
+            node_id, "function"
+        )
+        input_values = _get_task_input_values(result_object, abstract_inputs)
+
+        abstract_args = abstract_inputs["args"]
+        abstract_kwargs = abstract_inputs["kwargs"]
+        args = [input_values[node_id] for node_id in abstract_args]
+        kwargs = {k: input_values[v] for k, v in abstract_kwargs.items()}
+        task_input = {"args": args, "kwargs": kwargs}
+
+        app_log.debug(f"Collecting deps for task {node_id}")
+
+        call_before, call_after = _gather_deps(result_object, node_id)
+
+    except Exception as ex:
+        app_log.error(f"Exception when trying to resolve inputs or deps: {ex}")
+        node_result = datasvc.generate_node_result(
+            node_id=node_id,
+            start_time=timestamp,
+            end_time=timestamp,
+            status=Result.FAILED,
+            error=str(ex),
+        )
+        return node_result
+
+    node_result = datasvc.generate_node_result(
+        node_id=node_id,
+        start_time=timestamp,
+        status=Result.RUNNING,
+    )
+    app_log.debug(f"7: Marking node {node_id} as running (_run_abstract_task)")
+
+    await datasvc.update_node_result(result_object, node_result)
+
+    return await _run_task(
+        result_object=result_object,
+        node_id=node_id,
+        serialized_callable=serialized_callable,
+        selected_executor=selected_executor,
+        node_name=node_name,
+        call_before=call_before,
+        call_after=call_after,
+        inputs=task_input,
+        workflow_executor=workflow_executor,
+    )
+
+
+# Domain: runner
+async def _run_task(
+    result_object: Result,
+    node_id: int,
+    inputs: Dict,
+    serialized_callable: Any,
+    selected_executor: Any,
+    call_before: List,
+    call_after: List,
+    node_name: str,
+    workflow_executor: Any,
+) -> None:
+    """
+    Run a task with given inputs on the selected executor.
+    Also updates the status of current node execution while
+    checking if a redispatch has occurred. Exclude those nodes
+    from execution which were completed.
+
+    Also verifies if execution of this dispatch has been cancelled.
+
+    Args:
+        inputs: Inputs for the task.
+        result_object: Result object being used for current dispatch
+        node_id: Node id of the task to be executed.
+
+    Returns:
+        None
+    """
+
+    dispatch_id = result_object.dispatch_id
+    results_dir = result_object.results_dir
+
+    # Instantiate the executor from JSON
+    try:
+        short_name, object_dict = selected_executor
+
+        app_log.debug(f"Running task {node_name} using executor {short_name}, {object_dict}")
+
+        # the executor is determined during scheduling and provided in the execution metadata
+        executor = _executor_manager.get_executor(short_name)
+        executor.from_dict(object_dict)
+    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)
+        node_result = datasvc.generate_node_result(
+            node_id=node_id,
+            end_time=datetime.now(timezone.utc),
+            status=Result.FAILED,
+            error=error_msg,
+        )
+        return node_result
+
+    # run the task on the executor and register any failures
+    try:
+
+        if node_name.startswith(sublattice_prefix):
+            sub_electron_id = get_sublattice_electron_id(
+                parent_dispatch_id=dispatch_id, sublattice_node_id=node_id
+            )
+
+            sub_dispatch_id = await _dispatch_sublattice(
+                parent_result_object=result_object,
+                parent_node_id=node_id,
+                parent_electron_id=sub_electron_id,
+                inputs=inputs,
+                serialized_callable=serialized_callable,
+                workflow_executor=workflow_executor,
+            )
+
+            node_result = datasvc.generate_node_result(
+                node_id=node_id,
+                sub_dispatch_id=sub_dispatch_id,
+            )
+
+            dispatcher.run_dispatch(sub_dispatch_id)
+            app_log.debug(f"Running sublattice dispatch {sub_dispatch_id}")
+
+        else:
+            app_log.debug(f"Executing task {node_name}")
+            assembled_callable = partial(wrapper_fn, serialized_callable, call_before, call_after)
+            execute_callable = partial(
+                executor.execute,
+                function=assembled_callable,
+                args=inputs["args"],
+                kwargs=inputs["kwargs"],
+                dispatch_id=dispatch_id,
+                results_dir=results_dir,
+                node_id=node_id,
+            )
+            output, stdout, stderr, exception_raised = await executor._execute(
+                function=assembled_callable,
+                args=inputs["args"],
+                kwargs=inputs["kwargs"],
+                dispatch_id=dispatch_id,
+                results_dir=results_dir,
+                node_id=node_id,
+            )
+            if exception_raised:
+                status = Result.FAILED
+            else:
+                status = Result.COMPLETED
+
+            node_result = datasvc.generate_node_result(
+                node_id=node_id,
+                end_time=datetime.now(timezone.utc),
+                status=status,
+                output=output,
+                stdout=stdout,
+                stderr=stderr,
+            )
+
+    except Exception as ex:
+        tb = "".join(traceback.TracebackException.from_exception(ex).format())
+        app_log.debug(f"Exception occurred when running task {node_id}:")
+        app_log.debug(tb)
+        error_msg = tb if debug_mode else str(ex)
+        node_result = datasvc.generate_node_result(
+            node_id=node_id,
+            end_time=datetime.now(timezone.utc),
+            status=Result.FAILED,
+            error=error_msg,
+        )
+    app_log.debug(f"Node result: {node_result}")
+    return node_result
+
+
+# Domain: runner
+def _gather_deps(result_object: Result, node_id: int) -> Tuple[List, List]:
+    """Assemble deps for a node into the final call_before and call_after"""
+
+    deps = result_object.lattice.transport_graph.get_node_value(node_id, "metadata")["deps"]
+
+    # Assemble call_before and call_after from all the deps
+
+    call_before_objs_json = result_object.lattice.transport_graph.get_node_value(
+        node_id, "metadata"
+    )["call_before"]
+    call_after_objs_json = result_object.lattice.transport_graph.get_node_value(
+        node_id, "metadata"
+    )["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
+
+
+# Domain: runner
+async def run_abstract_task(
+    dispatch_id: str,
+    node_id: int,
+    node_name: str,
+    abstract_inputs: Dict,
+    selected_executor: Any,
+    workflow_executor: Any,
+) -> None:
+
+    node_result = await _run_abstract_task(
+        dispatch_id=dispatch_id,
+        node_id=node_id,
+        node_name=node_name,
+        abstract_inputs=abstract_inputs,
+        selected_executor=selected_executor,
+        workflow_executor=workflow_executor,
+    )
+    result_object = datasvc.get_result_object(dispatch_id)
+    await datasvc.update_node_result(result_object, node_result)
+
+
+# Domain: runner
+# This is to be run out-of-process
+def _post_process(lattice: Lattice, node_outputs: Dict) -> Any:
+    """
+    Post processing function to be called after the lattice execution.
+    This takes care of executing statements that were not an electron
+    but were inside the lattice's function. It also replaces any calls
+    to an electron with the result of that electron execution, hence
+    preventing a local execution of electron's function.
+
+    Note: Here `node_outputs` is used instead of `electron_outputs`
+    since an electron can be called multiple times with possibly different
+    arguments, but every time it's called, it will be executed as a separate node.
+    Thus, output of every node is used.
+
+    Args:
+        lattice: Lattice object that was dispatched.
+        node_outputs: Dictionary containing the output of all the nodes.
+        execution_order: List of lists containing the order of execution of the nodes.
+
+    Reurns:
+        result: The result of the lattice function.
+    """
+
+    ordered_node_outputs = []
+    app_log.debug(f"node_outputs: {node_outputs}")
+    app_log.debug(f"node_outputs: {node_outputs.items()}")
+    for i, item in enumerate(node_outputs.items()):
+        key, val = item
+        app_log.debug(f"Here's the key: {key}")
+        if not key.startswith(prefix_separator) or key.startswith(sublattice_prefix):
+            ordered_node_outputs.append((i, val))
+
+    with active_lattice_manager.claim(lattice):
+        lattice.post_processing = True
+        lattice.electron_outputs = ordered_node_outputs
+        args = [arg.get_deserialized() for arg in lattice.args]
+        kwargs = {k: v.get_deserialized() for k, v in lattice.kwargs.items()}
+        workflow_function = lattice.workflow_function.get_deserialized()
+        result = workflow_function(*args, **kwargs)
+        lattice.post_processing = False
+        return result
+
+
+# Domain: runner
+async def _postprocess_workflow(result_object: Result) -> Result:
+    """
+    Postprocesses a workflow with a completed computational graph
+
+    Args:
+        result_object: Result object being used for current dispatch
+
+    Returns:
+        The postprocessed result object
+    """
+
+    # Executor for post_processing
+    pp_executor = result_object.lattice.get_metadata("workflow_executor")
+    pp_executor_data = result_object.lattice.get_metadata("workflow_executor_data")
+    post_processor = [pp_executor, pp_executor_data]
+
+    result_object._status = Result.POSTPROCESSING
+    upsert._lattice_data(result_object)
+
+    app_log.debug(f"Preparing to post-process workflow {result_object.dispatch_id}")
+
+    if pp_executor == "client":
+        app_log.debug("Workflow to be postprocessed client side")
+        result_object._status = Result.PENDING_POSTPROCESSING
+        result_object._end_time = datetime.now(timezone.utc)
+        upsert._lattice_data(result_object)
+        return result_object
+
+    post_processing_inputs = {}
+    post_processing_inputs["args"] = [
+        TransportableObject.make_transportable(result_object.lattice),
+        TransportableObject.make_transportable(result_object.get_all_node_outputs()),
+    ]
+    post_processing_inputs["kwargs"] = {}
+
+    app_log.debug(f"Submitted post-processing job to executor {post_processor}")
+    post_process_result = await _run_task(
+        result_object=result_object,
+        node_id=-1,
+        serialized_callable=TransportableObject(_post_process),
+        selected_executor=post_processor,
+        node_name="post_process",
+        call_before=[],
+        call_after=[],
+        inputs=post_processing_inputs,
+        workflow_executor=post_processor,
+    )
+
+    if post_process_result["status"] != Result.COMPLETED:
+        stderr = post_process_result["stderr"] if post_process_result["stderr"] else ""
+        err = post_process_result["error"] if post_process_result["error"] else ""
+        error_msg = stderr + err
+
+        app_log.debug(f"Post-processing failed: {err}")
+        result_object._status = Result.POSTPROCESSING_FAILED
+        result_object._error = f"Post-processing failed: {error_msg}"
+        result_object._end_time = datetime.now(timezone.utc)
+        upsert._lattice_data(result_object)
+
+        app_log.debug("Returning from _postprocess_workflow")
+        return result_object
+
+    result_object._result = post_process_result["output"]
+    result_object._status = Result.COMPLETED
+    result_object._end_time = datetime.now(timezone.utc)
+
+    app_log.debug(
+        f"10: Successfully post-processed result {result_object.dispatch_id} (run_planned_workflow)"
+    )
+
+    return result_object
+
+
+async def postprocess_workflow(dispatch_id: str) -> Result:
+    result_object = datasvc.get_result_object(dispatch_id)
+    return await _postprocess_workflow(result_object)
diff --git a/covalent_dispatcher/_db/update.py b/covalent_dispatcher/_db/update.py
index 4d7a61e9..fa1475d5 100644
--- a/covalent_dispatcher/_db/update.py
+++ b/covalent_dispatcher/_db/update.py
@@ -68,6 +68,7 @@ def _node(
     status: "Status" = None,
     output: Any = None,
     error: Exception = None,
+    sub_dispatch_id: str = None,
     sublattice_result: "Result" = None,
     stdout: str = None,
     stderr: str = None,
@@ -100,6 +101,7 @@ def _node(
         status=status,
         output=output,
         error=error,
+        sub_dispatch_id=sub_dispatch_id,
         sublattice_result=sublattice_result,
         stdout=stdout,
         stderr=stderr,
diff --git a/covalent_dispatcher/_db/write_result_to_db.py b/covalent_dispatcher/_db/write_result_to_db.py
index 889f7df2..bb0bc8de 100644
--- a/covalent_dispatcher/_db/write_result_to_db.py
+++ b/covalent_dispatcher/_db/write_result_to_db.py
@@ -220,6 +220,7 @@ def insert_electrons_data(
 
     with workflow_db.session() as session:
         session.add(electron_row)
+        session.flush()
         electron_id = electron_row.id
         session.commit()
 
@@ -398,6 +399,22 @@ def get_sublattice_electron_id(parent_dispatch_id: str, sublattice_node_id: int)
     return sublattice_electron_id
 
 
+def resolve_electron_id(eid: int):
+    """Given an electron's unique id, return the corresponding
+    dispatch_id and node_id
+    """
+    with workflow_db.session() as session:
+        row = (
+            session.query(Lattice, Electron)
+            .where(Electron.id == eid, Lattice.id == Electron.parent_lattice_id)
+            .first()
+        )
+        dispatch_id = row.Lattice.dispatch_id
+        node_id = row.Electron.transport_graph_node_id
+
+    return dispatch_id, node_id
+
+
 def write_lattice_error(dispatch_id: str, error: str):
     with workflow_db.session() as session:
         valid_update = session.query(Lattice).where(Lattice.dispatch_id == dispatch_id).first()
diff --git a/covalent_dispatcher/entry_point.py b/covalent_dispatcher/entry_point.py
index e514133e..9351603c 100644
--- a/covalent_dispatcher/entry_point.py
+++ b/covalent_dispatcher/entry_point.py
@@ -22,7 +22,6 @@
 Self-contained entry point for the dispatcher
 """
 
-import asyncio
 
 from covalent._shared_files import logger
 
@@ -43,12 +42,10 @@ async def run_dispatcher(json_lattice: str):
         dispatch_id: A string containing the dispatch id of current dispatch.
     """
 
-    from ._core import initialize_result_object, run_workflow
+    from ._core import make_dispatch, run_dispatch
 
-    result_object = initialize_result_object(json_lattice)
-
-    dispatch_id = result_object.dispatch_id
-    asyncio.create_task(run_workflow(result_object))
+    dispatch_id = make_dispatch(json_lattice)
+    run_dispatch(dispatch_id)
 
     app_log.debug("Submitted result object to run_workflow.")
 


Test patch:
diff --git a/tests/.gitignore b/tests/.gitignore
index 6f666b68..1ac3b4ea 100644
--- a/tests/.gitignore
+++ b/tests/.gitignore
@@ -21,7 +21,6 @@
 
 # Ignore results folders
 results/**
-result_*
 
 # XML files
 *.xml
diff --git a/tests/covalent_dispatcher_tests/_core/data_manager_test.py b/tests/covalent_dispatcher_tests/_core/data_manager_test.py
new file mode 100644
index 00000000..bf71899d
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_core/data_manager_test.py
@@ -0,0 +1,265 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""
+Tests for the core functionality of the dispatcher.
+"""
+
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+import covalent as ct
+from covalent._results_manager import Result
+from covalent._workflow.lattice import Lattice
+from covalent_dispatcher._core.data_manager import (
+    _dispatch_status_queues,
+    _register_result_object,
+    _registered_dispatches,
+    _update_parent_electron,
+    finalize_dispatch,
+    get_result_object,
+    get_status_queue,
+    initialize_result_object,
+    make_dispatch,
+    persist_result,
+    update_node_result,
+    upsert_lattice_data,
+)
+from covalent_dispatcher._db.datastore import DataStore
+
+TEST_RESULTS_DIR = "/tmp/results"
+
+
+@pytest.fixture
+def test_db():
+    """Instantiate and return an in-memory database."""
+
+    return DataStore(
+        db_URL="sqlite+pysqlite:///:memory:",
+        initialize_db=True,
+    )
+
+
+def get_mock_result() -> Result:
+    """Construct a mock result object corresponding to a lattice."""
+
+    import sys
+
+    @ct.electron(executor="local")
+    def task(x):
+        print(f"stdout: {x}")
+        print("Error!", file=sys.stderr)
+        return x
+
+    @ct.lattice
+    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 test_initialize_result_object(mocker, test_db):
+    """Test the `initialize_result_object` function"""
+
+    @ct.electron
+    def task(x):
+        return x
+
+    @ct.lattice
+    def workflow(x):
+        return task(x)
+
+    workflow.build_graph(1)
+    json_lattice = workflow.serialize_to_json()
+    mocker.patch("covalent_dispatcher._db.upsert.workflow_db", return_value=test_db)
+    mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", return_value=test_db)
+    result_object = get_mock_result()
+
+    mock_persist = mocker.patch("covalent_dispatcher._db.update.persist")
+
+    sub_result_object = initialize_result_object(
+        json_lattice=json_lattice, parent_result_object=result_object, parent_electron_id=5
+    )
+
+    mock_persist.assert_called_with(sub_result_object, electron_id=5)
+    assert sub_result_object._root_dispatch_id == result_object.dispatch_id
+
+
+@pytest.mark.parametrize("node_status", [Result.COMPLETED, Result.FAILED, Result.CANCELLED])
+@pytest.mark.asyncio
+async def test_update_node_result(mocker, node_status):
+    """Check that update_node_result pushes the correct status updates"""
+
+    status_queue = AsyncMock()
+
+    result_object = get_mock_result()
+    mock_update_node = mocker.patch("covalent_dispatcher._db.update._node")
+    mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
+    )
+
+    node_result = {"node_id": 0, "status": node_status}
+    await update_node_result(result_object, node_result)
+    status_queue.put.assert_awaited_with((0, node_status))
+
+
+@pytest.mark.asyncio
+async def test_update_node_result_handles_db_exceptions(mocker):
+    """Check that update_node_result handles db write failures"""
+
+    status_queue = AsyncMock()
+
+    result_object = get_mock_result()
+    mock_update_node = mocker.patch(
+        "covalent_dispatcher._db.update._node", side_effect=RuntimeError()
+    )
+    mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
+    )
+    node_result = {"node_id": 0, "status": Result.COMPLETED}
+    await update_node_result(result_object, node_result)
+
+    status_queue.put.assert_awaited_with((0, Result.FAILED))
+
+
+def test_make_dispatch(mocker):
+    res = get_mock_result()
+    mock_init_result = mocker.patch(
+        "covalent_dispatcher._core.data_manager.initialize_result_object", return_value=res
+    )
+    mock_register = mocker.patch(
+        "covalent_dispatcher._core.data_manager._register_result_object", return_value=res
+    )
+    json_lattice = '{"workflow_function": "asdf"}'
+    dispatch_id = make_dispatch(json_lattice)
+
+    assert dispatch_id == res.dispatch_id
+    mock_register.assert_called_with(res)
+
+
+def test_get_result_object(mocker):
+    result_object = get_mock_result()
+    dispatch_id = result_object.dispatch_id
+    _registered_dispatches[dispatch_id] = result_object
+    assert get_result_object(dispatch_id) is result_object
+    del _registered_dispatches[dispatch_id]
+
+
+def test_register_result_object(mocker):
+    result_object = get_mock_result()
+    dispatch_id = result_object.dispatch_id
+    _register_result_object(result_object)
+    assert _registered_dispatches[dispatch_id] is result_object
+    del _registered_dispatches[dispatch_id]
+
+
+def test_unregister_result_object(mocker):
+    result_object = get_mock_result()
+    dispatch_id = result_object.dispatch_id
+    _registered_dispatches[dispatch_id] = result_object
+    finalize_dispatch(dispatch_id)
+    assert dispatch_id not in _registered_dispatches
+
+
+def test_get_status_queue():
+    import asyncio
+
+    dispatch_id = "dispatch"
+    q = asyncio.Queue()
+    _dispatch_status_queues[dispatch_id] = q
+    assert get_status_queue(dispatch_id) is q
+
+
+@pytest.mark.asyncio
+async def test_persist_result(mocker):
+    result_object = get_mock_result()
+
+    mock_get_result = mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_result_object", return_value=result_object
+    )
+    mock_update_parent = mocker.patch(
+        "covalent_dispatcher._core.data_manager._update_parent_electron"
+    )
+    mock_persist = mocker.patch("covalent_dispatcher._core.data_manager.update.persist")
+
+    await persist_result(result_object.dispatch_id)
+    mock_update_parent.assert_awaited_with(result_object)
+    mock_persist.assert_called_with(result_object)
+
+
+@pytest.mark.parametrize(
+    "sub_status,mapped_status",
+    [(Result.COMPLETED, Result.COMPLETED), (Result.POSTPROCESSING_FAILED, Result.FAILED)],
+)
+@pytest.mark.asyncio
+async def test_update_parent_electron(mocker, sub_status, mapped_status):
+    parent_result_obj = get_mock_result()
+    sub_result_obj = get_mock_result()
+    eid = 5
+    parent_dispatch_id = (parent_result_obj.dispatch_id,)
+    parent_node_id = 2
+    sub_result_obj._electron_id = eid
+    sub_result_obj._status = sub_status
+    sub_result_obj._result = 42
+
+    mock_node_result = {
+        "node_id": parent_node_id,
+        "end_time": sub_result_obj._end_time,
+        "status": mapped_status,
+        "output": sub_result_obj._result,
+        "error": sub_result_obj._error,
+    }
+
+    mock_gen_node_result = mocker.patch(
+        "covalent_dispatcher._core.data_manager.generate_node_result",
+        return_value=mock_node_result,
+    )
+
+    mock_update_node = mocker.patch("covalent_dispatcher._core.data_manager.update_node_result")
+    mock_resolve_eid = mocker.patch(
+        "covalent_dispatcher._core.data_manager.resolve_electron_id",
+        return_value=(parent_dispatch_id, parent_node_id),
+    )
+    mock_get_res = mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_result_object", return_value=parent_result_obj
+    )
+
+    await _update_parent_electron(sub_result_obj)
+
+    mock_get_res.assert_called_with(parent_dispatch_id)
+    mock_update_node.assert_awaited_with(parent_result_obj, mock_node_result)
+
+
+def test_upsert_lattice_data(mocker):
+    result_object = get_mock_result()
+    mock_get_result = mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_result_object", return_value=result_object
+    )
+    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
+    upsert_lattice_data(result_object.dispatch_id)
+    mock_upsert_lattice.assert_called_with(result_object)
diff --git a/tests/covalent_dispatcher_tests/_core/dispatcher_test.py b/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
new file mode 100644
index 00000000..965d8258
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
@@ -0,0 +1,426 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""
+Tests for the core functionality of the dispatcher.
+"""
+
+
+from typing import Dict, List
+
+import cloudpickle as pickle
+import pytest
+
+import covalent as ct
+from covalent._results_manager import Result
+from covalent._workflow.lattice import Lattice
+from covalent_dispatcher._core.dispatcher import (
+    _get_abstract_task_inputs,
+    _get_initial_tasks_and_deps,
+    _handle_cancelled_node,
+    _handle_completed_node,
+    _handle_failed_node,
+    _plan_workflow,
+    _run_planned_workflow,
+    cancel_workflow,
+    run_dispatch,
+    run_workflow,
+)
+from covalent_dispatcher._db.datastore import DataStore
+
+TEST_RESULTS_DIR = "/tmp/results"
+
+
+@pytest.fixture
+def test_db():
+    """Instantiate and return an in-memory database."""
+
+    return DataStore(
+        db_URL="sqlite+pysqlite:///:memory:",
+        initialize_db=True,
+    )
+
+
+def get_mock_result() -> Result:
+    """Construct a mock result object corresponding to a lattice."""
+
+    import sys
+
+    @ct.electron(executor="local")
+    def task(x):
+        print(f"stdout: {x}")
+        print("Error!", file=sys.stderr)
+        return x
+
+    @ct.lattice
+    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 test_plan_workflow():
+    """Test workflow planning method."""
+
+    @ct.electron
+    def task(x):
+        return x
+
+    @ct.lattice
+    def workflow(x):
+        return task(x)
+
+    workflow.metadata["schedule"] = True
+    received_workflow = Lattice.deserialize_from_json(workflow.serialize_to_json())
+    result_object = Result(received_workflow, "asdf")
+    _plan_workflow(result_object=result_object)
+
+    # Updated transport graph post planning
+    updated_tg = pickle.loads(result_object.lattice.transport_graph.serialize(metadata_only=True))
+
+    assert updated_tg["lattice_metadata"]["schedule"]
+
+
+def test_get_abstract_task_inputs():
+    """Test _get_abstract_task_inputs for both dicts and list parameter types"""
+
+    @ct.electron
+    def list_task(arg: List):
+        return len(arg)
+
+    @ct.electron
+    def dict_task(arg: Dict):
+        return len(arg)
+
+    @ct.electron
+    def multivariable_task(x, y):
+        return x, y
+
+    @ct.lattice
+    def list_workflow(arg):
+        return list_task(arg)
+
+    @ct.lattice
+    def dict_workflow(arg):
+        return dict_task(arg)
+
+    #    1   2
+    #     \   \
+    #      0   3
+    #     / /\/
+    #     4   5
+
+    @ct.electron
+    def identity(x):
+        return x
+
+    @ct.lattice
+    def multivar_workflow(x, y):
+        electron_x = identity(x)
+        electron_y = identity(y)
+        res1 = multivariable_task(electron_x, electron_y)
+        res2 = multivariable_task(electron_y, electron_x)
+        res3 = multivariable_task(electron_y, electron_x)
+        res4 = multivariable_task(electron_x, electron_y)
+        return 1
+
+    # list-type inputs
+
+    # Nodes 0=task, 1=:electron_list:, 2=1, 3=2, 4=3
+    list_workflow.build_graph([1, 2, 3])
+    abstract_args = [2, 3, 4]
+    tg = list_workflow.transport_graph
+
+    result_object = Result(lattice=list_workflow, dispatch_id="asdf")
+    abs_task_inputs = _get_abstract_task_inputs(1, tg.get_node_value(1, "name"), result_object)
+
+    expected_inputs = {"args": abstract_args, "kwargs": {}}
+
+    assert abs_task_inputs == expected_inputs
+
+    # dict-type inputs
+
+    # Nodes 0=task, 1=:electron_dict:, 2=1, 3=2
+    dict_workflow.build_graph({"a": 1, "b": 2})
+    abstract_args = {"a": 2, "b": 3}
+    tg = dict_workflow.transport_graph
+
+    result_object = Result(lattice=dict_workflow, dispatch_id="asdf")
+    task_inputs = _get_abstract_task_inputs(1, tg.get_node_value(1, "name"), result_object)
+    expected_inputs = {"args": [], "kwargs": abstract_args}
+
+    assert task_inputs == expected_inputs
+
+    # Check arg order
+    multivar_workflow.build_graph(1, 2)
+    received_lattice = Lattice.deserialize_from_json(multivar_workflow.serialize_to_json())
+    result_object = Result(lattice=received_lattice, dispatch_id="asdf")
+    tg = received_lattice.transport_graph
+
+    assert list(tg._graph.nodes) == [0, 1, 2, 3, 4, 5, 6, 7]
+    tg.set_node_value(0, "output", ct.TransportableObject(1))
+    tg.set_node_value(2, "output", ct.TransportableObject(2))
+
+    task_inputs = _get_abstract_task_inputs(4, tg.get_node_value(4, "name"), result_object)
+    assert task_inputs["args"] == [0, 2]
+
+    task_inputs = _get_abstract_task_inputs(5, tg.get_node_value(5, "name"), result_object)
+    assert task_inputs["args"] == [2, 0]
+
+    task_inputs = _get_abstract_task_inputs(6, tg.get_node_value(6, "name"), result_object)
+    assert task_inputs["args"] == [2, 0]
+
+    task_inputs = _get_abstract_task_inputs(7, tg.get_node_value(7, "name"), result_object)
+    assert task_inputs["args"] == [0, 2]
+
+
+@pytest.mark.asyncio
+async def test_handle_completed_node(mocker):
+    """Unit test for completed node handler"""
+    pending_parents = {}
+
+    result_object = get_mock_result()
+
+    # tg edges are (1, 0), (0, 2)
+    pending_parents[0] = 1
+    pending_parents[1] = 0
+    pending_parents[2] = 1
+
+    mock_upsert_lattice = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
+    )
+
+    node_result = {"node_id": 1, "status": Result.COMPLETED}
+
+    next_nodes = await _handle_completed_node(result_object, 1, pending_parents)
+    assert next_nodes == [0]
+    assert pending_parents == {0: 0, 1: 0, 2: 1}
+
+
+@pytest.mark.asyncio
+async def test_handle_failed_node(mocker):
+    """Unit test for failed node handler"""
+    pending_parents = {}
+
+    result_object = get_mock_result()
+    # tg edges are (1, 0), (0, 2)
+
+    mock_upsert_lattice = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
+    )
+    await _handle_failed_node(result_object, 1)
+
+    mock_upsert_lattice.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_handle_cancelled_node(mocker):
+    """Unit test for cancelled node handler"""
+    pending_parents = {}
+
+    result_object = get_mock_result()
+    # tg edges are (1, 0), (0, 2)
+
+    mock_upsert_lattice = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
+    )
+
+    node_result = {"node_id": 1, "status": Result.CANCELLED}
+
+    await _handle_cancelled_node(result_object, 1)
+    assert result_object._task_cancelled is True
+    mock_upsert_lattice.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_get_initial_tasks_and_deps(mocker):
+    """Test internal function for initializing status_queue and pending_parents"""
+    pending_parents = {}
+
+    result_object = get_mock_result()
+    num_tasks, initial_nodes, pending_parents = await _get_initial_tasks_and_deps(result_object)
+
+    assert initial_nodes == [1]
+    assert pending_parents == {0: 1, 1: 0, 2: 1}
+    assert num_tasks == len(result_object.lattice.transport_graph._graph.nodes)
+
+
+@pytest.mark.asyncio
+async def test_run_dispatch(mocker):
+    res = get_mock_result()
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.get_result_object", return_value=res
+    )
+    mock_run = mocker.patch("covalent_dispatcher._core.dispatcher.run_workflow")
+    run_dispatch(res.dispatch_id)
+    mock_run.assert_called_with(res)
+
+
+@pytest.mark.asyncio
+async def test_run_workflow_normal(mocker):
+    import asyncio
+
+    result_object = get_mock_result()
+    msg_queue = asyncio.Queue()
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.get_status_queue", return_value=msg_queue
+    )
+    mocker.patch("covalent_dispatcher._core.dispatcher._plan_workflow")
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher._run_planned_workflow", return_value=result_object
+    )
+    mock_persist = mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.persist_result")
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    await run_workflow(result_object)
+
+    mock_persist.assert_awaited_with(result_object.dispatch_id)
+    mock_unregister.assert_called_with(result_object.dispatch_id)
+
+
+@pytest.mark.asyncio
+async def test_run_completed_workflow(mocker):
+    import asyncio
+
+    result_object = get_mock_result()
+    result_object._status = Result.COMPLETED
+    msg_queue = asyncio.Queue()
+    mock_get_status_queue = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.get_status_queue", return_value=msg_queue
+    )
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    mock_plan = mocker.patch("covalent_dispatcher._core.dispatcher._plan_workflow")
+    mock_run_planned_workflow = mocker.patch(
+        "covalent_dispatcher._core.dispatcher._run_planned_workflow", return_value=result_object
+    )
+    mock_persist = mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.persist_result")
+
+    await run_workflow(result_object)
+
+    mock_plan.assert_not_called()
+    mock_get_status_queue.assert_not_called()
+    mock_unregister.assert_called_with(result_object.dispatch_id)
+
+
+@pytest.mark.asyncio
+async def test_run_workflow_exception(mocker):
+    import asyncio
+
+    result_object = get_mock_result()
+    msg_queue = asyncio.Queue()
+
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.get_status_queue", return_value=msg_queue
+    )
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    mocker.patch("covalent_dispatcher._core.dispatcher._plan_workflow")
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher._run_planned_workflow",
+        return_value=result_object,
+        side_effect=RuntimeError("Error"),
+    )
+    mock_persist = mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.persist_result")
+
+    result = await run_workflow(result_object)
+
+    assert result.status == Result.FAILED
+    mock_persist.assert_awaited_with(result_object.dispatch_id)
+    mock_unregister.assert_called_with(result_object.dispatch_id)
+
+
+@pytest.mark.asyncio
+async def test_run_planned_workflow_cancelled_update(mocker):
+    import asyncio
+
+    result_object = get_mock_result()
+
+    mock_upsert_lattice = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
+    )
+    tasks_left = 1
+    initial_nodes = [0]
+    pending_deps = {0: 0}
+
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher._get_initial_tasks_and_deps",
+        return_value=(tasks_left, initial_nodes, pending_deps),
+    )
+
+    mock_submit_task = mocker.patch("covalent_dispatcher._core.dispatcher._submit_task")
+
+    def side_effect(result_object, node_id):
+        result_object._task_cancelled = True
+
+    mock_handle_cancelled = mocker.patch(
+        "covalent_dispatcher._core.dispatcher._handle_cancelled_node", side_effect=side_effect
+    )
+    status_queue = asyncio.Queue()
+    status_queue.put_nowait((0, Result.CANCELLED))
+    await _run_planned_workflow(result_object, status_queue)
+    assert mock_submit_task.await_count == 1
+    mock_handle_cancelled.assert_awaited_with(result_object, 0)
+
+
+@pytest.mark.asyncio
+async def test_run_planned_workflow_failed_update(mocker):
+    import asyncio
+
+    result_object = get_mock_result()
+
+    mock_upsert_lattice = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
+    )
+    tasks_left = 1
+    initial_nodes = [0]
+    pending_deps = {0: 0}
+
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher._get_initial_tasks_and_deps",
+        return_value=(tasks_left, initial_nodes, pending_deps),
+    )
+
+    mock_submit_task = mocker.patch("covalent_dispatcher._core.dispatcher._submit_task")
+
+    def side_effect(result_object, node_id):
+        result_object._task_failed = True
+
+    mock_handle_failed = mocker.patch(
+        "covalent_dispatcher._core.dispatcher._handle_failed_node", side_effect=side_effect
+    )
+    status_queue = asyncio.Queue()
+    status_queue.put_nowait((0, Result.FAILED))
+    await _run_planned_workflow(result_object, status_queue)
+    assert mock_submit_task.await_count == 1
+    mock_handle_failed.assert_awaited_with(result_object, 0)
+
+
+def test_cancelled_workflow():
+    cancel_workflow("asdf")
diff --git a/tests/covalent_dispatcher_tests/_core/execution_test.py b/tests/covalent_dispatcher_tests/_core/execution_test.py
index 056d93bf..1769c604 100644
--- a/tests/covalent_dispatcher_tests/_core/execution_test.py
+++ b/tests/covalent_dispatcher_tests/_core/execution_test.py
@@ -19,38 +19,21 @@
 # Relief from the License may be granted by purchasing a commercial license.
 
 """
-Tests for the core functionality of the dispatcher.
+Integration tests for the dispatcher, runner, and result modules
 """
 
-
 import asyncio
-from asyncio import Queue
 from typing import Dict, List
-from unittest.mock import AsyncMock, MagicMock
 
-import cloudpickle as pickle
 import pytest
 
 import covalent as ct
 from covalent._results_manager import Result
 from covalent._shared_files.defaults import sublattice_prefix
 from covalent._workflow.lattice import Lattice
-from covalent_dispatcher._core.execution import (
-    _build_sublattice_graph,
-    _dispatch_sync_sublattice,
-    _gather_deps,
-    _get_task_inputs,
-    _handle_cancelled_node,
-    _handle_completed_node,
-    _handle_failed_node,
-    _initialize_deps_and_queue,
-    _plan_workflow,
-    _post_process,
-    _run_task,
-    _update_node_result,
-    initialize_result_object,
-    run_workflow,
-)
+from covalent_dispatcher._core.dispatcher import run_workflow
+from covalent_dispatcher._core.execution import _get_task_inputs
+from covalent_dispatcher._core.runner import _run_task
 from covalent_dispatcher._db import update
 from covalent_dispatcher._db.datastore import DataStore
 
@@ -91,112 +74,6 @@ def get_mock_result() -> Result:
     return result_object
 
 
-def test_initialize_result_object(mocker, test_db):
-    """Test the `initialize_result_object` function"""
-
-    @ct.electron
-    def task(x):
-        return x
-
-    @ct.lattice
-    def workflow(x):
-        return task(x)
-
-    workflow.build_graph(1)
-    json_lattice = workflow.serialize_to_json()
-    mocker.patch("covalent_dispatcher._db.upsert.workflow_db", return_value=test_db)
-    mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", return_value=test_db)
-    result_object = get_mock_result()
-
-    mock_persist = mocker.patch("covalent_dispatcher._db.update.persist")
-
-    sub_result_object = initialize_result_object(
-        json_lattice=json_lattice, parent_result_object=result_object, parent_electron_id=5
-    )
-
-    mock_persist.assert_called_with(sub_result_object, electron_id=5)
-    assert sub_result_object._root_dispatch_id == result_object.dispatch_id
-
-
-def test_plan_workflow():
-    """Test workflow planning method."""
-
-    @ct.electron
-    def task(x):
-        return x
-
-    @ct.lattice
-    def workflow(x):
-        return task(x)
-
-    workflow.metadata["schedule"] = True
-    received_workflow = Lattice.deserialize_from_json(workflow.serialize_to_json())
-    result_object = Result(received_workflow, "asdf")
-    _plan_workflow(result_object=result_object)
-
-    # Updated transport graph post planning
-    updated_tg = pickle.loads(result_object.lattice.transport_graph.serialize(metadata_only=True))
-
-    assert updated_tg["lattice_metadata"]["schedule"]
-
-
-def test_post_process():
-    """Test post-processing of results."""
-
-    import covalent as ct
-
-    @ct.electron
-    def construct_cu_slab(x):
-        return x
-
-    @ct.electron
-    def compute_system_energy(x):
-        return x
-
-    @ct.electron
-    def construct_n_molecule(x):
-        return x
-
-    @ct.electron
-    def get_relaxed_slab(x):
-        return x
-
-    @ct.lattice
-    def compute_energy():
-        N2 = construct_n_molecule(1)
-        e_N2 = compute_system_energy(N2)
-
-        slab = construct_cu_slab(2)
-        e_slab = compute_system_energy(slab)
-
-        relaxed_slab = get_relaxed_slab(3)
-        e_relaxed_slab = compute_system_energy(relaxed_slab)
-
-        return (N2, e_N2, slab, e_slab, relaxed_slab, e_relaxed_slab)
-
-    compute_energy.build_graph()
-
-    node_outputs = {
-        "construct_n_molecule(0)": 1,
-        ":parameter:1(1)": 1,
-        "compute_system_energy(2)": 1,
-        "construct_cu_slab(3)": 2,
-        ":parameter:2(4)": 2,
-        "compute_system_energy(5)": 2,
-        "get_relaxed_slab(6)": 3,
-        ":parameter:3(7)": 3,
-        "compute_system_energy(8)": 3,
-    }
-
-    encoded_node_outputs = {
-        k: ct.TransportableObject.make_transportable(v) for k, v in node_outputs.items()
-    }
-
-    execution_result = _post_process(compute_energy, encoded_node_outputs)
-
-    assert execution_result == compute_energy()
-
-
 def test_get_task_inputs():
     """Test _get_task_inputs for both dicts and list parameter types"""
 
@@ -253,12 +130,9 @@ def test_get_task_inputs():
     result_object = Result(lattice=list_workflow, dispatch_id="asdf")
     task_inputs = _get_task_inputs(1, tg.get_node_value(1, "name"), result_object)
 
-    expected_inputs = {"args": [], "kwargs": {"x": ct.TransportableObject(serialized_args)}}
+    expected_inputs = {"args": serialized_args, "kwargs": {}}
 
-    assert (
-        task_inputs["kwargs"]["x"].get_deserialized()
-        == expected_inputs["kwargs"]["x"].get_deserialized()
-    )
+    assert task_inputs == expected_inputs
 
     # dict-type inputs
 
@@ -271,12 +145,9 @@ def test_get_task_inputs():
 
     result_object = Result(lattice=dict_workflow, dispatch_id="asdf")
     task_inputs = _get_task_inputs(1, tg.get_node_value(1, "name"), result_object)
-    expected_inputs = {"args": [], "kwargs": {"x": ct.TransportableObject(serialized_args)}}
+    expected_inputs = {"args": [], "kwargs": serialized_args}
 
-    assert (
-        task_inputs["kwargs"]["x"].get_deserialized()
-        == expected_inputs["kwargs"]["x"].get_deserialized()
-    )
+    assert task_inputs == expected_inputs
 
     # Check arg order
     multivar_workflow.build_graph(1, 2)
@@ -306,183 +177,6 @@ def test_get_task_inputs():
     assert input_args == [1, 2]
 
 
-def test_gather_deps():
-    """Test internal _gather_deps for assembling deps into call_before and
-    call_after"""
-
-    def square(x):
-        return x * x
-
-    @ct.electron(deps_bash=ct.DepsBash("ls -l"), call_after=[ct.DepsCall(square, [3])])
-    def task(x):
-        return x
-
-    @ct.lattice
-    def workflow(x):
-        return task(x)
-
-    workflow.build_graph(5)
-
-    received_workflow = Lattice.deserialize_from_json(workflow.serialize_to_json())
-    result_object = Result(received_workflow, "asdf")
-
-    before, after = _gather_deps(result_object, 0)
-    assert len(before) == 1
-    assert len(after) == 1
-
-
-@pytest.mark.asyncio
-async def test_update_failed_node(mocker):
-    """Check that update_node_result correctly invokes _handle_failed_node"""
-
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-    mock_fail_handler = mocker.patch("covalent_dispatcher._core.execution._handle_failed_node")
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
-    mock_update_node = mocker.patch("covalent_dispatcher._db.update._node")
-
-    node_result = {"node_id": 0, "status": Result.FAILED}
-    await _update_node_result(result_object, node_result, pending_deps, tasks_queue)
-
-    mock_fail_handler.assert_called_once_with(
-        result_object, node_result, pending_deps, tasks_queue
-    )
-
-
-@pytest.mark.asyncio
-async def test_update_cancelled_node(mocker):
-    """Check that update_node_result correctly invokes _handle_cancelled_node"""
-
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-    mock_cancel_handler = mocker.patch(
-        "covalent_dispatcher._core.execution._handle_cancelled_node"
-    )
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
-    mock_update_node = mocker.patch("covalent_dispatcher._db.update._node")
-
-    node_result = {"node_id": 0, "status": Result.CANCELLED}
-    await _update_node_result(result_object, node_result, pending_deps, tasks_queue)
-
-    mock_cancel_handler.assert_called_once_with(
-        result_object, node_result, pending_deps, tasks_queue
-    )
-
-
-@pytest.mark.asyncio
-async def test_update_completed_node(mocker):
-    """Check that update_node_result correctly invokes _handle_completed_node"""
-
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-    mock_completed_handler = mocker.patch(
-        "covalent_dispatcher._core.execution._handle_completed_node"
-    )
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
-    mock_update_node = mocker.patch("covalent_dispatcher._db.update._node")
-
-    node_result = {"node_id": 0, "status": Result.COMPLETED}
-    await _update_node_result(result_object, node_result, pending_deps, tasks_queue)
-
-    mock_completed_handler.assert_called_once_with(
-        result_object, node_result, pending_deps, tasks_queue
-    )
-
-
-@pytest.mark.asyncio
-async def test_handle_completed_node(mocker):
-    """Unit test for completed node handler"""
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-
-    # tg edges are (1, 0), (0, 2)
-    pending_deps[0] = 1
-    pending_deps[1] = 0
-    pending_deps[2] = 1
-
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
-
-    node_result = {"node_id": 1, "status": Result.COMPLETED}
-
-    await _handle_completed_node(result_object, node_result, pending_deps, tasks_queue)
-
-    assert await asyncio.wait_for(tasks_queue.get(), timeout=1) == 0
-    assert pending_deps == {0: 0, 1: 0, 2: 1}
-
-
-@pytest.mark.asyncio
-async def test_handle_failed_node(mocker):
-    """Unit test for failed node handler"""
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-
-    # tg edges are (1, 0), (0, 2)
-    pending_deps[0] = 1
-    pending_deps[1] = 0
-    pending_deps[2] = 1
-
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
-    mock_get_node_name = mocker.patch("covalent._results_manager.result.Result._get_node_name")
-
-    mock_get_node_error = mocker.patch("covalent._results_manager.result.Result._get_node_error")
-
-    node_result = {"node_id": 1, "status": Result.FAILED}
-
-    await _handle_failed_node(result_object, node_result, pending_deps, tasks_queue)
-
-    assert await asyncio.wait_for(tasks_queue.get(), timeout=1) == -1
-    assert pending_deps == {0: 1, 1: 0, 2: 1}
-    assert result_object.status == Result.FAILED
-
-
-@pytest.mark.asyncio
-async def test_handle_cancelled_node(mocker):
-    """Unit test for cancelled node handler"""
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-
-    # tg edges are (1, 0), (0, 2)
-    pending_deps[0] = 1
-    pending_deps[1] = 0
-    pending_deps[2] = 1
-
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
-
-    node_result = {"node_id": 1, "status": Result.CANCELLED}
-
-    await _handle_cancelled_node(result_object, node_result, pending_deps, tasks_queue)
-
-    assert await asyncio.wait_for(tasks_queue.get(), timeout=1) == -1
-    assert pending_deps == {0: 1, 1: 0, 2: 1}
-    assert result_object.status == Result.CANCELLED
-
-
-@pytest.mark.asyncio
-async def test_initialize_deps_and_queue(mocker):
-    """Test internal function for initializing tasks_queue and pending_deps"""
-    tasks_queue = Queue()
-    pending_deps = {}
-
-    result_object = get_mock_result()
-    num_tasks = await _initialize_deps_and_queue(result_object, tasks_queue, pending_deps)
-
-    assert await asyncio.wait_for(tasks_queue.get(), timeout=1) == 1
-    assert pending_deps == {0: 1, 1: 0, 2: 1}
-    assert num_tasks == len(result_object.lattice.transport_graph._graph.nodes)
-
-
 @pytest.mark.asyncio
 async def test_run_workflow_with_failing_nonleaf(mocker):
     """Test running workflow with a failing intermediate node"""
@@ -519,8 +213,16 @@ async def test_run_workflow_with_failing_nonleaf(mocker):
     mocker.patch(
         "covalent._results_manager.result.Result._get_node_error", return_value="AssertionError"
     )
-    mocker.patch("covalent_dispatcher._core.execution.update_lattices_data")
-    mocker.patch("covalent_dispatcher._core.execution.write_lattice_error")
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
+    status_queue = asyncio.Queue()
+    mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
+    )
     mock_get_failed_nodes = mocker.patch(
         "covalent._results_manager.result.Result._get_failed_nodes",
         return_value=[(0, "failing_task")],
@@ -528,7 +230,7 @@ async def test_run_workflow_with_failing_nonleaf(mocker):
 
     update.persist(result_object)
     result_object = await run_workflow(result_object)
-
+    mock_unregister.assert_called_with(result_object.dispatch_id)
     assert result_object.status == Result.FAILED
 
     mock_get_failed_nodes.assert_called()
@@ -570,8 +272,17 @@ async def test_run_workflow_with_failing_leaf(mocker):
     mocker.patch(
         "covalent._results_manager.result.Result._get_node_error", return_value="AssertionError"
     )
-    mocker.patch("covalent_dispatcher._core.execution.update_lattices_data")
-    mocker.patch("covalent_dispatcher._core.execution.write_lattice_error")
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
+
+    status_queue = asyncio.Queue()
+    mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
+    )
     mock_get_failed_nodes = mocker.patch(
         "covalent._results_manager.result.Result._get_failed_nodes",
         return_value=[(0, "failing_task")],
@@ -580,7 +291,7 @@ async def test_run_workflow_with_failing_leaf(mocker):
     update.persist(result_object)
 
     result_object = await run_workflow(result_object)
-    mock_get_failed_nodes.assert_called()
+    mock_unregister.assert_called_with(result_object.dispatch_id)
     assert result_object.status == Result.FAILED
     assert result_object._error == "The following tasks failed:\n0: failing_task"
 
@@ -618,6 +329,9 @@ async def test_run_workflow_does_not_deserialize(mocker):
     result_object._initialize_nodes()
 
     mocker.patch("covalent_dispatcher._db.datastore.DataStore.factory", return_value=test_db)
+    mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
     update.persist(result_object)
 
     mock_to_deserialize = mocker.patch("covalent.TransportableObject.get_deserialized")
@@ -633,8 +347,6 @@ async def test_run_workflow_with_client_side_postprocess(test_db, mocker):
     """Check that run_workflow handles "client" workflow_executor for
     postprocessing"""
 
-    import asyncio
-
     dispatch_id = "asdf"
     result_object = get_mock_result()
     result_object.lattice.set_metadata("workflow_executor", "client")
@@ -643,10 +355,22 @@ async def test_run_workflow_with_client_side_postprocess(test_db, mocker):
 
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
+
+    status_queue = asyncio.Queue()
+    mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
+    )
 
     update.persist(result_object)
 
     result_object = await run_workflow(result_object)
+    mock_unregister.assert_called_with(result_object.dispatch_id)
     assert result_object.status == Result.PENDING_POSTPROCESSING
 
 
@@ -661,14 +385,26 @@ async def test_run_workflow_with_failed_postprocess(test_db, mocker):
 
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
+    mock_unregister = mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
+    )
+    mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
 
     update.persist(result_object)
 
+    status_queue = asyncio.Queue()
+    mocker.patch(
+        "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
+    )
+
     def failing_workflow(x):
         assert False
 
     result_object.lattice.set_metadata("workflow_executor", "bogus")
     result_object = await run_workflow(result_object)
+    mock_unregister.assert_called_with(result_object.dispatch_id)
 
     assert result_object.status == Result.POSTPROCESSING_FAILED
 
@@ -676,107 +412,11 @@ async def test_run_workflow_with_failed_postprocess(test_db, mocker):
     result_object.lattice.set_metadata("workflow_executor", "local")
 
     result_object = await run_workflow(result_object)
+    mock_unregister.assert_called_with(result_object.dispatch_id)
 
     assert result_object.status == Result.POSTPROCESSING_FAILED
 
 
-def test_build_sublattice_graph():
-    @ct.electron
-    def task(x):
-        return x
-
-    @ct.lattice
-    def workflow(x):
-        return task(x)
-
-    parent_metadata = {
-        "executor": "parent_executor",
-        "executor_data": {},
-        "workflow_executor": "my_postprocessor",
-        "workflow_executor_data": {},
-        "deps": {"bash": None, "pip": None},
-        "call_before": [],
-        "call_after": [],
-        "results_dir": None,
-    }
-
-    json_lattice = _build_sublattice_graph(workflow, parent_metadata, 1)
-    lattice = Lattice.deserialize_from_json(json_lattice)
-
-    assert list(lattice.transport_graph._graph.nodes) == [0, 1]
-
-    for k in lattice.metadata.keys():
-        # results_dir will be deprecated soon
-        if k != "results_dir" and k != "workflow_executor" and k != "workflow_executor_data":
-            assert parent_metadata[k] == lattice.metadata[k]
-
-
-@pytest.mark.asyncio
-async def test_dispatch_sync_sublattice(test_db, mocker):
-    @ct.electron(executor="local")
-    def task(x):
-        return x
-
-    @ct.lattice(executor="local", workflow_executor="local")
-    def sub_workflow(x):
-        return task(x)
-
-    mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
-    mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
-
-    result_object = get_mock_result()
-
-    serialized_callable = ct.TransportableObject(sub_workflow)
-    inputs = {"args": [ct.TransportableObject(2)], "kwargs": {}}
-
-    sub_result = await _dispatch_sync_sublattice(
-        parent_result_object=result_object,
-        parent_electron_id=1,
-        inputs=inputs,
-        serialized_callable=serialized_callable,
-        workflow_executor=["local", {}],
-    )
-
-    # check that sublattice inherits parent lattice's bash dep
-    sub_bash_dep = sub_result.lattice.metadata["deps"]["bash"]["attributes"]["commands"]
-    assert sub_bash_dep[0] == "ls"
-    assert sub_result.result == 2
-
-    sub_workflow.metadata["deps"]["bash"] = ct.DepsBash(["pwd"])
-    serialized_callable = ct.TransportableObject(sub_workflow)
-    sub_result = await _dispatch_sync_sublattice(
-        parent_result_object=result_object,
-        parent_electron_id=1,
-        inputs=inputs,
-        serialized_callable=serialized_callable,
-        workflow_executor=["local", {}],
-    )
-
-    # check that sublattice's explicit bash dep overrides parent lattice's bash dep
-    sub_bash_dep = sub_result.lattice.metadata["deps"]["bash"]["attributes"]["commands"]
-    assert sub_bash_dep[0] == "pwd"
-
-    # Now Check handling of invalid workflow executors
-
-    sub_result = await _dispatch_sync_sublattice(
-        parent_result_object=result_object,
-        parent_electron_id=1,
-        inputs=inputs,
-        serialized_callable=serialized_callable,
-        workflow_executor=["client", {}],
-    )
-    assert sub_result is None
-
-    sub_result = await _dispatch_sync_sublattice(
-        parent_result_object=result_object,
-        parent_electron_id=1,
-        inputs=inputs,
-        serialized_callable=serialized_callable,
-        workflow_executor=["fake_executor", {}],
-    )
-    assert sub_result is None
-
-
 @pytest.mark.asyncio
 async def test_run_task_sublattice_handling(test_db, mocker):
 
@@ -788,12 +428,13 @@ async def test_run_task_sublattice_handling(test_db, mocker):
 
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mock_get_sublattice_electron_id = mocker.patch(
-        "covalent_dispatcher._core.execution.get_sublattice_electron_id", return_value=1
+        "covalent_dispatcher._core.runner.get_sublattice_electron_id", return_value=1
     )
-    mock_dispatch_sync = mocker.patch(
-        "covalent_dispatcher._core.execution._dispatch_sync_sublattice",
-        return_value=sub_result_object,
+    mock_dispatch_sub = mocker.patch(
+        "covalent_dispatcher._core.runner._dispatch_sublattice",
+        return_value=sub_result_object.dispatch_id,
     )
+    mock_run_dispatch = mocker.patch("covalent_dispatcher._core.dispatcher.run_dispatch")
 
     inputs = {"args": [], "kwargs": {}}
 
@@ -810,32 +451,16 @@ async def test_run_task_sublattice_handling(test_db, mocker):
     )
 
     mock_get_sublattice_electron_id.assert_called_once()
-    mock_dispatch_sync.assert_awaited_once()
-    assert node_result["output"].get_deserialized() == 5
+    mock_dispatch_sub.assert_awaited_once()
+    assert node_result["sub_dispatch_id"] == sub_result_object.dispatch_id
+    mock_run_dispatch.assert_called_once()
 
     # Test failed sublattice workflows
     sub_result_object._status = Result.FAILED
-    mock_dispatch_sync = mocker.patch(
-        "covalent_dispatcher._core.execution._dispatch_sync_sublattice",
-        return_value=sub_result_object,
-    )
-    node_result = await _run_task(
-        result_object=result_object,
-        node_id=1,
-        inputs=inputs,
-        serialized_callable=None,
-        selected_executor=["local", {}],
-        call_before=[],
-        call_after=[],
-        node_name=sublattice_prefix,
-        workflow_executor=["local", {}],
-    )
+    mock_run_dispatch = mocker.patch("covalent_dispatcher._core.dispatcher.run_dispatch")
 
-    mock_dispatch_sync.assert_awaited_once()
-    assert node_result["status"] == Result.FAILED
-
-    mock_dispatch_sync = mocker.patch(
-        "covalent_dispatcher._core.execution._dispatch_sync_sublattice", return_value=None
+    mock_dispatch_sub = mocker.patch(
+        "covalent_dispatcher._core.runner._dispatch_sublattice", side_effect=RuntimeError()
     )
     node_result = await _run_task(
         result_object=result_object,
@@ -849,129 +474,6 @@ async def test_run_task_sublattice_handling(test_db, mocker):
         workflow_executor=["local", {}],
     )
 
+    mock_dispatch_sub.assert_awaited_once()
+    mock_run_dispatch.assert_not_called()
     assert node_result["status"] == Result.FAILED
-
-
-@pytest.mark.asyncio
-async def test_run_task_executor_exception_handling(mocker):
-    """Test that exceptions from initializing executors are caught"""
-
-    result_object = get_mock_result()
-    inputs = {"args": [], "kwargs": {}}
-    mock_get_executor = mocker.patch(
-        "covalent_dispatcher._core.execution._executor_manager.get_executor",
-        side_effect=Exception(),
-    )
-
-    node_result = await _run_task(
-        result_object=result_object,
-        node_id=1,
-        inputs=inputs,
-        serialized_callable=None,
-        selected_executor=["nonexistent", {}],
-        call_before=[],
-        call_after=[],
-        node_name=sublattice_prefix,
-        workflow_executor=["local", {}],
-    )
-
-    assert node_result["status"] == Result.FAILED
-
-
-@pytest.mark.asyncio
-async def test_run_task_runtime_exception_handling(mocker):
-
-    result_object = get_mock_result()
-    inputs = {"args": [], "kwargs": {}}
-    mock_executor = MagicMock()
-    mock_executor._execute = AsyncMock(return_value=("", "", "error", True))
-    mock_get_executor = mocker.patch(
-        "covalent_dispatcher._core.execution._executor_manager.get_executor",
-        return_value=mock_executor,
-    )
-
-    node_result = await _run_task(
-        result_object=result_object,
-        node_id=1,
-        inputs=inputs,
-        serialized_callable=None,
-        selected_executor=["local", {}],
-        call_before=[],
-        call_after=[],
-        node_name="task",
-        workflow_executor=["local", {}],
-    )
-
-    mock_executor._execute.assert_awaited_once()
-
-    assert node_result["stderr"] == "error"
-
-
-@pytest.mark.asyncio
-async def test_run_workflow_normal(mocker):
-    result_object = get_mock_result()
-    mocker.patch("covalent_dispatcher._core.execution._plan_workflow")
-    mocker.patch(
-        "covalent_dispatcher._core.execution._run_planned_workflow", return_value=result_object
-    )
-    mock_update_lattices_data = mocker.patch(
-        "covalent_dispatcher._core.execution.update_lattices_data"
-    )
-    mock_write_lattice_error = mocker.patch(
-        "covalent_dispatcher._core.execution.write_lattice_error"
-    )
-
-    await run_workflow(result_object)
-
-    mock_update_lattices_data.assert_not_called()
-    mock_write_lattice_error.assert_not_called()
-
-
-@pytest.mark.asyncio
-async def test_run_completed_workflow(mocker):
-    result_object = get_mock_result()
-    result_object._status = Result.COMPLETED
-    mock_plan = mocker.patch("covalent_dispatcher._core.execution._plan_workflow")
-    mock_run_planned_workflow = mocker.patch(
-        "covalent_dispatcher._core.execution._run_planned_workflow", return_value=result_object
-    )
-    mock_update_lattices_data = mocker.patch(
-        "covalent_dispatcher._core.execution.update_lattices_data"
-    )
-    mock_write_lattice_error = mocker.patch(
-        "covalent_dispatcher._core.execution.write_lattice_error"
-    )
-
-    await run_workflow(result_object)
-
-    mock_plan.assert_not_called()
-    mock_run_planned_workflow.assert_not_called()
-    mock_update_lattices_data.assert_not_called()
-    mock_write_lattice_error.assert_not_called()
-
-
-@pytest.mark.asyncio
-async def test_run_workflow_exception(mocker):
-    result_object = get_mock_result()
-    mocker.patch("covalent_dispatcher._core.execution._plan_workflow")
-    mocker.patch(
-        "covalent_dispatcher._core.execution._run_planned_workflow",
-        return_value=result_object,
-        side_effect=RuntimeError("Error"),
-    )
-    mock_update_lattices_data = mocker.patch(
-        "covalent_dispatcher._core.execution.update_lattices_data"
-    )
-    mock_upsert_lattice_data = mocker.patch(
-        "covalent_dispatcher._core.execution.upsert._lattice_data"
-    )
-    mock_write_lattice_error = mocker.patch(
-        "covalent_dispatcher._core.execution.write_lattice_error"
-    )
-
-    result = await run_workflow(result_object)
-
-    assert result.status == Result.FAILED
-    mock_update_lattices_data.assert_called_once()
-    mock_upsert_lattice_data.assert_called_once()
-    mock_write_lattice_error.assert_called_once()
diff --git a/tests/covalent_dispatcher_tests/_core/runner_test.py b/tests/covalent_dispatcher_tests/_core/runner_test.py
new file mode 100644
index 00000000..a3c1155a
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_core/runner_test.py
@@ -0,0 +1,380 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""
+Tests for the core functionality of the runner.
+"""
+
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import covalent as ct
+from covalent._results_manager import Result
+from covalent._workflow.lattice import Lattice
+from covalent_dispatcher._core.data_manager import get_result_object
+from covalent_dispatcher._core.runner import (
+    _build_sublattice_graph,
+    _dispatch_sublattice,
+    _gather_deps,
+    _post_process,
+    _postprocess_workflow,
+    _run_abstract_task,
+    _run_task,
+)
+from covalent_dispatcher._db.datastore import DataStore
+
+TEST_RESULTS_DIR = "/tmp/results"
+
+
+@pytest.fixture
+def test_db():
+    """Instantiate and return an in-memory database."""
+
+    return DataStore(
+        db_URL="sqlite+pysqlite:///:memory:",
+        initialize_db=True,
+    )
+
+
+def get_mock_result() -> Result:
+    """Construct a mock result object corresponding to a lattice."""
+
+    import sys
+
+    @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 test_gather_deps():
+    """Test internal _gather_deps for assembling deps into call_before and
+    call_after"""
+
+    def square(x):
+        return x * x
+
+    @ct.electron(
+        deps_bash=ct.DepsBash("ls -l"),
+        deps_pip=ct.DepsPip(["pandas"]),
+        call_before=[ct.DepsCall(square, [5])],
+        call_after=[ct.DepsCall(square, [3])],
+    )
+    def task(x):
+        return x
+
+    @ct.lattice
+    def workflow(x):
+        return task(x)
+
+    workflow.build_graph(5)
+
+    received_workflow = Lattice.deserialize_from_json(workflow.serialize_to_json())
+    result_object = Result(received_workflow, "asdf")
+
+    before, after = _gather_deps(result_object, 0)
+    assert len(before) == 3
+    assert len(after) == 1
+
+
+@pytest.mark.asyncio
+async def test_run_abstract_task_exception_handling(mocker):
+    """Test that exceptions from resolving abstract inputs are handled"""
+
+    result_object = get_mock_result()
+    inputs = {"args": [], "kwargs": {}}
+    mock_get_result = mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
+    mock_get_task_input_values = mocker.patch(
+        "covalent_dispatcher._core.runner._get_task_input_values",
+        side_effect=RuntimeError(),
+    )
+
+    node_result = await _run_abstract_task(
+        dispatch_id=result_object.dispatch_id,
+        node_id=0,
+        node_name="test_node",
+        abstract_inputs=inputs,
+        selected_executor=["local", {}],
+        workflow_executor=["local", {}],
+    )
+
+    assert node_result["status"] == Result.FAILED
+
+
+@pytest.mark.asyncio
+async def test_run_task_executor_exception_handling(mocker):
+    """Test that exceptions from initializing executors are caught"""
+
+    result_object = get_mock_result()
+    inputs = {"args": [], "kwargs": {}}
+    mock_get_executor = mocker.patch(
+        "covalent_dispatcher._core.runner._executor_manager.get_executor",
+        side_effect=Exception(),
+    )
+
+    node_result = await _run_task(
+        result_object=result_object,
+        node_id=1,
+        inputs=inputs,
+        serialized_callable=None,
+        selected_executor=["nonexistent", {}],
+        call_before=[],
+        call_after=[],
+        node_name="test_node",
+        workflow_executor=["local", {}],
+    )
+
+    assert node_result["status"] == Result.FAILED
+
+
+@pytest.mark.asyncio
+async def test_run_task_runtime_exception_handling(mocker):
+
+    result_object = get_mock_result()
+    inputs = {"args": [], "kwargs": {}}
+    mock_executor = MagicMock()
+    mock_executor._execute = AsyncMock(return_value=("", "", "error", True))
+    mock_get_executor = mocker.patch(
+        "covalent_dispatcher._core.runner._executor_manager.get_executor",
+        return_value=mock_executor,
+    )
+
+    node_result = await _run_task(
+        result_object=result_object,
+        node_id=1,
+        inputs=inputs,
+        serialized_callable=None,
+        selected_executor=["local", {}],
+        call_before=[],
+        call_after=[],
+        node_name="task",
+        workflow_executor=["local", {}],
+    )
+
+    mock_executor._execute.assert_awaited_once()
+
+    assert node_result["stderr"] == "error"
+
+
+def test_post_process():
+    """Test post-processing of results."""
+
+    import covalent as ct
+
+    @ct.electron
+    def construct_cu_slab(x):
+        return x
+
+    @ct.electron
+    def compute_system_energy(x):
+        return x
+
+    @ct.electron
+    def construct_n_molecule(x):
+        return x
+
+    @ct.electron
+    def get_relaxed_slab(x):
+        return x
+
+    @ct.lattice
+    def compute_energy():
+        N2 = construct_n_molecule(1)
+        e_N2 = compute_system_energy(N2)
+
+        slab = construct_cu_slab(2)
+        e_slab = compute_system_energy(slab)
+
+        relaxed_slab = get_relaxed_slab(3)
+        e_relaxed_slab = compute_system_energy(relaxed_slab)
+
+        return (N2, e_N2, slab, e_slab, relaxed_slab, e_relaxed_slab)
+
+    compute_energy.build_graph()
+
+    node_outputs = {
+        "construct_n_molecule(0)": 1,
+        ":parameter:1(1)": 1,
+        "compute_system_energy(2)": 1,
+        "construct_cu_slab(3)": 2,
+        ":parameter:2(4)": 2,
+        "compute_system_energy(5)": 2,
+        "get_relaxed_slab(6)": 3,
+        ":parameter:3(7)": 3,
+        "compute_system_energy(8)": 3,
+    }
+
+    encoded_node_outputs = {
+        k: ct.TransportableObject.make_transportable(v) for k, v in node_outputs.items()
+    }
+
+    execution_result = _post_process(compute_energy, encoded_node_outputs)
+
+    assert execution_result == compute_energy()
+
+
+@pytest.mark.asyncio
+async def test_postprocess_workflow(mocker):
+    """Unit test for _postprocess_workflow"""
+
+    result_object = get_mock_result()
+    node_result = {"node_id": -1, "status": Result.COMPLETED, "output": 42}
+    failed_node_result = {"node_id": -1, "status": Result.FAILED, "stderr": "OOM", "error": None}
+    mock_run_task = mocker.patch(
+        "covalent_dispatcher._core.runner._run_task", return_value=node_result
+    )
+    mock_get_node_outputs = mocker.patch(
+        "covalent._results_manager.result.Result.get_all_node_outputs", return_value=[0]
+    )
+    mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
+
+    await _postprocess_workflow(result_object)
+    assert result_object._status == Result.COMPLETED
+    assert result_object._result == 42
+
+    mock_run_task = mocker.patch(
+        "covalent_dispatcher._core.runner._run_task", return_value=failed_node_result
+    )
+    await _postprocess_workflow(result_object)
+    assert result_object._status == Result.POSTPROCESSING_FAILED
+    assert "OOM" in result_object._error
+
+
+def test_build_sublattice_graph():
+    @ct.electron
+    def task(x):
+        return x
+
+    @ct.lattice
+    def workflow(x):
+        return task(x)
+
+    parent_metadata = {
+        "executor": "parent_executor",
+        "executor_data": {},
+        "workflow_executor": "my_postprocessor",
+        "workflow_executor_data": {},
+        "deps": {"bash": None, "pip": None},
+        "call_before": [],
+        "call_after": [],
+        "results_dir": None,
+    }
+
+    json_lattice = _build_sublattice_graph(workflow, parent_metadata, 1)
+    lattice = Lattice.deserialize_from_json(json_lattice)
+
+    assert list(lattice.transport_graph._graph.nodes) == [0, 1]
+    for k in lattice.metadata.keys():
+        # results_dir will be deprecated soon
+        if k != "results_dir":
+            assert parent_metadata[k] == lattice.metadata[k]
+
+
+@pytest.mark.asyncio
+async def test_dispatch_sublattice(test_db, mocker):
+    @ct.electron(executor="local")
+    def task(x):
+        return x
+
+    @ct.lattice(executor="local", workflow_executor="local")
+    def sub_workflow(x):
+        return task(x)
+
+    mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
+    mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
+    mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch")
+
+    result_object = get_mock_result()
+
+    serialized_callable = ct.TransportableObject(sub_workflow)
+    inputs = {"args": [ct.TransportableObject(2)], "kwargs": {}}
+
+    sub_dispatch_id = await _dispatch_sublattice(
+        parent_result_object=result_object,
+        parent_node_id=2,
+        parent_electron_id=1,
+        inputs=inputs,
+        serialized_callable=serialized_callable,
+        workflow_executor=["local", {}],
+    )
+    sub_result = get_result_object(sub_dispatch_id)
+    assert sub_result.dispatch_id == sub_dispatch_id
+
+    # check that sublattice inherits parent lattice's bash dep
+    sub_bash_dep = sub_result.lattice.metadata["deps"]["bash"]["attributes"]["commands"]
+    assert sub_bash_dep[0] == "ls"
+    assert sub_result._electron_id == 1
+
+    # check that sublattice's explicit bash dep overrides parent lattice's bash dep
+    sub_workflow.metadata["deps"]["bash"] = ct.DepsBash(["pwd"])
+    serialized_callable = ct.TransportableObject(sub_workflow)
+    sub_dispatch_id = await _dispatch_sublattice(
+        parent_result_object=result_object,
+        parent_node_id=2,
+        parent_electron_id=1,
+        inputs=inputs,
+        serialized_callable=serialized_callable,
+        workflow_executor=["local", {}],
+    )
+    sub_result = get_result_object(sub_dispatch_id)
+    assert sub_result.dispatch_id == sub_dispatch_id
+
+    # check that sublattice inherits parent lattice's bash dep
+    sub_bash_dep = sub_result.lattice.metadata["deps"]["bash"]["attributes"]["commands"]
+    assert sub_bash_dep[0] == "pwd"
+
+    # Check handling of invalid workflow executors
+
+    with pytest.raises(RuntimeError):
+        sub_dispatch_id = await _dispatch_sublattice(
+            parent_result_object=result_object,
+            parent_node_id=2,
+            parent_electron_id=1,
+            inputs=inputs,
+            serialized_callable=serialized_callable,
+            workflow_executor=["client", {}],
+        )
+
+    with pytest.raises(RuntimeError):
+        sub_dispatch_id = await _dispatch_sublattice(
+            parent_result_object=result_object,
+            parent_node_id=2,
+            parent_electron_id=1,
+            inputs=inputs,
+            serialized_callable=serialized_callable,
+            workflow_executor=["fake_executor", {}],
+        )
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 7829d589..f4633c66 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
@@ -50,6 +50,7 @@ from covalent_dispatcher._db.write_result_to_db import (
     insert_electrons_data,
     insert_lattices_data,
     load_file,
+    resolve_electron_id,
     store_file,
     update_electrons_data,
     update_lattice_completed_electron_num,
@@ -678,6 +679,39 @@ def test_write_sublattice_electron_id(test_db, mocker):
         assert rows[1].dispatch_id == "dispatch_2"
 
 
+def test_resolve_electron_id(test_db, mocker):
+    """Test looking up dispatch_id and node_id corresponding to an electron_id"""
+
+    mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
+    cur_time = dt.now(timezone.utc)
+    insert_lattices_data(
+        **get_lattice_kwargs(created_at=cur_time, updated_at=cur_time, started_at=cur_time)
+    )
+
+    # Create electron records.
+    electron_ids = []
+    cur_time = dt.now(timezone.utc)
+    for (name, node_id) in [
+        ("task_1", 0),
+        (":parameter:1", 1),
+        (":parameter:2", 2),
+        (":sublattice:task_2", 3),  # Sublattice node id
+        (":parameter:2", 4),
+    ]:
+        electron_kwargs = get_electron_kwargs(
+            name=name,
+            transport_graph_node_id=node_id,
+            created_at=cur_time,
+            updated_at=cur_time,
+        )
+        eid = insert_electrons_data(**electron_kwargs)
+        electron_ids.append(eid)
+
+    dispatch_id, node_id = resolve_electron_id(electron_ids[3])
+    assert dispatch_id == "dispatch_1"
+    assert node_id == 3
+
+
 def test_store_file_invalid_extension():
     """Test the function used to write data corresponding to the filenames in the DB."""
 
diff --git a/tests/covalent_dispatcher_tests/entry_point_test.py b/tests/covalent_dispatcher_tests/entry_point_test.py
index 3eb0a017..db3e28ba 100644
--- a/tests/covalent_dispatcher_tests/entry_point_test.py
+++ b/tests/covalent_dispatcher_tests/entry_point_test.py
@@ -40,13 +40,15 @@ def mock_initialize_result_object(lattice):
 
 @pytest.mark.asyncio
 async def test_run_dispatcher(mocker):
-    mocker.patch(
-        "covalent_dispatcher._core.initialize_result_object", mock_initialize_result_object
+    mock_run_dispatch = mocker.patch("covalent_dispatcher._core.run_dispatch")
+    mock_make_dispatch = mocker.patch(
+        "covalent_dispatcher._core.make_dispatch", return_value=DISPATCH_ID
     )
-    mock_run_workflow = mocker.patch("covalent_dispatcher._core.run_workflow")
-    dispatch_id = await run_dispatcher({"dispatch_id": DISPATCH_ID})
+    json_lattice = '{"workflow_function": "asdf"}'
+    dispatch_id = await run_dispatcher(json_lattice)
     assert dispatch_id == DISPATCH_ID
-    mock_run_workflow.assert_called_once()
+    mock_make_dispatch.assert_called_with(json_lattice)
+    mock_run_dispatch.assert_called_with(dispatch_id)
 
 
 def test_cancel_running_dispatch(mocker):
diff --git a/tests/covalent_tests/results_manager_tests/results_test.py b/tests/covalent_tests/results_manager_tests/results_test.py
index 8741cb6c..f0746a82 100644
--- a/tests/covalent_tests/results_manager_tests/results_test.py
+++ b/tests/covalent_tests/results_manager_tests/results_test.py
@@ -171,7 +171,7 @@ def test_result_post_process(
         return x
 
     @ct.lattice
-    def compute_energy():
+    def compute_energy(n):
         N2 = construct_n_molecule(1)
         e_N2 = compute_system_energy(N2)
 
@@ -180,10 +180,12 @@ def test_result_post_process(
 
         relaxed_slab = get_relaxed_slab(3)
         e_relaxed_slab = compute_system_energy(relaxed_slab)
+        for i in range(n):
+            pass
 
         return (N2, e_N2, slab, e_slab, relaxed_slab, e_relaxed_slab)
 
-    compute_energy.build_graph()
+    compute_energy.build_graph(3)
 
     compute_energy = LatticeClass.deserialize_from_json(compute_energy.serialize_to_json())
 
@@ -215,4 +217,12 @@ def test_result_post_process(
 
     execution_result = res.post_process()
 
-    assert execution_result == compute_energy()
+    assert execution_result == compute_energy(3)
+
+
+def test_update_node(result_1, mocker):
+    result_1._update_node(node_id=0, status=Result.COMPLETED, sub_dispatch_id="subdispatch")
+
+    tg = result_1.lattice.transport_graph
+    assert tg.get_node_value(0, "sub_dispatch_id") == "subdispatch"
+    assert tg.get_node_value(0, "status") == Result.COMPLETED
diff --git a/tests/covalent_tests/workflow/electron_test.py b/tests/covalent_tests/workflow/electron_test.py
index a8e35611..3e368a1a 100644
--- a/tests/covalent_tests/workflow/electron_test.py
+++ b/tests/covalent_tests/workflow/electron_test.py
@@ -213,3 +213,46 @@ def test_electron_metadata_is_serialized_early():
     assert node_0_metadata == encode_metadata(node_0_metadata)
     node_1_metadata = workflow.transport_graph.get_node_value(1, "metadata")
     assert node_1_metadata == encode_metadata(node_1_metadata)
+
+
+def test_autogen_list_electrons():
+    @ct.electron
+    def task(x):
+        return x
+
+    @ct.lattice
+    def workflow(x):
+        return task(x)
+
+    workflow.build_graph([5, 7])
+
+    g = workflow.transport_graph._graph
+
+    assert list(g.nodes) == [0, 1, 2, 3]
+    fn = g.nodes[1]["function"].get_deserialized()
+    assert fn(2, 5, 7) == [2, 5, 7]
+
+    assert g.nodes[2]["value"].get_deserialized() == 5
+    assert g.nodes[3]["value"].get_deserialized() == 7
+    assert set(g.edges) == set([(1, 0, 0), (2, 1, 0), (3, 1, 0)])
+
+
+def test_autogen_dict_electrons():
+    @ct.electron
+    def task(x):
+        return x
+
+    @ct.lattice
+    def workflow(x):
+        return task(x)
+
+    workflow.build_graph({"x": 5, "y": 7})
+
+    g = workflow.transport_graph._graph
+
+    assert list(g.nodes) == [0, 1, 2, 3]
+    fn = g.nodes[1]["function"].get_deserialized()
+    assert fn(x=2, y=5, z=7) == {"x": 2, "y": 5, "z": 7}
+    assert g.nodes[2]["value"].get_deserialized() == 5
+    assert g.nodes[3]["value"].get_deserialized() == 7
+    assert set(g.edges) == set([(1, 0, 0), (2, 1, 0), (3, 1, 0)])
diff --git a/tests/functional_tests/error_reporting_test.py b/tests/functional_tests/error_reporting_test.py
index 1d9abf31..9fc783e4 100644
--- a/tests/functional_tests/error_reporting_test.py
+++ b/tests/functional_tests/error_reporting_test.py
@@ -38,10 +38,6 @@ def failing_task_1():
     import sys
 
     print("Hello from Task 1", file=sys.stdout)
-    print("Hello from Task 1", file=sys.stderr)
-
-    # return 1
-    raise RuntimeError("Error 1")
 
 
 @ct.electron(executor="nonexistent")
@@ -74,12 +70,12 @@ def test_task_runtime_error_reporting():
 
     assert res.status == Result.FAILED
     tg = res.lattice.transport_graph
+    assert tg.get_node_value(0, "status") == Result.FAILED
     assert tg.get_node_value(0, "stdout") == "Hello from Task 0\n"
     assert tg.get_node_value(0, "stderr").startswith("Hello from Task 0\n")
     assert "RuntimeError" in tg.get_node_value(0, "stderr")
+    assert tg.get_node_value(1, "status") == Result.COMPLETED
     assert tg.get_node_value(1, "stdout") == "Hello from Task 1\n"
-    assert tg.get_node_value(1, "stderr").startswith("Hello from Task 1\n")
-    assert "RuntimeError" in tg.get_node_value(1, "stderr")
 
 
 def test_task_covalent_error_reporting():
diff --git a/tests/functional_tests/workflow_failure_test.py b/tests/functional_tests/workflow_failure_test.py
new file mode 100644
index 00000000..e56d8be2
--- /dev/null
+++ b/tests/functional_tests/workflow_failure_test.py
@@ -0,0 +1,68 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""Test how Covalent handles task failures in a workflow"""
+
+import time
+
+import covalent as ct
+
+
+def test_workflow_continues_after_node_failure():
+    """Check that Covalent continues to dispatch runnable tasks even
+    if an unrelated task fails.
+    """
+
+    @ct.electron
+    def failing_task():
+        assert False
+
+    @ct.electron
+    def double(x):
+        import time
+
+        time.sleep(1)
+        return 2 * x
+
+    # Graph consists of
+    # 0 (failing)
+    # 2 -> 1 -> 3 -> 4
+    @ct.lattice
+    def workflow(x):
+        failing_task()
+        return double(double(double(x)))
+
+    dispatch_id = ct.dispatch(workflow)(1)
+
+    time_to_wait = 8
+
+    # This returns after the first task fails
+    res = ct.get_result(dispatch_id, wait=True)
+    tg = res.lattice.transport_graph
+    status = tg.get_node_value(4, "status")
+
+    while str(status) != "COMPLETED" and time_to_wait > 0:
+        time_to_wait -= 1
+        time.sleep(1)
+        res = ct.get_result(dispatch_id, wait=True)
+        tg = res.lattice.transport_graph
+        status = tg.get_node_value(4, "status")
+
+    assert tg.get_node_value(4, "output").get_deserialized() == 8