docs: Doc tweak

Author: Alan Christie

workflow/workflow_engine.py

@@ -1,4 +1,25 @@
 """The WorkflowEngine execution logic.
+
+It responds to Pod and Workflow protocol buffer messages received by its
+'handle_message()' function, messages delivered by the message handler in the PBC Pod.
+There are no other methods in this class.
+
+Its role is to translate a pre-validated workflow definition into the ordered execution
+of step "Jobs" that manifest as Pod "Instances" that run in a project directory in the
+DM.
+
+Workflow messages initiate (START) and terminate (STOP) workflows. Pod messages signal
+the end of individual workflow steps and carry the exit code of the executed Job.
+The engine used START messages to launch the first "step" in a workflow and the Pod
+messages to signal the success (or failure) of a prior step. A step's success is used,
+along with it's original workflow definition to determine the next action
+(run the next step or signal the end of the workflow).
+
+Before a START message is transmitted the author (typically the Workflow Validator)
+will have created a RunningWorkflow record in the DM. The ID of this record is passed
+in the START message that is sent. The engine uses this ID to find the running workflow
+and the workflow. The engine creates RunningWorkflowStep records for each step that
+is executed, and it uses thew InstanceLauncher to launch the Job (a Pod) for each step.
 """
 
 import logging
@@ -8,17 +29,15 @@
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 from informaticsmatters.protobuf.datamanager.workflow_message_pb2 import WorkflowMessage
 
-from .workflow_abc import APIAdapter, InstanceLauncher
+from .workflow_abc import APIAdapter, InstanceLauncher, LaunchResult
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 _LOGGER.setLevel(logging.INFO)
 _LOGGER.addHandler(logging.StreamHandler(sys.stdout))
 
 
 class WorkflowEngine:
-    """The workflow engine. An event-driven engine that manages the execution
-    of validated workflow definitions.
-    """
+    """The workflow engine."""
 
     def __init__(
         self,
@@ -31,35 +50,29 @@ def __init__(
         self._instance_launcher = instance_launcher
 
     def handle_message(self, msg: Message) -> None:
-        """Expect Pod and Workflow messages.
-
-        WorkflowMessages signal the need to start (or stop) validated workflows -
-        i.e. take a workflow and create a running workflows (and its steps).
-
-        PodMessages signal the completion of a Pod (an Instance/Job) that is part
-        of a running workflow. Given a Pod Message, we use it to identify the
-        Instance exit code, running workflow, and step and decide what to do next,
-        which might be to run the next step in the workflow or abandon the workflow.
+        """Expect Workflow and Pod messages.
 
         Only pod messages relating to workflow instances will be delivered to this method.
-        The Pod message has an 'instance' property that provides the UUID of
+        The Pod message has an 'instance' property that contains the UUID of
         the instance that was run. This is used to correlate the instance with the
-        running workflow step.
+        running workflow step, and (ultimately the running workflow and workflow).
         """
         assert msg
 
-        _LOGGER.debug("> WE.handle_message() : GOT WorkflowMessage:\n%s", str(msg))
+        _LOGGER.debug("Message:\n%s", str(msg))
 
-        # Is this a WorkflowMessage or a PodMessage?
         if isinstance(msg, PodMessage):
             self._handle_pod_message(msg)
         else:
             self._handle_workflow_message(msg)
 
     def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
-        """Handles a WorkflowMessage. This is a message that signals a START or STOP
-        of a workflow. On START we will load the workflow definition and run (launch)
-        the first step."""
+        """WorkflowMessages signal the need to start (or stop) a workflow using its
+        'action' string field (one of 'START' or 'START').
+        The message contains a 'running_workflow' field that contains the UUID
+        of an existing RunningWorkflow record in the DM. Using this
+        we can locate the Workflow record and interrogate that to identify which
+        step (or steps) to launch (run) first."""
         assert msg
 
         # ALL THIS CODE ADDED SIMPLY TO DEMONSTRATE THE USE OF THE API ADAPTER
@@ -71,40 +84,51 @@ def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
 
         _LOGGER.debug("WE> WorkflowMessage:\n%s", str(msg))
 
-        if msg.action == "START":
-            # Using the running workflow get the workflow definition
+        action = msg.action
+        r_wfid = msg.running_workflow
+
+        assert action in ["START", "STOP"]
+        if action == "START":
+            # Using the running workflow...
             response = self._api_adapter.get_running_workflow(
-                running_workflow_id=msg.running_workflow
+                running_workflow_id=r_wfid
             )
             assert "running_workflow" in response
             running_workflow = response["running_workflow"]
-            _LOGGER.info("RunningWorkflow: %s", running_workflow)
-
+            _LOGGER.debug("RunningWorkflow: %s", running_workflow)
+            # ...get the workflow definition...
             workflow_id = running_workflow["workflow"]["id"]
             response = self._api_adapter.get_workflow(workflow_id=workflow_id)
             assert "workflow" in response
-
             workflow = response["workflow"]
-            # Now find the first step
-            # and create a RunningWorkflowStep record prior to launching the instance
+
+            # Now find the first step and create a RunningWorkflowStep record...
+            first_step: str = workflow["steps"][0]["name"]
             response = self._api_adapter.create_running_workflow_step(
-                running_workflow_id=msg.running_workflow,
-                step=workflow["steps"][0]["name"],
+                running_workflow_id=r_wfid,
+                step=first_step,
             )
             assert "id" in response
             running_workflow_step_id = response["id"]
-            # The step specification is a string here - pass it directly to the launcher
-            # which will get the Job command and apply the provided variables to it.
+
+            # The step's 'specification' is a string here - pass it directly to the
+            # launcher along with any appropriate 'variables'. The launcher
+            # will get the step's Job command and apply the variables to it to
+            # form the command that will be executed for the step.
             step = workflow["steps"][0]
             project_id = running_workflow["project_id"]
             variables = running_workflow["variables"]
-            self._instance_launcher.launch(
+            lr: LaunchResult = self._instance_launcher.launch(
                 project_id=project_id,
                 running_workflow_id=msg.running_workflow,
                 running_workflow_step_id=running_workflow_step_id,
                 step_specification=step["specification"],
                 variables=variables,
             )
+            assert lr.error == 0
+            _LOGGER.info(
+                "Launched initial step: %s (command=%s)", first_step, lr.command
+            )
 
         else:
             _LOGGER.info("action=%s", msg.action)
@@ -142,12 +166,7 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
         _LOGGER.debug(
             "WE> PodMessage: instance=%s, exit_code=%d", instance_id, exit_code
         )
-
-        # Ignore instances without a running workflow step
         response = self._api_adapter.get_instance(instance_id=instance_id)
-        if "running_workflow_step" not in response:
-            _LOGGER.warning("WE> PodMessage: Without running_workflow_step")
-            return
         running_workflow_step_id: str = response["running_workflow_step"]
         response = self._api_adapter.get_running_workflow_step(
             running_workflow_step_id=running_workflow_step_id
@@ -176,9 +195,6 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
         if success:
             # Given the step for the instance just finished,
             # find the next step in the workflow and launch it.
-            # If there are no more steps then the workflow is done
-            # so we need to set the running workflow as done
-            # and set it's success status too.
             workflow = response["workflow"]
             for step in workflow["steps"]:
                 if step["name"] == step_name:
@@ -193,16 +209,25 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
                         running_workflow_step_id = response["id"]
                         project_id = running_workflow["project_id"]
                         variables = running_workflow["variables"]
-                        self._instance_launcher.launch(
+                        lr: LaunchResult = self._instance_launcher.launch(
                             project_id=project_id,
                             running_workflow_id=running_workflow_id,
                             running_workflow_step_id=running_workflow_step_id,
                             step_specification=next_step["specification"],
                             variables=variables,
                         )
                         end_of_workflow = False
+                        assert lr.error == 0
+                        _LOGGER.info(
+                            "Launched step: %s (command=%s)",
+                            next_step["name"],
+                            lr.command,
+                        )
                         break
 
+        # If there are no more steps then the workflow is done
+        # so we need to set the running workflow as done
+        # and set its success status too.
         if end_of_workflow:
             self._api_adapter.set_running_workflow_done(
                 running_workflow_id=running_workflow_id,