handle restart/reload/remove

cylc/flow/task_pool.py

@@ -1048,6 +1048,14 @@ def reload_taskdefs(self, config: 'WorkflowConfig') -> None:
                     self.check_task_output,
                 )
                 self._swap_out(new_task)
+                # Set xtrigger checking type for parentless spawning.
+                if (
+                    new_task.tdef.is_parentless(new_task.point)
+                    and set(new_task.state.xtriggers.keys()).intersection(
+                        self.xtrigger_mgr.sequential_xtrigger_labels
+                    )
+                ):
+                    new_task.is_xtrigger_sequential = True
                 self.data_store_mgr.delta_task_prerequisite(new_task)
                 LOG.info(f"[{itask}] reloaded task definition")
                 if itask.state(*TASK_STATUSES_ACTIVE):
@@ -1705,6 +1713,13 @@ def remove_tasks(self, items):
         """Remove tasks from the pool."""
         itasks, _, bad_items = self.filter_task_proxies(items)
         for itask in itasks:
+            # Spawn next occurance of xtrigger sequential task.
+            if itask.is_xtrigger_sequential:
+                self.spawn_to_rh_limit(
+                    itask.tdef,
+                    itask.tdef.next_point(itask.point),
+                    itask.flow_nums
+                )
             self.remove(itask, 'request')
         if self.compute_runahead():
             self.release_runahead_tasks()