Set: add implied outputs.

cylc/flow/scripts/set.py

@@ -29,8 +29,10 @@
 Setting outputs affects task completion and spawns downstream tasks that depend
 on those outputs.
 
-Implied outputs are set automatically: started implies submitted; succeeded and
-failed imply started; custom outputs and expired do not imply other outputs.
+Implied outputs are set automatically:
+  - succeeded and failed imply started
+  - started implies submitted
+  - custom outputs and expired do not imply other outputs
 
 Examples:
 
@@ -48,7 +50,7 @@
 
   # complete the ":file1" custom output of 3/bar:
   $ cylc set --out=file1 my-workflow//3/bar
-  # or via the associated message from the task definition:
+  # or use the associated output message from the task definition:
   $ cylc set --out="file 1 ready" my-workflow//3/bar
 
   # set multiple outputs at once:

cylc/flow/task_outputs.py

@@ -15,6 +15,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """Task output message manager and constants."""
 
+from typing import Iterable, Set
 
 # Standard task output strings, used for triggering.
 TASK_OUTPUT_EXPIRED = "expired"
@@ -48,6 +49,40 @@
 _IS_COMPLETED = 2
 
 
+def add_implied_outputs(outputs: Iterable[str]) -> Set[str]:
+    """Add implied outputs to a list of outputs.
+
+    - succeeded and failed imply started
+    - started implies submitted
+    - custom outputs and expired do not imply other outputs
+
+    Examples:
+        >>> sorted(add_implied_outputs(['cat', 'dog']))
+        ['cat', 'dog']
+
+        >>> sorted(add_implied_outputs(['cat', 'started']))
+        ['cat', 'started', 'submitted']
+
+        >>> sorted(add_implied_outputs(['cat', 'succeeded']))
+        ['cat', 'started', 'submitted', 'succeeded']
+
+        >>> sorted(add_implied_outputs(['cat', 'failed']))
+        ['cat', 'failed', 'started', 'submitted']
+
+    """
+    result = set(outputs)
+    if (
+        TASK_OUTPUT_SUCCEEDED in outputs or
+        TASK_OUTPUT_FAILED in outputs
+    ):
+        result.add(TASK_OUTPUT_STARTED)
+
+    if TASK_OUTPUT_STARTED in result:
+        result.add(TASK_OUTPUT_SUBMITTED)
+
+    return result
+
+
 class TaskOutputs:
     """Task output message manager.
 

cylc/flow/task_pool.py

@@ -68,6 +68,7 @@
 )
 from cylc.flow.wallclock import get_current_time_string
 from cylc.flow.platforms import get_platform
+from cylc.flow.task_outputs import add_implied_outputs
 from cylc.flow.task_queues.independent import IndepQueueManager
 
 from cylc.flow.flow_mgr import FLOW_ALL, FLOW_NONE, FLOW_NEW
@@ -1359,7 +1360,7 @@ def remove_if_complete(self, itask):
             return
 
         # Complete, can remove it from the pool.
-        self.remove(itask, 'complete')
+        self.remove(itask, 'completed')
 
         if itask.identity == self.stop_task_id:
             self.stop_task_finished = True
@@ -1626,7 +1627,7 @@ def _set_outputs(self, point, taskdef, outputs, flow_nums, flow_wait):
             # The parent task already exists in the pool.
             self.merge_flows(itask, flow_nums)
         else:
-            # Spawn a transient task instance to use.
+            # Spawn a transient task instance to use for spawning children.
             itask = self.spawn_task(
                 taskdef.name,
                 point,
@@ -1639,8 +1640,8 @@ def _set_outputs(self, point, taskdef, outputs, flow_nums, flow_wait):
             # submit a job. This will log task activity (e.g. event handlers)
             # in the previous-submit log directory.
 
-        # convert labels to messages, to send to task events manager.
-        for out in outputs:
+        # Convert labels to messages, to send to task events manager.
+        for out in add_implied_outputs(outputs):
             msg = itask.state.outputs.get_msg(out)
             if msg is None:
                 LOG.warning(f"{point}/{taskdef.name} has no output {out}")

tests/functional/clock-expire/00-basic/flow.cylc

@@ -23,7 +23,7 @@ Skip a daily post-processing workflow if the 'copy' task has expired."""
         # behind "now + 1 day". This makes the first two 'copy' tasks expire.
     [[graph]]
         P1D = """
-            model[-P1D] => model => copy => proc
+            model[-P1D] => model => copy? => proc
             copy:expired? => !proc
         """
 [runtime]