diff --git a/parsl/dataflow/dflow.py b/parsl/dataflow/dflow.py
index <HASH>..<HASH> 100644
--- a/parsl/dataflow/dflow.py
+++ b/parsl/dataflow/dflow.py
@@ -195,7 +195,8 @@ class DataFlowKernel(object):
                 self.checkpoint(tasks=[task_id])
 
         # Submit _*_stage_out tasks for output data futures that correspond with remote files
-        if self.tasks[task_id]['executor'] != 'data_manager':
+        if (self.tasks[task_id]['status'] == States.done and
+            self.tasks[task_id]['executor'] != 'data_manager'):
             for dfu in self.tasks[task_id]['app_fu'].outputs:
                 f = dfu.file_obj
                 if isinstance(f, File) and f.is_remote():
Check if a task is done before staging out output files