Add future final-incomplete tasks to n=0 for visibility.

changes.d/7248.feat.md

@@ -0,0 +1 @@
+Spawn future final-incomplete tasks into n=0 for visibility.

cylc/flow/task_pool.py

@@ -2254,6 +2254,21 @@ def _set_outputs_itask(
             # Can't be runahead limited or queued.
             itask.state_reset(is_runahead=False, is_queued=False)
             self.task_queue_mgr.remove_task(itask)
+            # If we skipped over runahead release and this tasks parentless
+            # it's next parentless instance might need to be spawned.
+            if itask.flow_nums and not itask.is_xtrigger_sequential:
+                self.spawn_next_parentless(itask)
+
+        if (
+            itask.state(*TASK_STATUSES_FINAL)
+            and not itask.state.outputs.is_complete()
+        ):
+            # Add future final-incomplete tasks to the pool for visibility.
+            # See https://github.com/cylc/cylc-flow/issues/6383.
+            LOG.debug(f"[{itask}] adding future incomplete task to n=0.")
+            self.add_to_pool(itask)
+            # (The transient used for spawning outputs is now not transient.)
+            itask.transient = False
 
         if no_op:
             return False

tests/integration/test_task_pool.py

@@ -1496,7 +1496,10 @@ async def test_set_outputs_future(
         {
             'scheduling': {
                 'graph': {
-                    'R1': "a:x & a:y => b => c"
+                    'R1': """
+                        a:x & a:y => b => c
+                        a:y => f
+                    """
                 }
             },
             'runtime': {
@@ -1521,6 +1524,12 @@ async def test_set_outputs_future(
             {TaskTokens('1', 'b')}, ["succeeded"], [], [])
         assert schd.pool.get_task_ids() == {"1/a", "1/c"}
 
+        # setting inactive task f failed should add it to n=0 as final
+        # incomplete - see https://github.com/cylc/cylc-flow/pull/7248
+        schd.pool.set_prereqs_and_outputs(
+            {TaskTokens('1', 'f')}, ["failed"], [], [])
+        assert schd.pool.get_task_ids() == {"1/a", "1/c", "1/f"}
+
         schd.pool.set_prereqs_and_outputs(
             items={TaskTokens('1', 'a')},
             outputs=["x", "y", "cheese"],