Fix handling of 'submitted' message, for manual set.

cylc/flow/task_events_mgr.py

@@ -758,19 +758,12 @@ def process_message(
         elif message == self.EVENT_SUBMITTED:
             if (
                     flag == self.FLAG_RECEIVED
-                    and itask.state.is_gt(TASK_STATUS_SUBMITTED)
+                    and itask.state.is_gte(TASK_STATUS_SUBMITTED)
             ):
                 # Already submitted.
                 return True
-            if (
-                itask.state.status == TASK_STATUS_PREPARING
-                or itask.tdef.run_mode == 'simulation'
-            ):
-                # If not in the preparing state we already assumed and handled
-                # job submission under the started event above...
-                # (sim mode does not have the job prep state)
-                self._process_message_submitted(itask, event_time, forced)
-                self.spawn_children(itask, TASK_OUTPUT_SUBMITTED)
+            self._process_message_submitted(itask, event_time, forced)
+            self.spawn_children(itask, TASK_OUTPUT_SUBMITTED)
 
             # ... but either way update the job ID in the job proxy (it only
             # comes in via the submission message).

cylc/flow/task_state.py

@@ -460,6 +460,11 @@ def is_gt(self, status):
         return (TASK_STATUSES_ORDERED.index(self.status) >
                 TASK_STATUSES_ORDERED.index(status))
 
+    def is_gte(self, status):
+        """"Return True if self.status >= status."""
+        return (TASK_STATUSES_ORDERED.index(self.status) >=
+                TASK_STATUSES_ORDERED.index(status))
+
     def _add_prerequisites(self, point, tdef):
         """Add task prerequisites."""
         # Triggers for sequence_i only used if my cycle point is a

tests/unit/test_task_state.py

@@ -21,9 +21,12 @@
 from cylc.flow.task_trigger import Dependency, TaskTrigger
 from cylc.flow.task_state import (
     TaskState,
+    TASK_STATUS_PREPARING,
+    TASK_STATUS_SUBMIT_FAILED,
+    TASK_STATUS_SUBMITTED,
     TASK_STATUS_SUCCEEDED,
-    TASK_STATUS_FAILED,
     TASK_STATUS_WAITING,
+    TASK_STATUS_RUNNING,
 )
 
 
@@ -100,3 +103,19 @@ def test_task_prereq_duplicates(set_cycling_type):
     prereqs = [p.satisfied for p in tstate.prerequisites]
 
     assert prereqs == [{("1", "a", "succeeded"): False}]
+
+
+def test_task_state_order():
+    """Test is_gt and is_gte methods."""
+
+    tdef = TaskDef('foo', {}, 'live', IntegerPoint("1"), IntegerPoint("1"))
+    tstate = TaskState(tdef, IntegerPoint("1"), TASK_STATUS_SUBMITTED, False)
+
+    assert tstate.is_gt(TASK_STATUS_WAITING)
+    assert tstate.is_gt(TASK_STATUS_PREPARING)
+    assert tstate.is_gt(TASK_STATUS_SUBMIT_FAILED)
+    assert not tstate.is_gt(TASK_STATUS_SUBMITTED)
+    assert tstate.is_gte(TASK_STATUS_SUBMITTED)
+    assert not tstate.is_gt(TASK_STATUS_RUNNING)
+    assert not tstate.is_gte(TASK_STATUS_RUNNING)
+