From ddcdc964ab90e4ffdb346f40f4d34a8d737e98a6 Mon Sep 17 00:00:00 2001
From: Hilary James Oliver <hilary.j.oliver@gmail.com>
Date: Mon, 22 Jan 2024 11:33:30 +1300
Subject: [PATCH] Clean up expire validation.

---
 cylc/flow/config.py              | 22 ++++++++++------------
 cylc/flow/graph_parser.py        |  5 +++++
 tests/integration/test_config.py |  7 +++++++
 3 files changed, 22 insertions(+), 12 deletions(-)

diff --git a/cylc/flow/config.py b/cylc/flow/config.py
index b536f52c702..93a49a435c3 100644
--- a/cylc/flow/config.py
+++ b/cylc/flow/config.py
@@ -2142,28 +2142,26 @@ def set_required_outputs(
                     continue
                 taskdef.set_required_output(output, not optional)
 
-        # Expire is dangerous, it must be visible and optional in the graph
-        bad_exp = set()
-        good_exp = set()
-        for (task, output), (opt, _, _) in task_output_opt.items():
+        # Add expired outputs to taskdefs if flagged in the graph.
+        graph_exp = set()
+        for (task, output) in task_output_opt.keys():
             if output == TASK_OUTPUT_EXPIRED:
-                if not opt:
-                    bad_exp.add(task)
-                    continue
-                good_exp.add(task)
+                graph_exp.add(task)
                 self.taskdefs[task].add_output(
                     TASK_OUTPUT_EXPIRED, TASK_OUTPUT_EXPIRED
                 )
 
-        # likewise clock-expiry is only legal if flagged in the graph
+        # clock-expire must be flagged in the graph for visibility
+        bad_exp = set()
         for task in self.expiration_offsets:
-            if task not in good_exp:
+            if task not in graph_exp:
                 bad_exp.add(task)
 
         if bad_exp:
-            msg = '\n   '.join([t + ":expired?" for t in bad_exp])
+            msg = '\n   '.join(
+                [t + f":{TASK_OUTPUT_EXPIRED}?" for t in bad_exp])
             raise WorkflowConfigError(
-                f"Expiring tasks must be optional in the graph as:\n   {msg}"
+                f"Clock-expire must be visible in the graph:\n  {msg}"
             )
 
     def find_taskdefs(self, name: str) -> Set[TaskDef]:
diff --git a/cylc/flow/graph_parser.py b/cylc/flow/graph_parser.py
index ab3890961f9..090ebae1c5a 100644
--- a/cylc/flow/graph_parser.py
+++ b/cylc/flow/graph_parser.py
@@ -61,6 +61,7 @@
 
 class Replacement:
     """A class to remember match group information in re.sub() calls"""
+
     def __init__(self, replacement):
         self.replacement = replacement
         self.substitutions = []
@@ -745,6 +746,10 @@ def _set_output_opt(
         if suicide:
             return
 
+        if output == TASK_OUTPUT_EXPIRED and not optional:
+            raise GraphParseError(
+                f"Output {name}:{output} must be optional (append '?')")
+
         if output == TASK_OUTPUT_FINISHED:
             # Interpret :finish pseudo-output
             if optional:
diff --git a/tests/integration/test_config.py b/tests/integration/test_config.py
index acf24d17eaf..dd8da5edb24 100644
--- a/tests/integration/test_config.py
+++ b/tests/integration/test_config.py
@@ -263,6 +263,13 @@ def test_parse_special_tasks_families(flow, scheduler, validate, section):
         with pytest.raises(WorkflowConfigError) as exc_ctx:
             config = validate(id_)
         assert 'external triggers must be used only once' in str(exc_ctx.value)
+
+    elif section == 'clock-expire':
+        with pytest.raises(WorkflowConfigError) as exc_ctx:
+            config = validate(id_)
+        assert (
+            'Clock-expire must be visible in the graph' in str(exc_ctx.value)
+        )
     else:
         config = validate(id_)
         assert set(config.cfg['scheduling']['special tasks'][section]) == {