Fix wfs polling compat.

cylc/flow/config.py

@@ -1549,22 +1549,29 @@ def configure_workflow_state_polling_tasks(self):
                 f"{tdef.workflow_polling_cfg['task']}"
             )
             graph_trigger = tdef.workflow_polling_cfg['status']
-            config_trigger = rtc['workflow state polling']['output']
+            config_output = rtc['workflow state polling']['output']
+            config_message = rtc['workflow state polling']['message']
+            config_trigger = config_message or config_output
             if (
                 graph_trigger is not None and
-                config_trigger is not None and
-                graph_trigger != config_trigger
+                (
+                    config_trigger is not None
+                ) and (
+                    graph_trigger != config_trigger
+                )
             ):
                 raise WorkflowConfigError(
                     f'Polling task "{name}" must configure a target status or'
                     f' output in the graph (:{graph_trigger}) or in its task'
                     f' definition (output = "{config_trigger}") but not both.'
                 )
-            elif graph_trigger is not None:
+            if graph_trigger is not None:
                 comstr += f":{graph_trigger}"
-            elif config_trigger is not None:
-                # quote this; it may be a task message
-                comstr += f':"{config_trigger}" --message'
+            elif config_output is not None:
+                comstr += f":{config_trigger} --output"
+            elif config_message is not None:
+                # quote: may contain spaces
+                comstr += f':"{config_message}" --message'
             else:
                 # default to :succeeded
                 comstr += f":{TASK_OUTPUT_SUCCEEDED}"

cylc/flow/dbstatecheck.py

@@ -41,7 +41,19 @@
 
 
 class CylcWorkflowDBChecker:
-    """Object for querying a workflow database."""
+    """Object for querying task status or outputs from a workflow database.
+
+    Back-compat and task outputs:
+        # Cylc 7 stored {trigger: message} for custom outputs only.
+        1|foo|{"x": "the quick brown"}
+
+        # Cylc 8 (pre-8.3.0) stored [message] only, for all outputs.
+        1|foo|[1]|["submitted", "started", "succeeded", "the quick brown"]
+
+        # Cylc 8 (8.3.0+) stores {trigger: message} for all ouputs.
+        1|foo|[1]|{"submitted": "submitted", "started": "started",
+                   "succeeded": "succeeded", "x": "the quick brown"}
+    """
 
     def __init__(self, rund, workflow, db_path=None):
         # (Explicit dp_path arg is to make testing easier).
@@ -268,12 +280,12 @@ def workflow_state_query(
         results = []
         for row in db_res:
             outputs_map = json.loads(row[2])
-            if self.back_compat_mode or is_message:
+            if is_message:
                 # task message
                 try:
                     outputs = list(outputs_map.values())
                 except AttributeError:
-                    # Cylc 8 pre 8.3.0 back-compat: only output messages stored
+                    # Cylc 8 pre 8.3.0 back-compat: list of output messages
                     outputs = list(outputs_map)
             else:
                 # task output