Implement Skip Mode

* Add `[runtime][<namespace>]run mode` and `[runtime][<namespace>][skip]`.
* Spin run mode functionality into separate modules.
* Run sim mode check with every main loop - we don't know if any tasks are
  in sim mode from the scheduler, but it doesn't cost much to check
  if none are.
* Implemented separate job "submission" pathway switching.
* Implemented skip mode, including output control logic.
* Add a linter and a validation check for tasks in nonlive modes,
  and for combinations of outputs
* Enabled setting outputs as if task ran in skip mode using
  `cylc set --out skip`.
* Testing for the above.

changes.d/6039.feat.md

@@ -0,0 +1 @@
+Allow setting of run mode on a task by task basis. Add a new mode "skip".

cylc/flow/cfgspec/workflow.py

@@ -1334,6 +1334,22 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                     "[platforms][<platform name>]submission retry delays"
                 )
             )
+            Conf(
+                'run mode', VDR.V_STRING,
+                options=['workflow', 'simulation', 'dummy', 'live', 'skip'],
+                default='workflow',
+                desc='''
+                    Override the workflow's run mode.
+
+                    By default workflows run in "live mode" - tasks run
+                    in the way defined by the runtime config.
+                    This setting allows individual tasks to be run using
+                    a different run mode.
+
+                    .. TODO: Reference updated documention.
+
+                    .. versionadded:: 8.4.0
+            ''')
             with Conf('meta', desc=r'''
                 Metadata for the task or task family.
 
@@ -1406,9 +1422,44 @@ 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 task :ref:`SkipMode`.
+
+                For a full description of skip run mode see
+                :ref:`SkipMode`.
+
+                .. versionadded:: 8.4.0
+            '''):
+                Conf(
+                    'outputs',
+                    VDR.V_STRING_LIST,
+                    desc='''
+                        Outputs to be emitted by a task in skip mode.
+
+                        By default started, submitted, succeeded and all
+                        required outputs will be emitted.
+
+                        If outputs are specified, but neither succeeded or
+                        failed are specified, succeeded will automatically be
+                        emitted.
 
+                        .. 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
+                Task configuration for *simulation* and *dummy* run
                 modes.
 
                 For a full description of simulation and dummy run modes see

cylc/flow/commands.py

@@ -77,14 +77,14 @@
 from cylc.flow.network.schema import WorkflowStopMode
 from cylc.flow.parsec.exceptions import ParsecError
 from cylc.flow.task_id import TaskID
-from cylc.flow.task_state import TASK_STATUSES_ACTIVE, TASK_STATUS_FAILED
-from cylc.flow.workflow_status import RunMode, StopMode
+from cylc.flow.task_state import (
+    TASK_STATUSES_ACTIVE, TASK_STATUS_FAILED, RunMode)
+from cylc.flow.workflow_status import StopMode
 
 from metomi.isodatetime.parsers import TimePointParser
 
 if TYPE_CHECKING:
     from cylc.flow.scheduler import Scheduler
-
     # define a type for command implementations
     Command = Callable[
         ...,

cylc/flow/config.py

@@ -82,7 +82,7 @@
 )
 from cylc.flow.print_tree import print_tree
 from cylc.flow.task_qualifiers import ALT_QUALIFIERS
-from cylc.flow.simulation import configure_sim_modes
+from cylc.flow.run_modes.nonlive import mode_validate_checks
 from cylc.flow.subprocctx import SubFuncContext
 from cylc.flow.task_events_mgr import (
     EventData,
@@ -114,8 +114,8 @@
     WorkflowFiles,
     check_deprecation,
 )
-from cylc.flow.workflow_status import RunMode
 from cylc.flow.xtrigger_mgr import XtriggerCollator
+from cylc.flow.task_state import RunMode
 
 if TYPE_CHECKING:
     from optparse import Values
@@ -513,10 +513,6 @@ def __init__(
 
         self.process_runahead_limit()
 
-        run_mode = self.run_mode()
-        if run_mode in {RunMode.SIMULATION, RunMode.DUMMY}:
-            configure_sim_modes(self.taskdefs.values(), run_mode)
-
         self.configure_workflow_state_polling_tasks()
 
         self._check_task_event_handlers()
@@ -567,6 +563,8 @@ def __init__(
 
         self.mem_log("config.py: end init config")
 
+        mode_validate_checks(self.taskdefs)
+
     @staticmethod
     def _warn_if_queues_have_implicit_tasks(
         config, taskdefs, max_warning_lines

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;
 }