wip

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:])
@@ -1859,10 +1860,6 @@ def generate_triggers(self, lexpression, left_nodes, right, seq,
                 # Qualifier is a custom task message.
                 qualifier = outputs[output]
             elif output:
-                if not TaskOutputs.is_valid_std_name(output):
-                    raise WorkflowConfigError(
-                        f"Undefined custom output: {name}:{output}"
-                    )
                 qualifier = output
             else:
                 # No qualifier specified => use "succeeded".
@@ -2261,6 +2258,7 @@ def load_graph(self):
                 task_output_opt=task_output_opt
             )
             parser.parse_graph(graph)
+            self.check_outputs(parser.task_output_opt.keys())
             task_output_opt.update(parser.task_output_opt)
             self.workflow_polling_tasks.update(
                 parser.workflow_state_polling_tasks)
@@ -2278,6 +2276,29 @@ def load_graph(self):
         for tdef in self.taskdefs.values():
             tdef.tweak_outputs()
 
+
+    def check_outputs(self, tasks_and_ouputs):
+        """Check that task outputs have been registered with tasks.
+
+
+        TODO - have a bit more of a think about whether the try except is
+        safe
+        """
+        for task, output in tasks_and_ouputs:
+
+            try:
+                registered_outputs = self.cfg['runtime'][task]['outputs']
+            except KeyError:
+                pass
+            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