diff --git a/cylc/flow/task_pool.py b/cylc/flow/task_pool.py
index ccc336286e..3c99a5e0af 100644
--- a/cylc/flow/task_pool.py
+++ b/cylc/flow/task_pool.py
@@ -244,20 +244,6 @@ def _swap_out(self, itask):
             self.active_tasks[itask.point][itask.identity] = itask
             self.active_tasks_changed = True
 
-    def spawn_to_runahead_limit(self):
-        """Spawn the task pool out to the runahead limit in one go.
-
-        Not strictly necessary, it will spawn ahead per main loop iteration,
-        but useful back-compat for tests that expect this prior to
-        https://github.com/cylc/cylc-flow/pull/7237
-
-        """
-        self.compute_runahead()
-        # Arbitrary limit to avoid infinite loop if something goes wrong.
-        for _ in range(10):
-            if not self.release_runahead_tasks():
-                break
-
     def queue_if_ready(self, itask: 'TaskProxy') -> None:
         """Queue itask if it is ready to run.
 
@@ -301,12 +287,6 @@ def load_from_point(self):
                 ntask, _ = self.get_or_spawn_task(point, tdef, flow_nums)
                 if ntask is not None:
                     self.add_to_pool(ntask)
-        # Spawning to the runahead limit immediately is not strictly necessary
-        # as it would occur over several scheduler main loop iterations; we do
-        # it mainly for compatibility with integration tests pre PR #7237.
-        self.spawn_to_runahead_limit()
-        for itask in self.get_tasks():
-            self.queue_if_ready(itask)
 
     def db_add_new_flow_rows(self, itask: TaskProxy) -> None:
         """Add new rows to DB task tables that record flow_nums.
diff --git a/tests/integration/README.md b/tests/integration/README.md
index 1a776232a2..b1d1761515 100644
--- a/tests/integration/README.md
+++ b/tests/integration/README.md
@@ -78,8 +78,10 @@ Workflows can be run in two ways:
 
 ```
 with start(schd):
-    # starts the Scheduler but does not start the main loop
-    # (always the better option if its possible)
+    # Starts the Scheduler but does not start the main loop
+    #   (always the better option if its possible).
+    # Use the spawn_ahead fixture if you need to examine how
+    #   parentless tasks advance (without iterating the main loop).
     ...
 
 with run(schd):
@@ -92,3 +94,10 @@ These methods both shut down the workflow / clean up after themselves.
 
 It is necessary to shut down workflows correctly to clean up resorces and
 running tasks.
+
+Note at start-up tasks states remain as as initially loaded until evolved by
+the main loop (or otherwise by direct manipulation in tests).
+
+Parentless tasks can spawn freely to the runahead limit, but that happens
+incremently via main loop iterations. Use the spawn_ahead fixture to do the
+same thing without running the main loop in tests.
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 228ceed49a..41e8d94399 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -777,3 +777,35 @@ def get_submissions():
         }
 
     return get_submissions
+
+
+def _spawn_ahead(task_pool):
+    """ Release runahead tasks and queue any that are ready to run.
+
+    Use to make parentless tasks spawn ahead without the main loop.
+
+    """
+    n_limit = 10  # arbitrary safety net
+    count = 0
+    while True:
+        count += 1
+        task_pool.compute_runahead()
+        if not task_pool.release_runahead_tasks():
+            break
+        for itask in task_pool.get_tasks():
+            task_pool.queue_if_ready(itask)
+        if count > n_limit:
+            raise AssertionError(
+                f"Exceeded {n_limit} iterations"
+                " waiting for task pool to settle"
+            )
+
+
+@pytest.fixture
+def spawn_ahead():
+    return _spawn_ahead
+
+
+@pytest.fixture(scope='module')
+def mod_spawn_ahead():
+    return _spawn_ahead
diff --git a/tests/integration/run_modes/test_mode_overrides.py b/tests/integration/run_modes/test_mode_overrides.py
index fa148c3075..22f3a8f47e 100644
--- a/tests/integration/run_modes/test_mode_overrides.py
+++ b/tests/integration/run_modes/test_mode_overrides.py
@@ -135,7 +135,7 @@ async def test_run_mode_skip_abides_by_held(flow, scheduler, run):
 
 
 async def test_run_mode_override_from_broadcast(
-    flow, scheduler, start, complete, log_filter, capture_live_submissions
+    flow, scheduler, start, capture_live_submissions, spawn_ahead
 ):
     """Test that run_mode modifications only apply to one task.
     """
@@ -155,6 +155,8 @@ async def test_run_mode_override_from_broadcast(
         schd.broadcast_mgr.put_broadcast(
             ['1000'], ['foo'], [{'run mode': 'skip'}])
 
+        spawn_ahead(schd.pool)
+
         foo_1000 = schd.pool.get_task(ISO8601Point('1000'), 'foo')
         foo_1001 = schd.pool.get_task(ISO8601Point('1001'), 'foo')
 
diff --git a/tests/integration/run_modes/test_simulation.py b/tests/integration/run_modes/test_simulation.py
index 615fa9fc49..dfcdbc996e 100644
--- a/tests/integration/run_modes/test_simulation.py
+++ b/tests/integration/run_modes/test_simulation.py
@@ -90,7 +90,7 @@ def _run_simjob(schd, point, task):
 
 @pytest.fixture(scope='module')
 async def sim_time_check_setup(
-    mod_flow, mod_scheduler, mod_start, mod_one_conf,
+    mod_flow, mod_scheduler, mod_start, mod_spawn_ahead
 ):
     schd = mod_scheduler(mod_flow({
         'scheduler': {'cycle point format': '%Y'},
@@ -129,6 +129,7 @@ async def sim_time_check_setup(
         }
     }))
     async with mod_start(schd):
+        mod_spawn_ahead(schd.pool)
         itasks = schd.pool.get_tasks()
         [schd.task_job_mgr._set_retry_timers(i) for i in itasks]
         yield schd, itasks
@@ -170,7 +171,7 @@ def test_false_if_not_running(
             id='fail-no-submits'),
     )
 )
-def test_fail_once(sim_time_check_setup, itask, point, results, monkeypatch):
+def test_fail_once(sim_time_check_setup, itask, point, results):
     """A task with a fail cycle point only fails
     at that cycle point, and then only on the first submission.
     """
diff --git a/tests/integration/run_modes/test_skip.py b/tests/integration/run_modes/test_skip.py
index 814ea5d2a6..6ba4a1016e 100644
--- a/tests/integration/run_modes/test_skip.py
+++ b/tests/integration/run_modes/test_skip.py
@@ -168,7 +168,7 @@ async def test_skip_mode_outputs(
 
 
 async def test_doesnt_release_held_tasks(
-    one_conf, flow, scheduler, start, log_filter, capture_live_submissions
+    one_conf, flow, scheduler, start, log_filter, spawn_ahead
 ):
     """Point 5 of the proposal
     https://github.com/cylc/cylc-admin/blob/master/docs/proposal-skip-mode.md
@@ -190,6 +190,7 @@ async def test_doesnt_release_held_tasks(
         assert not log_filter(contains='=> succeeded'), msg.format('succeed')
 
         # Release held task and assert that it now skips successfully:
+        spawn_ahead(schd.pool)
         schd.pool.release_held_tasks({TaskTokens('1', 'one')})
         schd.release_tasks_to_run()
 
diff --git a/tests/integration/scripts/test_dump.py b/tests/integration/scripts/test_dump.py
index 8746bc1ef8..ecaa420268 100644
--- a/tests/integration/scripts/test_dump.py
+++ b/tests/integration/scripts/test_dump.py
@@ -29,7 +29,7 @@
 DumpOptions = Options(get_option_parser())
 
 
-async def test_dump_tasks(flow, scheduler, start):
+async def test_dump_tasks(flow, scheduler, start, spawn_ahead):
     """It should show n=0 tasks.
 
     See: https://github.com/cylc/cylc-flow/pull/5600
@@ -46,7 +46,8 @@ async def test_dump_tasks(flow, scheduler, start):
     })
     schd = scheduler(id_)
     async with start(schd):
-        # schd.release_tasks_to_run()
+        spawn_ahead(schd.pool)
+        schd.release_tasks_to_run()
         await schd.update_data_structure()
         ret = []
         await dump(
diff --git a/tests/integration/scripts/test_set.py b/tests/integration/scripts/test_set.py
index 1f6f018fe4..905c24696a 100644
--- a/tests/integration/scripts/test_set.py
+++ b/tests/integration/scripts/test_set.py
@@ -70,6 +70,7 @@ async def test_set_parentless_spawning(
     scheduler,
     run,
     complete,
+    spawn_ahead
 ):
     """Ensure that setting outputs does not interfere with parentless spawning.
 
@@ -96,7 +97,8 @@ async def test_set_parentless_spawning(
             ['1'],
         )
 
-        schd.pool.spawn_to_runahead_limit()
+        spawn_ahead(schd.pool)
+
         # the parentless task "a" should be spawned out to the runahead limit
         assert schd.pool.get_task_ids() == {'2/a', '3/a'}
 
diff --git a/tests/integration/test_compat_mode.py b/tests/integration/test_compat_mode.py
deleted file mode 100644
index 86719c5621..0000000000
--- a/tests/integration/test_compat_mode.py
+++ /dev/null
@@ -1,131 +0,0 @@
-# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
-# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-"""Tests for Cylc 7 compatibility mode."""
-
-from typing import TYPE_CHECKING
-
-from cylc.flow.cycling.integer import IntegerPoint
-from cylc.flow.data_store_mgr import TASK_PROXIES
-
-if TYPE_CHECKING:
-    from cylc.flow.scheduler import Scheduler
-
-
-async def test_blocked_tasks_in_n0(flow, scheduler, run, complete):
-    """Ensure that tasks with no satisfied dependencies remain in the pool.
-
-    In this example, the "recover" task is not satisfiable because its upstream
-    dependency "foo:failed" will never be satisfied. The unsatisfiable
-    "recover" task should remain in n=0 until removed/completed.
-
-    See https://github.com/cylc/cylc-flow/issues/4983
-    """
-    id_ = flow(
-        {
-            'scheduling': {
-                'initial cycle point': '1',
-                'cycling mode': 'integer',
-                'runahead limit': 'P2',
-                'dependencies': {
-                    'P1': {
-                        'graph': '''
-                            foo:fail => recover
-                            foo | recover => bar
-                        ''',
-                    },
-                },
-            },
-        },
-        filename='suite.rc',
-    )
-    schd: 'Scheduler' = scheduler(id_, paused_start=False, debug=True)
-    async with run(schd):
-        # the workflow should run for three cycles, then runahead stall
-        await complete(schd, *(f'{cycle}/bar' for cycle in range(1, 4)))
-
-        # (complete happens before stall test)
-        await schd._main_loop()
-        assert schd.is_stalled
-
-        # the "blocked" recover tasks should remain in the pool
-        assert schd.pool.get_task_ids() == {
-            '1/recover',
-            '2/recover',
-            '3/recover',
-            '4/foo',
-        }
-
-        # the "blocked" tasks should remain visible in the data store
-        assert {
-            (x.cycle_point, x.graph_depth, x.name)
-            for x in schd.data_store_mgr.data[schd.tokens.id][
-                TASK_PROXIES
-            ].values()
-        } == {
-            ('1', 1, 'foo'),
-            ('1', 0, 'recover'),
-            ('1', 1, 'bar'),
-            ('2', 1, 'foo'),
-            ('2', 0, 'recover'),
-            ('2', 1, 'bar'),
-            ('3', 1, 'foo'),
-            ('3', 0, 'recover'),
-            ('3', 1, 'bar'),
-            ('4', 0, 'foo'),
-            ('4', 1, 'recover'),
-            ('4', 1, 'bar'),
-        }
-
-        # remove the unsatisfiable tasks
-        # (i.e. manually implement a suicide trigger)
-        for cycle in range(1, 4):
-            itask = schd.pool.get_task(IntegerPoint(str(cycle)), 'recover')
-            schd.pool.remove(itask, 'suicide-trigger')
-
-        schd.pool.spawn_to_runahead_limit()
-        assert schd.pool.get_task_ids() == {
-            '4/foo',
-            '5/foo',
-            '6/foo',
-            '7/foo',
-        }
-
-        # the workflow continue into the next three cycles, then stall again
-        # (i.e. the runahead limit should move forward after the removes)
-        await complete(schd, *(f'{cycle}/bar' for cycle in range(4, 7)))
-        await schd._main_loop()
-        assert schd.is_stalled
-
-        assert {
-            (x.cycle_point, x.graph_depth, x.name)
-            for x in schd.data_store_mgr.data[schd.tokens.id][
-                TASK_PROXIES
-            ].values()
-        } == {
-            ('4', 1, 'foo'),
-            ('4', 0, 'recover'),
-            ('4', 1, 'bar'),
-            ('5', 1, 'foo'),
-            ('5', 0, 'recover'),
-            ('5', 1, 'bar'),
-            ('6', 1, 'foo'),
-            ('6', 0, 'recover'),
-            ('6', 1, 'bar'),
-            ('7', 0, 'foo'),
-            ('7', 1, 'recover'),
-            ('7', 1, 'bar'),
-        }
diff --git a/tests/integration/test_data_store_mgr.py b/tests/integration/test_data_store_mgr.py
index 78d963f52c..8e7ac5bf8b 100644
--- a/tests/integration/test_data_store_mgr.py
+++ b/tests/integration/test_data_store_mgr.py
@@ -631,7 +631,9 @@ def test_delta_task_xtrigger(xharness):
     assert xtrig_y.satisfied
 
 
-async def test_absolute_graph_edges(flow, scheduler, start):
+async def test_absolute_graph_edges(
+    flow, scheduler, start, spawn_ahead
+):
     """It should add absolute graph edges to the store.
 
     See: https://github.com/cylc/cylc-flow/issues/5845
@@ -651,6 +653,7 @@ async def test_absolute_graph_edges(flow, scheduler, start):
     schd = scheduler(id_)
 
     async with start(schd):
+        spawn_ahead(schd.pool)
         await schd.update_data_structure()
 
         assert {
diff --git a/tests/integration/test_force_trigger.py b/tests/integration/test_force_trigger.py
index 2598cffd9e..32bcfdb090 100644
--- a/tests/integration/test_force_trigger.py
+++ b/tests/integration/test_force_trigger.py
@@ -835,6 +835,7 @@ async def test_pre_warm_start_interraction_with_auto_restart(
     scheduler,
     start,
     log_filter,
+    spawn_ahead
 ):
     """Test interaction between warm-start pre-startcp tasks and auto-restart.
 
@@ -858,6 +859,7 @@ async def test_pre_warm_start_interraction_with_auto_restart(
     async with start(schd):
         # trigger a pre-initial task (pre start-cycle point)
         await run_cmd(force_trigger_tasks(schd, ['^/foo'], []))
+        spawn_ahead(schd.pool)
         assert schd.pool.pre_start_tasks_to_trigger
 
         # configure the workflow to auto-retsart
diff --git a/tests/integration/test_id_match.py b/tests/integration/test_id_match.py
index f6f7fad13d..94891c2630 100644
--- a/tests/integration/test_id_match.py
+++ b/tests/integration/test_id_match.py
@@ -24,7 +24,9 @@
     from cylc.flow.id import TaskTokens
 
 
-async def test_id_match(flow, scheduler, start, caplog):
+async def test_id_match(
+    flow, scheduler, start, caplog, spawn_ahead
+):
     id_ = flow({
         'scheduling': {
             'cycling mode': 'integer',
@@ -69,6 +71,7 @@ def match(*ids: str) -> Tuple[Set[str], Set[str]]:
                 schd, ['1/b2'], ['1'], ['failed'], None
             )
         )
+        spawn_ahead(schd.pool)
 
         # task pool state:
         # * cycle 1
diff --git a/tests/integration/test_optional_outputs.py b/tests/integration/test_optional_outputs.py
index 2914a4d055..3d4067813b 100644
--- a/tests/integration/test_optional_outputs.py
+++ b/tests/integration/test_optional_outputs.py
@@ -425,6 +425,9 @@ async def test_clock_expiry(
         one = schd.pool.get_task(ISO8601Point('20000101T0000Z'), 'x')
         assert one
 
+        # spawn the second one
+        schd.pool.spawn_next_parentless(one)
+
         # the second task (preparing)
         two = schd.pool.get_task(ISO8601Point('20010101T0000Z'), 'x')
         assert two
diff --git a/tests/integration/test_queues.py b/tests/integration/test_queues.py
index d00667fe92..2d01c24018 100644
--- a/tests/integration/test_queues.py
+++ b/tests/integration/test_queues.py
@@ -60,6 +60,7 @@ async def test_queue_release(
     capture_submission,
     start_paused,
     queue_limit,
+    spawn_ahead
 ):
     """Tasks should be released up to the limit if the scheduler is not paused.
 
@@ -80,7 +81,7 @@ async def test_queue_release(
         # release runahead/queued tasks
         # (if scheduler is paused we should not have any submissions)
         # (otherwise a number of tasks up to the limit should be released)
-        schd.pool.release_runahead_tasks()
+        spawn_ahead(schd.pool)
         schd.release_tasks_to_run()
         assert len(submitted_tasks) == expected_submissions
 
@@ -94,7 +95,8 @@ async def test_queue_release(
 async def test_queue_held_tasks(
     param_workflow,
     start,
-    capture_submission
+    capture_submission,
+    spawn_ahead
 ):
     """Held tasks should not be released from queues.
 
@@ -106,6 +108,7 @@ async def test_queue_held_tasks(
     schd: Scheduler = param_workflow(paused_start=True, queue_limit=1)
 
     async with start(schd):
+        spawn_ahead(schd.pool)
         # capture task submissions (prevents real submissions)
         submitted_tasks = capture_submission(schd)
 
diff --git a/tests/integration/test_reload.py b/tests/integration/test_reload.py
index 30961706c0..007a3e183e 100644
--- a/tests/integration/test_reload.py
+++ b/tests/integration/test_reload.py
@@ -38,6 +38,7 @@ async def test_reload_waits_for_pending_tasks(
     start,
     monkeypatch,
     log_scan,
+    spawn_ahead
 ):
     """Reload should flush out preparing tasks and pause the workflow.
 
@@ -83,6 +84,7 @@ def submit_task_jobs(*a, **k):
         assert foo.state(TASK_STATUS_WAITING)
 
         # put the task into the preparing state
+        spawn_ahead(schd.pool)
         schd.release_tasks_to_run()
         assert foo.state(TASK_STATUS_PREPARING)
 
@@ -116,6 +118,7 @@ async def test_reload_waits_for_preparing_tasks_in_remote_init(
     scheduler,
     start,
     monkeypatch,
+    spawn_ahead
 ):
     """Reload should wait for preparing tasks stuck in remote-init.
 
@@ -166,6 +169,7 @@ def submit_task_jobs(itasks, *a, **k):
         )
 
         # Put the task into the preparing state
+        spawn_ahead(schd.pool)
         schd.release_tasks_to_run()
         assert foo.state(TASK_STATUS_PREPARING)
         assert foo.waiting_on_job_prep
diff --git a/tests/integration/test_scheduler.py b/tests/integration/test_scheduler.py
index ca358a1f79..39e65b895f 100644
--- a/tests/integration/test_scheduler.py
+++ b/tests/integration/test_scheduler.py
@@ -154,6 +154,7 @@ async def test_holding_tasks_whilst_scheduler_paused(
     one_conf,
     start,
     scheduler,
+    spawn_ahead
 ):
     """It should hold tasks irrespective of workflow state.
 
@@ -164,6 +165,7 @@ async def test_holding_tasks_whilst_scheduler_paused(
 
     # run the workflow
     async with start(one):
+        spawn_ahead(one.pool)
         # capture any job submissions
         submitted_tasks = capture_submission(one)
         assert submitted_tasks == set()
diff --git a/tests/integration/test_sequential_xtriggers.py b/tests/integration/test_sequential_xtriggers.py
index 6a30e92b0e..8ccbf85f7d 100644
--- a/tests/integration/test_sequential_xtriggers.py
+++ b/tests/integration/test_sequential_xtriggers.py
@@ -154,14 +154,24 @@ async def test_reload(sequential, start):
 
 @pytest.mark.parametrize('is_sequential', [True, False])
 @pytest.mark.parametrize('xtrig_def', [
-    'wall_clock(sequential={})',
-    'wall_clock(PT1H, sequential={})',
-    'xrandom(1, 1, sequential={})',
+    'xrandom(0, 0, sequential={})',
+    'wall_clock(P100Y, sequential={})',
 ])
 async def test_sequential_arg_ok(
-    flow, scheduler, start, xtrig_def: str, is_sequential: bool
+    flow,
+    scheduler,
+    start,
+    xtrig_def: str,
+    is_sequential: bool,
+    spawn_ahead
 ):
-    """Test passing the sequential argument to xtriggers."""
+    """Test passing the sequential argument to xtriggers.
+
+    The xtriggers are chosen to never be satisfied, so when we evolve the
+    workflow out to the runahead limit, sequential parentless tasks will
+    not spawn ahead to the limit, but non-sequential ones will.
+
+    """
     wid = flow({
         'scheduler': {
             'cycle point format': 'CCYY',
@@ -180,6 +190,7 @@ async def test_sequential_arg_ok(
     schd: Scheduler = scheduler(wid)
     expected_num_cycles = 1 if is_sequential else 3
     async with start(schd):
+        spawn_ahead(schd.pool)
         itask = schd.pool.get_task(ISO8601Point('2000'), 'foo')
         assert itask.is_xtrigger_sequential is is_sequential
         assert len(list_cycles(schd)) == expected_num_cycles
diff --git a/tests/integration/test_task_events_mgr.py b/tests/integration/test_task_events_mgr.py
index 9814c58c2d..6fff087a5c 100644
--- a/tests/integration/test_task_events_mgr.py
+++ b/tests/integration/test_task_events_mgr.py
@@ -236,7 +236,7 @@ async def get_ds_job_id(schd: Scheduler):
 
 
 async def test__process_message_failed_with_retry(
-    one: Scheduler, start, log_filter
+    one: Scheduler, start, log_filter, spawn_ahead
 ):
     """Log job failure, even if a retry is scheduled.
 
@@ -253,6 +253,10 @@ async def test__process_message_failed_with_retry(
                 'submission retry delays': [1]
             })
 
+        spawn_ahead(one.pool)
+        # Note this won't release queued tasks as schd is paused:
+        one.release_tasks_to_run()
+
         # Process submit failed message with and without retries:
         one.task_events_mgr._process_message_submit_failed(fail_once, None)
         record = log_filter(contains='1/one:waiting(queued)] retrying in')
diff --git a/tests/integration/test_task_pool.py b/tests/integration/test_task_pool.py
index cca42f0dbe..79b913d81f 100644
--- a/tests/integration/test_task_pool.py
+++ b/tests/integration/test_task_pool.py
@@ -329,7 +329,9 @@ async def test_release_held_tasks(
 async def test_hold_point(
     hold_after_point: str,
     expected_held_task_ids: List[str],
-    example_flow: 'Scheduler', db_select: Callable
+    example_flow: 'Scheduler',
+    db_select: Callable,
+    spawn_ahead
 ) -> None:
     """Test TaskPool.set_hold_point() and .release_hold_point()"""
     expected_held_task_ids = sorted(expected_held_task_ids)
@@ -341,6 +343,7 @@ async def test_hold_point(
     assert ('holdcp', str(hold_after_point)) in db_select(
         example_flow, True, 'workflow_params')
 
+    spawn_ahead(example_flow.pool)
     for itask in task_pool.get_tasks():
         hold_expected = itask.identity in expected_held_task_ids
         assert itask.state.is_held is hold_expected
@@ -458,6 +461,7 @@ async def test_runahead_after_remove(
 
     """
     task_pool = example_flow.pool
+    task_pool.compute_runahead()
     assert int(task_pool.runahead_limit_point) == 4
 
     # No change after removing an intermediate cycle.
@@ -560,7 +564,8 @@ def list_tasks(schd):
 async def test_restart_prereqs(
     flow, scheduler, start,
     graph_1, graph_2,
-    expected_1, expected_2, expected_3, expected_4
+    expected_1, expected_2, expected_3, expected_4,
+    spawn_ahead
 ):
     """It should handle graph prerequisites change on restart.
 
@@ -580,13 +585,13 @@ async def test_restart_prereqs(
     schd: Scheduler = scheduler(id_, paused_start=False)
 
     async with start(schd):
-        # Release tasks 1/a and 1/b
-        schd.pool.release_runahead_tasks()
+        # Release tasks 1/a and 1/b to run
+        spawn_ahead(schd.pool)
         schd.release_tasks_to_run()
         assert list_tasks(schd) == expected_1
 
-        # Mark 1/a as succeeded and spawn 1/z
         task_a = schd.pool.get_tasks()[0]
+        # Mark 1/a as succeeded and spawn 1/z
         schd.pool.task_events_mgr.process_message(task_a, 1, 'succeeded')
         assert list_tasks(schd) == expected_2
 
@@ -683,7 +688,8 @@ async def test_restart_prereqs(
 async def test_reload_prereqs(
     flow, scheduler, start,
     graph_1, graph_2,
-    expected_1, expected_2, expected_3, expected_4
+    expected_1, expected_2, expected_3, expected_4,
+    spawn_ahead
 ):
     """It should handle graph prerequisites change on reload.
 
@@ -703,8 +709,8 @@ async def test_reload_prereqs(
     schd: Scheduler = scheduler(id_, paused_start=False)
 
     async with start(schd):
-        # Release tasks 1/a and 1/b
-        schd.pool.release_runahead_tasks()
+        # Release tasks 1/a and 1/b to run
+        spawn_ahead(schd.pool)
         schd.release_tasks_to_run()
         assert list_tasks(schd) == expected_1
 
@@ -734,14 +740,14 @@ async def test_reload_prereqs(
         ) == expected_4
 
 
-async def _test_restart_prereqs_sat():
+async def _test_restart_prereqs_sat(spawn_ahead):
     schd: Scheduler
     # YIELD: the workflow has now started...
     schd = yield
     await schd.update_data_structure()
 
-    # Release tasks 1/a and 1/b
-    schd.pool.release_runahead_tasks()
+    # Release tasks 1/a and 1/b to run
+    spawn_ahead(schd.pool)
     schd.release_tasks_to_run()
     assert list_tasks(schd) == [
         ('1', 'a', 'running'),
@@ -799,7 +805,7 @@ async def _test_restart_prereqs_sat():
 
 @pytest.mark.parametrize('do_restart', [True, False])
 async def test_graph_change_prereq_satisfaction(
-    flow, scheduler, start, do_restart
+    flow, scheduler, start, do_restart, spawn_ahead
 ):
     """It should handle graph prerequisites change on reload/restart.
 
@@ -824,7 +830,7 @@ async def test_graph_change_prereq_satisfaction(
     id_ = flow(conf)
     schd = scheduler(id_, run_mode='simulation', paused_start=False)
 
-    test = _test_restart_prereqs_sat()
+    test = _test_restart_prereqs_sat(spawn_ahead)
     await test.asend(None)
 
     if do_restart:
@@ -843,8 +849,8 @@ async def test_graph_change_prereq_satisfaction(
 
     else:
         async with start(schd):
-            await test.asend(schd)
 
+            await test.asend(schd)
             # Modify flow.cylc to add a new dependency on "b"
             conf['scheduling']['graph']['R1'] += '\nb => c'
             flow(conf, workflow_id=id_)
@@ -880,6 +886,7 @@ async def test_runahead_limit_for_sequence_before_start_cycle(
     })
     schd = scheduler(id_, startcp='2005')
     async with start(schd):
+        schd.pool.compute_runahead()
         assert str(schd.pool.runahead_limit_point) == '20070101T0000Z'
 
 
@@ -1002,7 +1009,7 @@ async def test_no_flow_tasks_dont_spawn(
 
 
 async def test_task_proxy_remove_from_queues(
-    flow, one_conf, scheduler, start,
+    flow, one_conf, scheduler, start, spawn_ahead
 ):
     """TaskPool.remove should delete task proxies from queues.
 
@@ -1015,6 +1022,7 @@ async def test_task_proxy_remove_from_queues(
     }
     schd = scheduler(flow(one_conf))
     async with start(schd):
+        spawn_ahead(schd.pool)
         # Get a list of itasks:
         itasks = schd.pool.get_tasks()
 
@@ -1093,6 +1101,7 @@ async def test_trigger_icp_fcp_syntax(
     scheduler,
     start,
     log_filter,
+    spawn_ahead
 ):
     """It should support the ^/$ syntax for referencing the initial/final cp.
 
@@ -1114,9 +1123,11 @@ async def test_trigger_icp_fcp_syntax(
     id_ = flow(cfg)
     schd = scheduler(id_)
     async with start(schd) as log:
+        spawn_ahead(schd.pool)
         await commands.run_cmd(
             commands.force_trigger_tasks(schd, ['^/start', '$/end'], ['1'])
         )
+
         assert log_filter(contains='[1/start:waiting(queued)] => waiting')
         assert log_filter(contains='[2/end:waiting(queued)] => waiting')
         log.clear()
@@ -1129,6 +1140,7 @@ async def test_trigger_icp_fcp_syntax(
     id_ = flow(cfg)
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         await commands.run_cmd(
             commands.force_trigger_tasks(schd, ['^/start', '$/end'], ['1'])
         )
@@ -1181,13 +1193,16 @@ async def test_set_failed_complete(
     start,
     one_conf,
     log_filter,
-    db_select: Callable
+    db_select: Callable,
+    spawn_ahead
 ):
     """Test manual completion of an incomplete failed task."""
     id_ = flow(one_conf)
     schd: Scheduler = scheduler(id_)
     async with start(schd, level=logging.DEBUG):
         one = schd.pool.get_tasks()[0]
+
+        spawn_ahead(schd.pool)
         one.state_reset(is_queued=False)
 
         schd.pool.task_events_mgr.process_message(one, 1, "failed")
@@ -1197,8 +1212,8 @@ async def test_set_failed_complete(
             regex="1/one.* setting implied output: started")
         assert log_filter(
             regex="failed.* did not complete the required outputs")
-
         # Set failed task complete via default "set" args.
+
         schd.pool.set_prereqs_and_outputs({one.tokens}, [], [], [])
 
         assert log_filter(
@@ -1573,8 +1588,8 @@ async def test_set_outputs_from_skip_settings(
     schd: Scheduler = scheduler(id_)
 
     async with start(schd):
-        # it should start up with just tasks a:
-        assert schd.pool.get_task_ids() == {'1/a', '2/a'}
+        # it should start up with just 1/a:
+        assert schd.pool.get_task_ids() == {'1/a'}
 
         # setting 1/a output to skip should set output x, but not
         # y (because y is optional).
@@ -1657,17 +1672,15 @@ async def test_prereq_satisfaction(
         assert b.prereqs_are_satisfied()
 
 
-@pytest.mark.parametrize('compat_mode', ['compat-mode', 'normal-mode'])
 @pytest.mark.parametrize('cycling_mode', ['integer', 'datetime'])
 @pytest.mark.parametrize('runahead_format', ['P3Y', 'P3'])
 async def test_compute_runahead(
     cycling_mode,
-    compat_mode,
     runahead_format,
     flow,
     scheduler,
     start,
-    monkeypatch,
+    spawn_ahead
 ):
     """Test the calculation of the runahead limit.
 
@@ -1675,7 +1688,6 @@ async def test_compute_runahead(
     * Runahead tasks are excluded from computations
       see https://github.com/cylc/cylc-flow/issues/5825
     * Tasks are initiated with the correct is_runahead status on startup.
-    * Behaviour in compat/regular modes is same unless failed tasks are present
     * Behaviour is the same for integer/datetime cycling modes.
 
     """
@@ -1711,48 +1723,37 @@ async def test_compute_runahead(
         }
         point = ISO8601Point
 
-    monkeypatch.setattr(
-        'cylc.flow.flags.cylc7_back_compat',
-        compat_mode == 'compat-mode',
-    )
-
     id_ = flow(config)
     schd = scheduler(id_)
     async with start(schd):
         schd.pool.compute_runahead(force=True)
         assert int(str(schd.pool.runahead_limit_point)) == 4
 
-        # ensure task states are initiated with is_runahead status
-        assert schd.pool.get_task(point('0001'), 'a').state(is_runahead=False)
-        assert schd.pool.get_task(point('0005'), 'a').state(is_runahead=True)
+        # It should start with 1/a runahead
+        tasks = schd.pool.get_tasks()
+        assert len(tasks) == 1
+        assert schd.pool.get_task(point('0001'), 'a').state(is_runahead=True)
 
-        # mark the first three cycles as running
-        for cycle in range(1, 4):
-            schd.pool.get_task(point(f'{cycle:04}'), 'a').state.reset(
-                TASK_STATUS_RUNNING
-            )
+        spawn_ahead(schd.pool)
+
+        assert schd.pool.get_task(point('0005'), 'a').state(is_runahead=True)
 
         schd.pool.compute_runahead(force=True)
         assert int(str(schd.pool.runahead_limit_point)) == 4  # no change
 
-        # In Cylc 8 all incomplete tasks hold back runahead.
+        # All incomplete tasks hold back runahead.
 
-        # In Cylc 7, submit-failed tasks hold back runahead..
         schd.pool.get_task(point('0001'), 'a').state.reset(
             TASK_STATUS_SUBMIT_FAILED
         )
         schd.pool.compute_runahead(force=True)
         assert int(str(schd.pool.runahead_limit_point)) == 4
 
-        # ... but failed ones don't. Go figure.
         schd.pool.get_task(point('0001'), 'a').state.reset(
             TASK_STATUS_FAILED
         )
         schd.pool.compute_runahead(force=True)
-        if compat_mode == 'compat-mode':
-            assert int(str(schd.pool.runahead_limit_point)) == 5
-        else:
-            assert int(str(schd.pool.runahead_limit_point)) == 4  # no change
+        assert int(str(schd.pool.runahead_limit_point)) == 4  # no change
 
         # mark cycle 1 as complete
         # (via task message so the task gets removed before runahead compute)
@@ -1817,14 +1818,12 @@ async def test_compute_runahead_with_no_sequences(
 
 
 @pytest.mark.parametrize('rhlimit', ['P2D', 'P2'])
-@pytest.mark.parametrize('compat_mode', ['compat-mode', 'normal-mode'])
 async def test_runahead_future_trigger(
     flow,
     scheduler,
     start,
     monkeypatch,
     rhlimit,
-    compat_mode,
 ):
     """Equivalent time interval and cycle count runahead limits should yield
     the same limit point, even if there is a future trigger.
@@ -1848,12 +1847,9 @@ async def test_runahead_future_trigger(
         }
     })
 
-    monkeypatch.setattr(
-        'cylc.flow.flags.cylc7_back_compat',
-        compat_mode == 'compat-mode',
-    )
     schd = scheduler(id_,)
     async with start(schd, level=logging.DEBUG):
+        schd.pool.compute_runahead()
         assert str(schd.pool.runahead_limit_point) == '20010103'
         schd.pool.release_runahead_tasks()
         for itask in schd.pool.get_tasks():
@@ -1891,52 +1887,6 @@ async def mod_blah(
         yield schd
 
 
-@pytest.mark.parametrize(
-    'status, expected',
-    [
-        # (Status, Are we expecting an update?)
-        (TASK_STATUS_WAITING, False),
-        (TASK_STATUS_EXPIRED, False),
-        (TASK_STATUS_PREPARING, False),
-        (TASK_STATUS_SUBMIT_FAILED, False),
-        (TASK_STATUS_SUBMITTED, False),
-        (TASK_STATUS_RUNNING, False),
-        (TASK_STATUS_FAILED, True),
-        (TASK_STATUS_SUCCEEDED, True)
-    ]
-)
-async def test_runahead_c7_compat_task_state(
-    status,
-    expected,
-    mod_blah,
-    monkeypatch,
-):
-    """For each task status check whether changing the oldest task
-    to that status will cause compute_runahead to make a change.
-
-    Compat mode: Cylc 7 ignored failed tasks but not submit-failed!
-
-    """
-
-    def max_cycle(tasks):
-        return max([int(t.tokens.get("cycle")) for t in tasks])
-
-    monkeypatch.setattr(
-        'cylc.flow.flags.cylc7_back_compat', True)
-    monkeypatch.setattr(
-        'cylc.flow.task_events_mgr.TaskEventsManager._insert_task_job',
-        lambda *_: True)
-
-    mod_blah.pool.compute_runahead()
-    before_pt = max_cycle(mod_blah.pool.get_tasks())
-    before = mod_blah.pool.runahead_limit_point
-    itask = mod_blah.pool.get_task(ISO8601Point(f'{before_pt - 2:04}'), 'a')
-    itask.state_reset(status, is_queued=False)
-    mod_blah.pool.compute_runahead()
-    after = mod_blah.pool.runahead_limit_point
-    assert bool(before != after) == expected
-
-
 async def test_fast_respawn(
     example_flow: 'Scheduler',
     caplog: pytest.LogCaptureFixture,
@@ -2197,7 +2147,7 @@ async def test_trigger_queue(one, run, db_select, complete):
     async with run(one):
         # the workflow should start up with one task in the original flow
         task = one.pool.get_tasks()[0]
-        assert task.state(TASK_STATUS_WAITING, is_queued=True)
+        assert task.state(TASK_STATUS_WAITING, is_runahead=True)
         assert task.flow_nums == {1}
 
         # trigger this task even though is already queued in flow 1
@@ -2376,8 +2326,8 @@ def method(schd):
         schd.task_events_mgr.process_message(foo, 0, 'failed')
         schd.task_events_mgr._retry_task(foo, 0)
 
-        # the task should start as "waiting(queued)"
-        assert foo.state(TASK_STATUS_WAITING, is_queued=True)
+        # the task should start as "waiting(runahead)"
+        assert foo.state(TASK_STATUS_WAITING, is_runahead=True)
 
         # expire the task via whichever method we are testing
         method(schd)
@@ -2527,7 +2477,7 @@ async def test_start_tasks(
     flow,
     scheduler,
     start,
-    capture_submission,
+    capture_submission
 ):
     """Check starting from "start-tasks" with and without clock-triggers.
 
@@ -2565,13 +2515,8 @@ async def test_start_tasks(
         submitted_tasks = capture_submission(schd)
         assert submitted_tasks == set()
 
-        # This test expects tasks to be spawned to the runahead limit for
-        # historical reasons - pre GitHub #7237. We can remove this call
-        # if we change test expectations below.
-        schd.pool.spawn_to_runahead_limit()
-
         # It should start up with:
-        # - 2050/foo and 2051/foo (spawned to runahead limit)
+        # - 2050/foo
         # - 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)
@@ -2579,23 +2524,25 @@ async def test_start_tasks(
         assert (
             set(itask.identity for itask in itasks) == {
                 "2050/foo",
-                "2051/foo",
                 "2050/bar",
                 "2050/baz",
             }
         )
 
+        # do all the stuff, including checking xtriggers
         await schd._main_loop()
 
         # It should submit 2050/foo, 2050/baz
         # It should not submit 2050/bar (waiting on clock trigger)
-        # It should not submit 2051/foo (runahead limited)
         assert (
             set(itask.identity for itask in submitted_tasks) == {
                 "2050/foo",
                 "2050/baz",
             }
         )
+        # It should spawn 2051/foo as runahead limited
+        foo = schd.pool._get_task_by_id("2051/foo")
+        assert foo.state(is_runahead=True)
 
 
 async def test_add_new_flow_rows_on_spawn(
diff --git a/tests/integration/test_taskdef.py b/tests/integration/test_taskdef.py
index 5a10990425..ce50e9af95 100644
--- a/tests/integration/test_taskdef.py
+++ b/tests/integration/test_taskdef.py
@@ -22,7 +22,9 @@
 from cylc.flow.workflow_files import WorkflowFiles
 
 
-async def test_almost_self_suicide(flow, scheduler, start):
+async def test_almost_self_suicide(
+    flow, scheduler, start, spawn_ahead
+):
     """Suicide triggers should not count as upstream tasks when looking
     to spawn parentless tasks.
 
@@ -46,6 +48,7 @@ async def test_almost_self_suicide(flow, scheduler, start):
     })
     schd = scheduler(wid)
     async with start(schd):
+        spawn_ahead(schd.pool)
         tasks = [str(t) for t in schd.pool.get_tasks()]
         for task in ['1990/a:waiting', '1991/a:waiting', '1992/a:waiting']:
             assert task in tasks
diff --git a/tests/integration/tui/test_app.py b/tests/integration/tui/test_app.py
index 8871e81b92..772a24e881 100644
--- a/tests/integration/tui/test_app.py
+++ b/tests/integration/tui/test_app.py
@@ -121,12 +121,13 @@ async def test_tui_basics(rakiura):
 
 
 async def test_subscribe_unsubscribe(
-    one_conf, flow, scheduler, start, rakiura
+    one_conf, flow, scheduler, start, rakiura, spawn_ahead
 ):
     """Test a simple workflow with one task."""
     id_ = flow(one_conf, name='one')
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         await schd.update_data_structure()
         with rakiura(size='80,15') as rk:
             rk.compare_screenshot(
@@ -223,7 +224,9 @@ async def test_workflow_states(one_conf, flow, scheduler, start, rakiura):
                 )
 
 
-async def test_task_states(flow, scheduler, start, rakiura):
+async def test_task_states(
+    flow, scheduler, start, rakiura, spawn_ahead
+):
     id_ = flow({
         'scheduler': {
             'allow implicit tasks': 'true',
@@ -249,6 +252,7 @@ async def test_task_states(flow, scheduler, start, rakiura):
     }, name='test_task_states')
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         set_task_state(
             schd,
             [
@@ -332,7 +336,9 @@ async def test_task_states(flow, scheduler, start, rakiura):
             )
 
 
-async def test_task_modifiers(flow, scheduler, start, rakiura):
+async def test_task_modifiers(
+    flow, scheduler, start, rakiura, spawn_ahead
+):
     """It should display task modifiers and text summaries of them."""
     id_ = flow({
         'scheduling': {
@@ -346,6 +352,7 @@ async def test_task_modifiers(flow, scheduler, start, rakiura):
     }, name='test_task_modifiers')
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         set_task_state(
             schd,
             [
@@ -390,7 +397,9 @@ async def test_task_modifiers(flow, scheduler, start, rakiura):
             )
 
 
-async def test_navigation(flow, scheduler, start, rakiura):
+async def test_navigation(
+    flow, scheduler, start, rakiura, spawn_ahead
+):
     """Test navigating with the arrow keys."""
     id_ = flow({
         'scheduling': {
@@ -413,6 +422,7 @@ async def test_navigation(flow, scheduler, start, rakiura):
     }, name='one')
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         await schd.update_data_structure()
 
         with rakiura(size='80,30') as rk:
@@ -449,7 +459,9 @@ async def test_navigation(flow, scheduler, start, rakiura):
             )
 
 
-async def test_auto_expansion(flow, scheduler, start, rakiura):
+async def test_auto_expansion(
+    flow, scheduler, start, rakiura, spawn_ahead
+):
     """It should automatically expand cycles and top-level families.
 
     When a workflow is expanded, Tui should auto expand cycles and top-level
@@ -474,6 +486,7 @@ async def test_auto_expansion(flow, scheduler, start, rakiura):
     schd = scheduler(id_)
     with rakiura(size='80,20') as rk:
         async with start(schd):
+            spawn_ahead(schd.pool)
             await schd.update_data_structure()
             # wait for the workflow to appear (collapsed)
             rk.wait_until_loaded('#spring')
@@ -505,7 +518,9 @@ async def test_auto_expansion(flow, scheduler, start, rakiura):
             )
 
 
-async def test_restart_reconnect(one_conf, flow, scheduler, start, rakiura):
+async def test_restart_reconnect(
+    one_conf, flow, scheduler, start, rakiura, spawn_ahead
+):
     """It should handle workflow shutdown and restart.
 
     The Cylc client can raise exceptions e.g. WorkflowStopped. Any text written
@@ -518,6 +533,7 @@ async def test_restart_reconnect(one_conf, flow, scheduler, start, rakiura):
 
         # 1- start the workflow
         async with start(schd):
+            spawn_ahead(schd.pool)
             await schd.update_data_structure()
             # wait for the workflow to appear (collapsed)
             rk.wait_until_loaded('#spring')
@@ -543,7 +559,7 @@ async def test_restart_reconnect(one_conf, flow, scheduler, start, rakiura):
         # 3- restart the workflow
         schd = scheduler(flow(one_conf, name='one'))
         async with start(schd):
-            await schd._main_loop()
+            spawn_ahead(schd.pool)
             await schd.update_data_structure()
             rk.wait_until_loaded(schd.tokens.id)
             rk.compare_screenshot(
@@ -552,7 +568,9 @@ async def test_restart_reconnect(one_conf, flow, scheduler, start, rakiura):
             )
 
 
-async def test_states(flow, scheduler, start, rakiura):
+async def test_states(
+    flow, scheduler, start, rakiura, spawn_ahead
+):
     """It should dim no-flow tasks and display state summary in context menus.
     """
     id_ = flow(
@@ -569,6 +587,7 @@ async def test_states(flow, scheduler, start, rakiura):
     schd: Scheduler = scheduler(id_)
 
     async with start(schd):
+        spawn_ahead(schd.pool)
         a = schd.pool.get_task(IntegerPoint('1'), 'a')
         b = schd.pool.get_task(IntegerPoint('1'), 'b')
         c = schd.pool.get_task(IntegerPoint('1'), 'c')
diff --git a/tests/integration/tui/test_mutations.py b/tests/integration/tui/test_mutations.py
index 200b6a361c..e526bdeae6 100644
--- a/tests/integration/tui/test_mutations.py
+++ b/tests/integration/tui/test_mutations.py
@@ -54,12 +54,14 @@ async def test_online_mutation(
     rakiura,
     monkeypatch,
     log_filter,
+    spawn_ahead
 ):
     """Test a simple workflow with one task."""
     id_ = flow(one_conf, name='one')
     schd = scheduler(id_)
     with rakiura(size='80,15') as rk:
         async with start(schd):
+            spawn_ahead(schd.pool)
             await schd.update_data_structure()
             assert schd.command_queue.empty()
 
@@ -236,6 +238,7 @@ async def test_set_mutation(
     scheduler,
     start,
     rakiura,
+    spawn_ahead
 ):
     id_ = flow({
         'scheduling': {
@@ -246,6 +249,7 @@ async def test_set_mutation(
     }, name='one')
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         await schd.update_data_structure()
         with rakiura(schd.tokens.id, size='80,15') as rk:
             # open the context menu on 1/a
diff --git a/tests/integration/tui/test_show.py b/tests/integration/tui/test_show.py
index 5327e1b887..86ed3953b7 100644
--- a/tests/integration/tui/test_show.py
+++ b/tests/integration/tui/test_show.py
@@ -18,7 +18,9 @@
 from cylc.flow.exceptions import ClientError
 
 
-async def test_show(flow, scheduler, start, rakiura, monkeypatch):
+async def test_show(
+    flow, scheduler, start, rakiura, monkeypatch, spawn_ahead
+):
     """Test "cylc show" support in Tui."""
     id_ = flow({
         'scheduling': {
@@ -37,6 +39,7 @@ async def test_show(flow, scheduler, start, rakiura, monkeypatch):
     }, name='one')
     schd = scheduler(id_)
     async with start(schd):
+        spawn_ahead(schd.pool)
         await schd.update_data_structure()
 
         with rakiura(size='80,40') as rk: