Skip Mode

changes.d/6039.feat.md

@@ -0,0 +1 @@
+Added a new task run mode "skip", which allows skipping of tasks in a running workflow as well as creation of dummy tasks as part of workflow design.

cylc/flow/cfgspec/workflow.py

@@ -56,7 +56,10 @@
 from cylc.flow.platforms import (
     fail_if_platform_and_host_conflict, get_platform_deprecated_settings,
     is_platform_definition_subshell)
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_events_mgr import EventData
+from cylc.flow.run_modes import TASK_CONFIG_RUN_MODES
+
 
 # Regex to check whether a string is a command
 REC_COMMAND = re.compile(r'(`|\$\()\s*(.*)\s*([`)])$')
@@ -1334,6 +1337,28 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                     "[platforms][<platform name>]submission retry delays"
                 )
             )
+            Conf(
+                'run mode', VDR.V_STRING,
+                options=list(TASK_CONFIG_RUN_MODES),
+                default=RunMode.LIVE.value,
+                desc=f'''
+                    When the workflow is running in live mode, run this *task*
+                    in one of the following modes:
+
+                    ``{RunMode.LIVE.value}`` (default):
+                        {RunMode.LIVE.describe()}
+                    ``{RunMode.SKIP.value}``:
+                        {RunMode.SKIP.describe()}
+
+
+                    .. seealso::
+
+                       - :ref:`task-run-modes`
+                       - :cylc:conf:`flow.cylc[runtime][<namespace>][skip]`
+
+                    .. versionadded:: 8.4.0
+
+            ''')
             with Conf('meta', desc=r'''
                 Metadata for the task or task family.
 
@@ -1406,7 +1431,45 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                     determine how an event handler responds to task failure
                     events.
                 ''')
+            with Conf('skip', desc='''
+                Task configuration for :ref:`task-run-modes.skip`.
+
+                .. seealso::
 
+                   - :ref:`task-run-modes.skip`
+                   - :cylc:conf:`flow.cylc[runtime][<namespace>]run mode`
+
+                .. versionadded:: 8.4.0
+            '''):
+                Conf(
+                    'outputs',
+                    VDR.V_STRING_LIST,
+                    desc='''
+                        Outputs to be emitted by a task in skip mode.
+
+                        * By default, all required outputs will be generated
+                          plus succeeded if success is optional.
+                        * If skip-mode outputs is specified and does not
+                          include either succeeded or failed then succeeded
+                          will be produced.
+                        * The outputs submitted and started are always
+                          produced and do not need to be defined in here.
+
+                        .. versionadded:: 8.4.0
+                    '''
+                )
+                Conf(
+                    'disable task event handlers',
+                    VDR.V_BOOLEAN,
+                    default=True,
+                    desc='''
+                        Task event handlers are turned off by default for
+                        skip mode tasks. Changing this setting to ``False``
+                        will re-enable task event handlers.
+
+                        .. versionadded:: 8.4.0
+                    '''
+                )
             with Conf('simulation', desc='''
                 Task configuration for workflow *simulation* and *dummy* run
                 modes.

cylc/flow/commands.py

@@ -83,11 +83,9 @@
 from cylc.flow.log_level import log_level_to_verbosity
 from cylc.flow.network.schema import WorkflowStopMode
 from cylc.flow.parsec.exceptions import ParsecError
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_id import TaskID
-from cylc.flow.workflow_status import (
-    RunMode,
-    StopMode,
-)
+from cylc.flow.workflow_status import StopMode
 
 
 if TYPE_CHECKING:

cylc/flow/config.py

@@ -81,7 +81,8 @@
     is_relative_to,
 )
 from cylc.flow.task_qualifiers import ALT_QUALIFIERS
-from cylc.flow.simulation import configure_sim_modes
+from cylc.flow.run_modes.simulation import configure_sim_mode
+from cylc.flow.run_modes.skip import skip_mode_validate
 from cylc.flow.subprocctx import SubFuncContext
 from cylc.flow.task_events_mgr import (
     EventData,
@@ -98,6 +99,7 @@
     get_trigger_completion_variable_maps,
     trigger_to_completion_variable,
 )
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_trigger import TaskTrigger, Dependency
 from cylc.flow.taskdef import TaskDef
 from cylc.flow.unicode_rules import (
@@ -113,7 +115,6 @@
     WorkflowFiles,
     check_deprecation,
 )
-from cylc.flow.workflow_status import RunMode
 from cylc.flow.xtrigger_mgr import XtriggerCollator
 
 if TYPE_CHECKING:
@@ -512,9 +513,10 @@ def __init__(
 
         self.process_runahead_limit()
 
-        run_mode = self.run_mode()
+        run_mode = RunMode.get(self.options)
         if run_mode in {RunMode.SIMULATION, RunMode.DUMMY}:
-            configure_sim_modes(self.taskdefs.values(), run_mode)
+            for taskdef in self.taskdefs.values():
+                configure_sim_mode(taskdef.rtconfig, None, False)
 
         self.configure_workflow_state_polling_tasks()
 
@@ -566,6 +568,8 @@ def __init__(
 
         self.mem_log("config.py: end init config")
 
+        skip_mode_validate(self.taskdefs)
+
     @staticmethod
     def _warn_if_queues_have_implicit_tasks(
         config, taskdefs, max_warning_lines
@@ -1700,10 +1704,6 @@ def process_config_env(self):
                 ]
             )
 
-    def run_mode(self) -> str:
-        """Return the run mode."""
-        return RunMode.get(self.options)
-
     def _check_task_event_handlers(self):
         """Check custom event handler templates can be expanded.
 
@@ -2455,7 +2455,9 @@ def _get_taskdef(self, name: str) -> TaskDef:
 
         # Get the taskdef object for generating the task proxy class
         taskd = TaskDef(
-            name, rtcfg, self.run_mode(), self.start_point,
+            name,
+            rtcfg,
+            self.start_point,
             self.initial_point)
 
         # TODO - put all taskd.foo items in a single config dict

cylc/flow/data_messages.proto

@@ -128,6 +128,7 @@ message PbRuntime {
     optional string environment = 16;
     optional string outputs = 17;
     optional string completion = 18;
+    optional string run_mode = 19;
 }