Merge pull request cylc#6247 from cylc/8.3.x-sync

🤖 Merge 8.3.x-sync into master

6103.fix.md

@@ -0,0 +1,2 @@
+Absolute dependencies (dependencies on tasks in a specified cycle rather than at a specified offset) are now visible in the GUI beyond the specified cycle.
+

changes.d/6242.fix.md

@@ -0,0 +1 @@
+Put `share/bin` in the `PATH` of scheduler environment, event handlers therein will now be found.

cylc/flow/config.py

@@ -1724,8 +1724,21 @@ def process_config_env(self):
         os.environ['CYLC_CYCLING_MODE'] = self.cfg['scheduling'][
             'cycling mode']
         # Add workflow bin directory to PATH for workflow and event handlers
-        os.environ['PATH'] = os.pathsep.join([
-            os.path.join(self.fdir, 'bin'), os.environ['PATH']])
+        if self.share_dir is not None:
+            os.environ['PATH'] = os.pathsep.join(
+                [
+                    os.path.join(self.share_dir, 'bin'),
+                    os.path.join(self.fdir, 'bin'),
+                    os.environ['PATH']
+                ]
+            )
+        else:
+            os.environ['PATH'] = os.pathsep.join(
+                [
+                    os.path.join(self.fdir, 'bin'),
+                    os.environ['PATH']
+                ]
+            )
 
     def run_mode(self) -> str:
         """Return the run mode."""
@@ -2439,9 +2452,18 @@ def get_taskdef(
                 raise WorkflowConfigError(str(exc)) from None
             else:
                 # Record custom message outputs from [runtime].
+                messages = set(self.cfg['runtime'][name]['outputs'].values())
                 for output, message in (
                     self.cfg['runtime'][name]['outputs'].items()
                 ):
+                    try:
+                        messages.remove(message)
+                    except KeyError:
+                        raise WorkflowConfigError(
+                            'Duplicate task message in'
+                            f' "[runtime][{name}][outputs]'
+                            f'{output} = {message}" - messages must be unique'
+                        )
                     valid, msg = TaskOutputValidator.validate(output)
                     if not valid:
                         raise WorkflowConfigError(

cylc/flow/taskdef.py

@@ -101,11 +101,7 @@ def generate_graph_parents(tdef, point, taskdefs):
                 # where (point -Px) does not land on a valid point for woo.
                 # TODO ideally validation would flag this as an error.
                 continue
-            is_abs = (trigger.offset_is_absolute or
-                      trigger.offset_is_from_icp)
-            if is_abs and parent_point != point:
-                # If 'foo[^] => bar' only spawn off of '^'.
-                continue
+            is_abs = trigger.offset_is_absolute or trigger.offset_is_from_icp
             graph_parents[seq].append((parent_name, parent_point, is_abs))
 
     if tdef.sequential:

tests/integration/test_data_store_mgr.py

@@ -18,6 +18,7 @@
 from typing import TYPE_CHECKING
 
 from cylc.flow.data_store_mgr import (
+    EDGES,
     FAMILY_PROXIES,
     JOBS,
     TASKS,
@@ -316,3 +317,36 @@ def test_delta_task_prerequisite(harness):
         p.satisfied
         for t in schd.data_store_mgr.updated[TASK_PROXIES].values()
         for p in t.prerequisites})
+
+
+async def test_absolute_graph_edges(flow, scheduler, start):
+    """It should add absolute graph edges to the store.
+
+    See: https://github.com/cylc/cylc-flow/issues/5845
+    """
+    runahead_cycles = 1
+    id_ = flow({
+        'scheduling': {
+            'initial cycle point': '1',
+            'cycling mode': 'integer',
+            'runahead limit': f'P{runahead_cycles}',
+            'graph': {
+                'R1': 'build',
+                'P1': 'build[^] => run',
+            },
+        },
+    })
+    schd = scheduler(id_)
+
+    async with start(schd):
+        await schd.update_data_structure()
+
+        assert {
+            (Tokens(edge.source).relative_id, Tokens(edge.target).relative_id)
+            for edge in schd.data_store_mgr.data[schd.id][EDGES].values()
+        } == {
+            ('1/build', f'{cycle}/run')
+            # +1 for Python's range()
+            # +2 for Cylc's runahead
+            for cycle in range(1, runahead_cycles + 3)
+        }

tests/integration/validate/test_outputs.py

@@ -339,3 +339,37 @@ def test_completion_expression_cylc7_compat(
         match="completion cannot be used in Cylc 7 compatibility mode."
     ):
         validate(id_)
+
+
+def test_unique_messages(
+    flow,
+    validate
+):
+    """Task messages must be unique in the [outputs] section.
+
+    See: https://github.com/cylc/cylc-flow/issues/6056
+    """
+    id_ = flow({
+        'scheduling': {
+            'graph': {'R1': 'foo'}
+        },
+        'runtime': {
+            'foo': {
+                'outputs': {
+                    'a': 'foo',
+                    'b': 'bar',
+                    'c': 'baz',
+                    'd': 'foo',
+                }
+            },
+        }
+    })
+
+    with pytest.raises(
+        WorkflowConfigError,
+        match=(
+            r'"\[runtime\]\[foo\]\[outputs\]d = foo"'
+            ' - messages must be unique'
+        ),
+    ):
+        validate(id_)