Ensure that undefined optional outputs raise an error everywhere in g…

…raph.

changes.d/6583.fix.md

@@ -0,0 +1 @@
+Fix bug where graph items with undefined outputs were missed at validation if the graph item was not an upstream dependency of another graph item.

cylc/flow/config.py

@@ -1844,6 +1844,7 @@ def generate_triggers(self, lexpression, left_nodes, right, seq,
 
         triggers = {}
         xtrig_labels = set()
+
         for left in left_nodes:
             if left.startswith('@'):
                 xtrig_labels.add(left[1:])
@@ -2265,6 +2266,17 @@ def load_graph(self):
             self.workflow_polling_tasks.update(
                 parser.workflow_state_polling_tasks)
             self._proc_triggers(parser, seq, task_triggers)
+            self.check_outputs(
+                [
+                    task_output
+                    for task_output in parser.task_output_opt
+                    if task_output[0]
+                    in [
+                        task_output.split(':')[0]
+                        for task_output in parser.terminals
+                    ]
+                ]
+            )
 
         self.set_required_outputs(task_output_opt)
 
@@ -2278,6 +2290,26 @@ def load_graph(self):
         for tdef in self.taskdefs.values():
             tdef.tweak_outputs()
 
+    def check_outputs(
+        self, tasks_and_outputs: Iterable[Tuple[str, str]]
+    ) -> None:
+        """Check that task outputs have been registered with tasks.
+
+        Args: tasks_and_outputs: ((task, output), ...)
+
+        Raises: WorkflowConfigError is a user has defined a task with a
+        custom output, but has not registered a custom output.
+        """
+        for task, output in tasks_and_outputs:
+            registered_outputs = self.cfg['runtime'][task]['outputs']
+            if (
+                not TaskOutputs.is_valid_std_name(output)
+                and output not in registered_outputs
+            ):
+                raise WorkflowConfigError(
+                    f"Undefined custom output: {task}:{output}"
+                )
+
     def _proc_triggers(self, parser, seq, task_triggers):
         """Define graph edges, taskdefs, and triggers, from graph sections."""
         suicides = 0

cylc/flow/graph_parser.py

@@ -470,10 +470,12 @@ def parse_graph(self, graph_string: str) -> None:
                 pairs.add((chain[i], chain[i + 1]))
 
         # Get a set of RH nodes which are not at the LH of another pair:
-        terminals = {p[1] for p in pairs}.difference({p[0] for p in pairs})
+        self.terminals = {p[1] for p in pairs}.difference(
+            {p[0] for p in pairs}
+        )
 
         for pair in sorted(pairs, key=lambda p: str(p[0])):
-            self._proc_dep_pair(pair, terminals)
+            self._proc_dep_pair(pair, self.terminals)
 
     @classmethod
     def _report_invalid_lines(cls, lines: List[str]) -> None:

tests/integration/test_dbstatecheck.py

@@ -45,7 +45,7 @@ async def checker(
         },
         'runtime': {
             'bad': {'simulation': {'fail cycle points': '1000'}},
-            'output': {'outputs': {'trigger': 'message'}}
+            'output': {'outputs': {'trigger': 'message', 'custom_output': 'foo'}}
         }
     })
     schd: Scheduler = mod_scheduler(wid, paused_start=False)
@@ -119,13 +119,13 @@ def test_output(checker):
             'output',
             '10000101T0000Z',
             "{'submitted': 'submitted', 'started': 'started', 'succeeded': "
-            "'succeeded', 'trigger': 'message'}",
+            "'succeeded', 'trigger': 'message', 'custom_output': 'foo'}",
         ],
         [
             'output',
             '10010101T0000Z',
             "{'submitted': 'submitted', 'started': 'started', 'succeeded': "
-            "'succeeded', 'trigger': 'message'}",
+            "'succeeded', 'trigger': 'message', 'custom_output': 'foo'}",
         ],
     ]
     assert result == expect

tests/integration/test_optional_outputs.py

@@ -288,11 +288,7 @@ def implicit_completion_config(mod_flow, mod_validate):
         },
         'runtime': {
             'root': {
-                'outputs': {
-                    'x': 'xxx',
-                    'y': 'yyy',
-                    'z': 'zzz',
-                }
+                'outputs': {x: f'{x * 3}' for x in 'abcdefghijklxyz'}
             }
         }
     })

tests/integration/validate/test_outputs.py

@@ -276,6 +276,7 @@ def test_completion_expression_invalid(
                 'outputs': {
                     'x': 'xxx',
                     'y': 'yyy',
+                    'file-1': 'asdf'
                 },
             },
         },