feat: Add outputs to step prepration response

Author: Alan Christie

workflow/decoder.py

@@ -7,6 +7,7 @@
 from dataclasses import dataclass
 from typing import Any
 
+import decoder.decoder as job_defintion_decoder
 import jsonschema
 import yaml
 
@@ -82,6 +83,13 @@ def get_description(definition: dict[str, Any]) -> str | None:
     return definition.get("description")
 
 
+def is_workflow_output_variable(definition: dict[str, Any], variable_name: str) -> bool:
+    """True if the variable name is in the workflow variables outputs list."""
+    # We can safely pass on the workflow defitnion as its
+    # root-level 'variables' block complies with job-defintion variables.
+    return variable_name in job_defintion_decoder.get_outputs(definition)
+
+
 def get_workflow_variable_names(definition: dict[str, Any]) -> set[str]:
     """Given a Workflow definition this function returns all the names of the
     variables defined in steps that need to be defined at the workflow level.
@@ -97,40 +105,6 @@ def get_workflow_variable_names(definition: dict[str, Any]) -> set[str]:
     return wf_variable_names
 
 
-def get_step_output_variable_names(
-    definition: dict[str, Any], step_name: str
-) -> list[str]:
-    """Given a Workflow definition and a Step name this function returns all the names
-    of the output variables defined at the Step level. These are the names
-    of variables that have files assocaited with them that need copying to
-    the Project directory (from the Instance)."""
-    variable_names: list[str] = []
-    steps: list[dict[str, Any]] = get_steps(definition)
-    for step in steps:
-        if step["name"] == step_name and "plumbing" in step:
-            for v_map in step["plumbing"]:
-                if "to-project" in v_map:
-                    variable_names.append(v_map["variable"])
-    return variable_names
-
-
-def get_step_input_variable_names(
-    definition: dict[str, Any], step_name: str
-) -> list[str]:
-    """Given a Workflow definition and a Step name this function returns all the names
-    of the input variables defined at the Step level. These are the names
-    of variables that have files assocaited with them that need copying to
-    the Instance directory (from the Project)."""
-    variable_names: list[str] = []
-    steps: list[dict[str, Any]] = get_steps(definition)
-    for step in steps:
-        if step["name"] == step_name and "plumbing" in step:
-            for v_map in step["plumbing"]:
-                if "from-project" in v_map:
-                    variable_names.append(v_map["variable"])
-    return variable_names
-
-
 def get_step_workflow_variable_connections(
     *, step_definition: dict[str, Any]
 ) -> list[Connector]:

workflow/workflow_abc.py

@@ -58,6 +58,10 @@ class LaunchParameters:
     # These are required so that the step can access project files.
     # It is a set project-relative filenames (or directroies).
     step_project_inputs: set[str] | None = None
+    # A set of step instance files that are expected to be hard-linked
+    # into the surrounding Project directory.
+    # It is a set instance-relative filenames (or directroies).
+    step_project_outputs: set[str] | None = None
     # The application ID (a custom resource name)
     # used to identify the 'type' of Instance to create.
     # For DM Jobs this will be 'datamanagerjobs.squonk.it'

workflow/workflow_engine.py

@@ -46,6 +46,7 @@
     get_step_predefined_variable_connections,
     get_step_prior_step_connections,
     get_step_workflow_variable_connections,
+    is_workflow_output_variable,
 )
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -56,12 +57,13 @@
 @dataclass
 class StepPreparationResponse:
     """Step preparation response object. 'replicas' is +ve (non-zero) if a step
-    can be launched - it's value indicates how many times. If a step can be launched
+    can be launched - its value indicates how many times. If a step can be launched
     'variables' will not be None. If a parallel set of steps can take place
     (even just one) 'replica_variable' will be set and 'replica_values'
     will be a list containing a value for each step instance. If the step
     depends on a prior step the instance UUIDs of the steps will be listed
-    in the 'dependent_instances' string list.
+    in the 'dependent_instances' string list. If a step's outputs (files) are expected
+    in the project directory they will be listed in 'outputs'.
 
     If preparation fails 'error_num' wil be set, and 'error_msg'
     should contain something useful."""
@@ -71,6 +73,7 @@ class StepPreparationResponse:
     replica_variable: str | None = None
     replica_values: list[str] | None = None
     dependent_instances: set[str] | None = None
+    outputs: set[str] | None = None
     error_num: int = 0
     error_msg: str | None = None
 
@@ -479,6 +482,13 @@ def _prepare_step(
         # I think we can start this step,
         # so compile a set of variables for it.
 
+        # Outputs - a list of step files that are outputs,
+        # and also designated as workflow outputs.
+        # Any step can write files to the Projetc directory
+        # but only job outputs that are also workflow outputs
+        # are put in this list.
+        outputs: set[str] = set()
+
         # Start with any variables provided in the step's specification.
         # A map that we will add to (and maybe even over-write)...
         variables: dict[str, Any] = step_definition["specification"].get(
@@ -498,6 +508,8 @@ def _prepare_step(
         ):
             assert connector.in_ in rwf_variables
             variables[connector.out] = rwf_variables[connector.in_]
+            if is_workflow_output_variable(wf, connector.in_):
+                outputs.add(rwf_variables[connector.in_])
 
         # Process the step's "plumbing" relating to pre-defined variables.
         for connector in get_step_predefined_variable_connections(
@@ -607,6 +619,7 @@ def _prepare_step(
             replica_variable=iter_variable,
             replica_values=iter_values,
             dependent_instances=dependent_instances,
+            outputs=outputs,
         )
 
     def _launch(
@@ -623,6 +636,16 @@ def _launch(
         rwf_id: str = rwf["id"]
         project_id = rwf["project"]["id"]
 
+        _LOGGER.info("SPR.variable=%s", step_preparation_response.variables)
+        _LOGGER.info(
+            "SPR.replica_variable=%s", step_preparation_response.replica_variable
+        )
+        _LOGGER.info("SPR.replica_values=%s", step_preparation_response.replica_values)
+        _LOGGER.info(
+            "SPR.dependent_instances=%s", step_preparation_response.dependent_instances
+        )
+        _LOGGER.info("SPR.outputs=%s", step_preparation_response.outputs)
+
         # Total replicas must be 1 or more
         total_replicas: int = step_preparation_response.replicas
         assert total_replicas >= 1

workflow/workflow_validator.py

@@ -5,7 +5,6 @@
 from typing import Any
 
 from .decoder import (
-    get_step_output_variable_names,
     get_steps,
     get_workflow_variable_names,
     validate_schema,
@@ -86,26 +85,11 @@ def _validate_tag_level(
         # and all the output variable names in the step are unique.
         duplicate_names: set[str] = set()
         all_step_names: set[str] = set()
-        variable_names: set[str] = set()
         for step in get_steps(workflow_definition):
             step_name: str = step["name"]
             if step_name not in duplicate_names and step_name in all_step_names:
                 duplicate_names.add(step_name)
             all_step_names.add(step_name)
-            # Are output variable names unique?
-            variable_names.clear()
-            step_variables: list[str] = get_step_output_variable_names(
-                workflow_definition, step_name
-            )
-            for step_variable in step_variables:
-                if step_variable in variable_names:
-                    return ValidationResult(
-                        error_num=3,
-                        error_msg=[
-                            f"Duplicate step output variable: {step_variable} (step={step_name})"
-                        ],
-                    )
-                variable_names.add(step_variable)
         if duplicate_names:
             return ValidationResult(
                 error_num=2,