Use ID tokens for prereq validation.

cylc/flow/id.py

@@ -699,6 +699,16 @@ def detokenise(
     selectors: bool = False,
     relative: bool = False,
 ) -> str:
+    return '/'.join(
+        detokenise_to_list(tokens, selectors, relative)
+    )
+
+
+def detokenise_to_list(
+    tokens: Tokens,
+    selectors: bool = False,
+    relative: bool = False,
+) -> list:
     """Convert Cylc tokens into a string identifier.
 
     Args:
@@ -807,7 +817,7 @@ def detokenise(
         if token == lowest_token:
             break
 
-    return '/'.join(identifier)
+    return identifier
 
 
 def upgrade_legacy_ids(*ids: str, relative=False) -> List[str]:

cylc/flow/task_pool.py

@@ -16,7 +16,6 @@
 
 """Wrangle task proxies to manage the workflow."""
 
-import re
 from contextlib import suppress
 from collections import Counter
 import json
@@ -38,7 +37,7 @@
 from cylc.flow.cycling.loader import get_point, standardise_point_string
 from cylc.flow.exceptions import (
     WorkflowConfigError, PointParsingError, PlatformLookupError)
-from cylc.flow.id import Tokens, detokenise
+from cylc.flow.id import Tokens, detokenise, detokenise_to_list
 from cylc.flow.id_cli import contains_fnmatch
 from cylc.flow.id_match import filter_ids
 from cylc.flow.network.resolvers import TaskMsg
@@ -73,6 +72,7 @@
 
 from cylc.flow.flow_mgr import FLOW_ALL, FLOW_NONE, FLOW_NEW
 
+
 if TYPE_CHECKING:
     from queue import Queue
     from cylc.flow.config import WorkflowConfig
@@ -83,16 +83,8 @@
     from cylc.flow.workflow_db_mgr import WorkflowDatabaseManager
     from cylc.flow.flow_mgr import FlowMgr, FlowNums
 
-Pool = Dict['PointBase', Dict[str, TaskProxy]]
-
 
-# CLI prerequisite pattern: point/name:label
-REC_CLI_PREREQ = re.compile(
-    rf"({TaskID.POINT_RE})" +
-    rf"{TaskID.DELIM2}" +
-    rf"({TaskID.NAME_RE})" +
-    r':' + r'(\w+)'  # TODO: formally define qualifier RE?
-)
+Pool = Dict['PointBase', Dict[str, TaskProxy]]
 
 
 class TaskPool:
@@ -1676,11 +1668,20 @@ def set_prereqs(self, point, taskdef, prereqs, flow_nums):
         if prereqs == ["all"]:
             itask.state.set_all_satisfied()
         else:
-            for pre in prereqs:
-                m = REC_CLI_PREREQ.match(pre)
-                if m is not None:
-                    itask.satisfy_me({m.groups()})
-                # (regex already checked in the CLI)
+            # Generate a set of individual prerequisites in the right format.
+            itask.satisfy_me(
+                {
+                    (i, *j.split(':'))
+                    for i, j in [
+                        detokenise_to_list(
+                            Tokens(p, relative=True),
+                            selectors=True,
+                            relative=True
+                        )
+                        for p in prereqs
+                    ]
+                }
+            )
 
         self.data_store_mgr.delta_task_prerequisite(itask)
         self.add_to_pool(itask)