Start-tasks: wait on xtriggers

changes.d/6528.fix.md

@@ -0,0 +1 @@
+Make start-tasks wait on xtriggers (see "cylc play --start-task"). 

cylc/flow/scheduler.py

@@ -837,8 +837,10 @@ def _load_pool_from_tasks(self):
         """Load task pool with specified tasks, for a new run."""
         LOG.info(f"Start task: {self.options.starttask}")
         # flow number set in this call:
-        self.pool.force_trigger_tasks(
+        self.pool.set_prereqs_and_outputs(
             self.options.starttask,
+            outputs=[],
+            prereqs=["all"],
             flow=[FLOW_NEW],
             flow_descr=f"original flow from {self.options.starttask}"
         )

cylc/flow/scheduler_cli.py

@@ -196,12 +196,13 @@
     OptionSettings(
         ["--start-task", "--starttask", "-t"],
         help=(
-            "Start from this task instance, given by '<cycle>/<name>'."
-            " This can be used multiple times to start from multiple"
-            " tasks at once. Dependence on tasks with cycle points earlier"
-            " than the earliest start-task will be ignored. A"
-            " sub-graph of the workflow will run if selected tasks do"
-            " not lead on to the full graph."),
+            "Start from this task instance instead of the beginning of the"
+            " graph. Dependence on other tasks is ignored but clock and"
+            " xtriggers are respected. Reuse the option for multiple start"
+            " tasks. Any dependence on cycle points prior to the earliest"
+            " start task will be ignored throughout the graph. Check that"
+            " start tasks flow into the full downstream graph, if needed."
+        ),
         metavar="TASK_ID",
         action='append',
         dest="starttask",

tests/integration/test_task_pool.py

@@ -2305,3 +2305,80 @@ async def test_job_insert_on_crash(one_conf, flow, scheduler, start):
 
         # and a job entry should be added
         assert len(task_1.jobs) == 1
+
+
+async def test_start_tasks(
+    flow,
+    scheduler,
+    start,
+    log_filter,
+    capture_submission,
+):
+    """Check starting from "start-tasks" with and without clock-triggers.
+
+    """
+    id_ = flow(
+        {
+            'scheduler': {
+                'cycle point format': '%Y',
+            },
+            'scheduling': {
+                'initial cycle point': '2040',
+                'runahead limit': 'P0Y',
+                'xtriggers': {
+                    'wall_clock_satisfied': "wall_clock(offset='-P100Y')"
+                },
+                'graph': {
+                    'P1Y': """
+                        foo
+                        @wall_clock => bar
+                        @wall_clock_satisfied => baz
+                        qux
+                    """
+                }
+            }
+        }
+    )
+    schd = scheduler(
+        id_,
+        starttask=['2050/foo', '2050/bar', '2050/baz'],
+        paused_start=False
+    )
+
+    async with start(schd) as log:
+        # capture any job submissions
+        submitted_tasks = capture_submission(schd)
+        assert submitted_tasks == set()
+
+        # It should start up with:
+        # - 2050/foo and 2051/foo (spawned to runahead limit)
+        # - 2050/bar waiting on its (unsatisfied) clock-trigger
+        # - 2050/baz waiting on its (satisfied) clock-trigger
+        # - no qux instances (not listed as a start-task)
+        itasks = schd.pool.get_tasks()
+        assert (
+            set(itask.identity for itask in itasks) == set([
+                "2050/foo",
+                "2051/foo",
+                "2050/bar",
+                "2050/baz",
+            ])
+        )
+
+        # Check xtriggers
+        for itask in itasks:
+            schd.pool.xtrigger_mgr.call_xtriggers_async(itask)
+            schd.pool.rh_release_and_queue(itask)
+
+        # Release tasks that are ready to run.
+        schd.release_tasks_to_run()
+
+        # It should submit 2050/foo, 2051/foo, 2050/baz
+        # It should not submit 2050/bar (waiting on clock trigger)
+        assert (
+            set(itask.identity for itask in submitted_tasks) == set([
+                "2050/foo",
+                "2051/foo",
+                "2050/baz",
+            ])
+        )