diff --git a/taskflow/engines/action_engine/task_action.py b/taskflow/engines/action_engine/task_action.py
index 0bf61878..5ab7eeec 100644
--- a/taskflow/engines/action_engine/task_action.py
+++ b/taskflow/engines/action_engine/task_action.py
@@ -28,8 +28,6 @@ LOG = logging.getLogger(__name__)
 
 RESET_TASK_STATES = (states.PENDING,)
 SAVE_RESULT_STATES = (states.SUCCESS, states.FAILURE)
-ALREADY_FINISHED_STATES = (states.SUCCESS,)
-NEVER_RAN_STATES = (states.PENDING,)
 
 
 @contextlib.contextmanager
@@ -55,16 +53,11 @@ class TaskAction(base.Action):
     def uuid(self):
         return self._id
 
-    def _change_state(self, engine, state,
-                      result=None, progress=None, force=False):
+    def _change_state(self, engine, state, result=None, progress=None):
         """Update result and change state."""
         old_state = engine.storage.get_task_state(self.uuid)
-        state_check = states.check_task_transition(old_state, state)
-        if not force and not state_check:
-            # NOTE(harlowja): if we are forcing this state change, we don't
-            # care if the state transition should be ignored, if it's not being
-            # forced then we just ignore this state change.
-            return
+        if not states.check_task_transition(old_state, state):
+            return False
         if state in RESET_TASK_STATES:
             engine.storage.reset(self.uuid)
         if state in SAVE_RESULT_STATES:
@@ -74,6 +67,7 @@ class TaskAction(base.Action):
         if progress is not None:
             engine.storage.set_task_progress(self.uuid, progress)
         engine._on_task_state_change(self, state, result=result)
+        return True
 
     def _on_update_progress(self, task, event_data, progress, **kwargs):
         """Update task progress value that stored in engine."""
@@ -86,16 +80,17 @@ class TaskAction(base.Action):
             LOG.exception("Failed setting task progress for %s (%s) to %0.3f",
                           task, self.uuid, progress)
 
-    def _force_state(self, engine, state, progress, result=None):
-        self._change_state(engine, state,
-                           result=result, progress=progress, force=True)
+    def _change_state_update_task(self, engine, state, progress, result=None):
+        stated_changed = self._change_state(engine, state,
+                                            result=result, progress=progress)
+        if not stated_changed:
+            return False
         self._task.update_progress(progress)
+        return True
 
     def execute(self, engine):
-        if engine.storage.get_task_state(self.uuid) in ALREADY_FINISHED_STATES:
-            # Skip tasks that already finished.
+        if not self._change_state_update_task(engine, states.RUNNING, 0.0):
             return
-        self._force_state(engine, states.RUNNING, 0.0)
         with _autobind(self._task,
                        'update_progress', self._on_update_progress,
                        engine=engine):
@@ -106,15 +101,15 @@ class TaskAction(base.Action):
                 failure = misc.Failure()
                 self._change_state(engine, states.FAILURE, result=failure)
                 failure.reraise()
-        self._force_state(engine, states.SUCCESS, 1.0, result=result)
+        self._change_state_update_task(engine, states.SUCCESS, 1.0,
+                                       result=result)
 
     def revert(self, engine):
-        if engine.storage.get_task_state(self.uuid) in NEVER_RAN_STATES:
+        if not self._change_state_update_task(engine, states.REVERTING, 0.0):
             # NOTE(imelnikov): in all the other states, the task
             # execution was at least attempted, so we should give
             # task a chance for cleanup
             return
-        self._force_state(engine, states.REVERTING, 0.0)
         with _autobind(self._task,
                        'update_progress', self._on_update_progress,
                        engine=engine):
@@ -125,5 +120,5 @@ class TaskAction(base.Action):
             except Exception:
                 with excutils.save_and_reraise_exception():
                     self._change_state(engine, states.FAILURE)
-        self._force_state(engine, states.REVERTED, 1.0)
-        self._force_state(engine, states.PENDING, 0.0)
+        self._change_state_update_task(engine, states.REVERTED, 1.0)
+        self._change_state_update_task(engine, states.PENDING, 0.0)
diff --git a/taskflow/states.py b/taskflow/states.py
index 1ffed0b1..dcb65165 100644
--- a/taskflow/states.py
+++ b/taskflow/states.py
@@ -128,7 +128,7 @@ def check_flow_transition(old_state, new_state):
 ## Task state transitions
 # https://wiki.openstack.org/wiki/TaskFlow/States_of_Task_and_Flow#Task_States
 
-_ALLOWED_TASK_TRANSITIONS = [
+_ALLOWED_TASK_TRANSITIONS = frozenset((
     (PENDING, RUNNING),       # run it!
 
     (RUNNING, SUCCESS),       # the task finished successfully
@@ -141,17 +141,37 @@ _ALLOWED_TASK_TRANSITIONS = [
     (REVERTING, FAILURE),     # revert failed
 
     (REVERTED, PENDING),      # try again
-]
-_ALLOWED_TASK_TRANSITIONS.extend(
-    # NOTE(harlowja): the task was 'killed' while in one of the below 'a'
+
+    # NOTE(harlowja): allow the tasks to restart if in the same state
+    # as a they were in before as a task may be 'killed' while in one of the
+    # below states and it is permissible to let the task to re-enter that
+    # same state to try to finish
+    (REVERTING, REVERTING),
+    (RUNNING, RUNNING),
+
+    # NOTE(harlowja): the task was 'killed' while in one of the starting/ending
     # states and it is permissible to let the task to start running or
-    # reverting immediately
-    (a, b)
-    for a in (REVERTED, REVERTING, RUNNING)
-    for b in (RUNNING, REVERTING)
-    if a != b
+    # reverting again (if it really wants too)
+    (REVERTING, RUNNING),
+    (RUNNING, REVERTING),
+))
+
+_IGNORED_TASK_TRANSITIONS = [
+    (SUCCESS, RUNNING),       # already finished
+    (PENDING, REVERTING),     # never ran in the first place
+    (REVERTED, REVERTING),    # the task already reverted
+]
+
+# NOTE(harlowja): ignore transitions to the same state (in these cases).
+#
+# NOTE(harlowja): the above ALLOWED_TASK_TRANSITIONS does allow
+# transitions to certain equivalent states (but only for a few special
+# cases)
+_IGNORED_TASK_TRANSITIONS.extend(
+    (a, a) for a in (PENDING, FAILURE, SUCCESS, REVERTED)
 )
-_ALLOWED_TASK_TRANSITIONS = frozenset(_ALLOWED_TASK_TRANSITIONS)
+
+_IGNORED_TASK_TRANSITIONS = frozenset(_IGNORED_TASK_TRANSITIONS)
 
 
 def check_task_transition(old_state, new_state):
@@ -161,11 +181,11 @@ def check_task_transition(old_state, new_state):
     should be ignored, it returns False. If transition is not
     valid, it raises InvalidStateException.
     """
-    if old_state == new_state:
-        return False
     pair = (old_state, new_state)
     if pair in _ALLOWED_TASK_TRANSITIONS:
         return True
+    if pair in _IGNORED_TASK_TRANSITIONS:
+        return False
     # TODO(harlowja): Should we check/allow for 3rd party states to be
     # triggered during RUNNING by having a concept of a sub-state that we also
     # verify against??