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.

Schema: use `Enum` for task run mode instead of `String` (#61)

* Schema: use `Enum` for task run mode instead of `String`

* Tidy

fixup merge

fix broken functional test

Improve cylc set --out skip
* Improve documentation of feature in cylc set --help
* Allow cylc set --out=skip,optional_output
* Test the above

Remove test: We don't want users opting out of validating
[runtime][ns][simulation/skip] because we can now
changes these in a running workflow.

stop users opting out of validating workflows without validating ski/simulation taskdef sections

added tests for db entries in nonlive mode

ensure db entries for all four modes are correct.

move the change file toi the correct name

get localhost of platforms 'simulation' 'skip' or 'dummy' not defined. (They probably shouldn't be, but that's a site specific choice...)

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,30 +1422,63 @@ 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
                 :ref:`SimulationMode`.
             '''):
                 Conf('default run length', VDR.V_INTERVAL, DurationFloat(10),
                      desc='''
-                    The default simulated job run length.
-
                     Used if :cylc:conf:`flow.cylc[runtime][<namespace>]
                     execution time limit` **and**
                     :cylc:conf:`flow.cylc[runtime][<namespace>][simulation]
                     speedup factor` are not set.
                 ''')
                 Conf('speedup factor', VDR.V_FLOAT, desc='''
-                    Simulated run length = speedup factor * execution time
+                    Simulated run time = speedup factor * execution time
                     limit.
 
                     If :cylc:conf:`flow.cylc[runtime][<namespace>]
                     execution time limit` is set, the task
-                    simulated run length is computed by dividing it by this
+                    simulated run time is computed by dividing it by this
                     factor.
                 ''')
                 Conf('time limit buffer', VDR.V_INTERVAL, DurationFloat(30),
@@ -2118,6 +2167,10 @@ def upg(cfg, descr):
             ['runtime', '__MANY__', 'events', f"{old}s"],
             silent=cylc.flow.flags.cylc7_back_compat,
         )
+    u.deprecate(
+        '8.4.0',
+        ['runtime', '__MANY__', 'simulation', 'default run length'],
+        ['runtime', '__MANY__', 'simulation', 'default run length'])
 
     u.obsolete('8.0.0', ['cylc', 'events', 'abort on stalled'])
     u.obsolete('8.0.0', ['cylc', 'events', 'abort if startup handler fails'])

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