Refactor names and comments

Author: erlenlh

docs/ert/reference/workflows/complete_workflows.rst

@@ -88,6 +88,9 @@ Observe that the workflows being 'hooked in' with the
 :code:`HOOK_WORKFLOW` must be loaded with the :code:`LOAD_WORKFLOW`
 keyword.
 
+Workflow output
+-----------------
+
 Output from workflow jobs is written to the ERT log. Every job invocation gets an
 entry holding whatever the job wrote to stdout and stderr::
 
@@ -105,9 +108,14 @@ Workflows hooked in with :code:`HOOK_WORKFLOW` are in addition recorded
 alongside the experiment they belong to, in
 :code:`<ENSPATH>/experiments/<experiment_id>/workflow_events.jsonl`. That file
 holds one JSON object per job invocation and exists so the GUI can show the
-output of a workflow again later; it is not meant to be read directly. Output
-from hooks that run before the experiment is created, such as
-:code:`PRE_EXPERIMENT`, is held back and written once the experiment exists.
+output of a workflow again later; it is not meant to be read directly. To view this
+in the GUI, select the experiment in the *Experiments* tool, then select the
+*Workflow events* tab.
+
+The GUI will show a list of all workflow jobs that have
+been run for that experiment, and clicking on a job will show its output.
+Output from hooks that run before the experiment is created, such as
+:code:`PRE_EXPERIMENT`, is held back and written once the storage is created.
 
 .. _runpath-file-workflows:
 

src/ert/config/ert_script.py

@@ -153,8 +153,8 @@ def _capturing(stream_name: str) -> Iterator[io.StringIO]:
 class ExternalScriptError(RuntimeError):
     """Raised when an external workflow job exits with a non-zero exit code.
 
-    Reported without a stack trace, since the trace would only show the ert
-    internals that started the job, and not what went wrong inside it.
+    Reported without a stack trace, since it would only
+    show ert internals and could be confusing
     """
 
 

src/ert/run_models/event.py

@@ -93,10 +93,10 @@ def write_as_csv(self, output_path: Path | None) -> None:
             self.data.to_csv("Report", output_path / str(self.run_id))
 
 
-class RunModelWorkflowLogEvent(BaseModel, extra="forbid"):
+class WorkflowEvent(BaseModel, extra="forbid"):
     """The output of a single workflow job invocation."""
 
-    event_type: Literal["RunModelWorkflowLogEvent"] = "RunModelWorkflowLogEvent"
+    event_type: Literal["WorkflowEvent"] = "WorkflowEvent"
     run_id: UUID
     hook: str
     workflow_name: str
@@ -144,10 +144,10 @@ class RunPathCreatedEvent(RunPathCreationEvent):
     | RunModelTimeEvent
     | RunModelUpdateBeginEvent
     | RunModelUpdateEndEvent
-    | RunModelWorkflowLogEvent
     | SnapshotUpdateEvent
     | StartEvent
     | WarningEvent
+    | WorkflowEvent
     | EnsembleEvaluationWarning
     | StartingTotalRunPathCreationEvent
     | FinishedTotalRunPathCreationEvent

src/ert/run_models/run_model.py

@@ -81,10 +81,10 @@
 from .event import (
     EndEvent,
     FullSnapshotEvent,
-    RunModelWorkflowLogEvent,
     SnapshotUpdateEvent,
     StartEvent,
     StatusEvents,
+    WorkflowEvent,
 )
 
 if TYPE_CHECKING:
@@ -180,10 +180,8 @@ class RunModel(RunModelConfig, ABC):
     _start_iteration: int = PrivateAttr(default=0)
     _max_parallelism_violation: ParallelismViolation = ParallelismViolation()
     _workflow_runner: WorkflowRunner | None = PrivateAttr(default=None)
-    _workflow_log_id: uuid.UUID = PrivateAttr(default_factory=uuid.uuid4)
-    _pending_workflow_events: list[RunModelWorkflowLogEvent] = PrivateAttr(
-        default_factory=list
-    )
+    _workflow_run_id: uuid.UUID = PrivateAttr(default_factory=uuid.uuid4)
+    _pending_workflow_events: list[WorkflowEvent] = PrivateAttr(default_factory=list)
 
     def __init__(
         self,
@@ -386,7 +384,7 @@ def handle_captured_event(message: Warning | str) -> None:
             self.send_event(WarningEvent(msg=str(message)))
 
         start_timestamp = datetime.datetime.now(tz=datetime.UTC)
-        self._workflow_log_id = uuid.uuid4()
+        self._workflow_run_id = uuid.uuid4()
         self._pending_workflow_events = []
         try:  # ruff: ignore[too-many-statements-in-try-clause]
             self.send_event(StartEvent(timestamp=start_timestamp))
@@ -847,11 +845,8 @@ def run_workflows(
         try:
             for workflow in self.hooked_workflows[fixtures.hook]:
                 if self._end_event.is_set():
-                    # The experiment was already cancelled before this
-                    # workflow got a chance to start. Report every one of
-                    # its jobs as cancelled so it still shows up in the
-                    # workflow log, instead of disappearing silently.
-                    self._send_cancelled_workflow_log_events(
+                    # Cancel all remaining workflows
+                    self._send_cancelled_workflow_events(
                         workflow=workflow,
                         hook=fixtures.hook,
                         iteration=iteration,
@@ -868,28 +863,28 @@ def run_workflows(
                     workflow_runner.run_blocking()
                 finally:
                     self._workflow_runner = None
-                    self._send_workflow_log_events(
+                    self._send_workflow_events(
                         workflow_runner=workflow_runner,
                         hook=fixtures.hook,
                         workflow_name=workflow.name,
                         iteration=iteration,
                     )
         finally:
-            self._persist_workflow_log(experiment)
+            self._persist_workflow_events_to_storage(experiment)
 
         if self._end_event.is_set():
             raise UserCancelled("Experiment cancelled by user during workflows")
 
-    def _send_workflow_log_events(
+    def _send_workflow_events(
         self,
         workflow_runner: WorkflowRunner,
         hook: HookRuntime,
         workflow_name: str,
         iteration: int | None,
     ) -> None:
         events = [
-            RunModelWorkflowLogEvent(
-                run_id=self._workflow_log_id,
+            WorkflowEvent(
+                run_id=self._workflow_run_id,
                 hook=str(hook),
                 workflow_name=workflow_name,
                 job_name=result.name,
@@ -908,20 +903,17 @@ def _send_workflow_log_events(
             self.send_event(event)
         self._pending_workflow_events.extend(events)
 
-    def _send_cancelled_workflow_log_events(
+    def _send_cancelled_workflow_events(
         self,
         workflow: Workflow,
         hook: HookRuntime,
         iteration: int | None,
     ) -> None:
-        """Report every job in a hooked workflow that never got a chance to
-        start because the experiment was already cancelled, so it still
-        shows up in the workflow log instead of disappearing silently.
-        """
+        # Report jobs not started due to cancellation
         now = datetime.datetime.now(tz=datetime.UTC)
         events = [
-            RunModelWorkflowLogEvent(
-                run_id=self._workflow_log_id,
+            WorkflowEvent(
+                run_id=self._workflow_run_id,
                 hook=str(hook),
                 workflow_name=workflow.name,
                 job_name=job.name,
@@ -940,18 +932,18 @@ def _send_cancelled_workflow_log_events(
             self.send_event(event)
         self._pending_workflow_events.extend(events)
 
-    def _persist_workflow_log(self, experiment: Experiment | None) -> None:
-        """Output from hooks that run before the experiment exists in storage,
-        such as PRE_EXPERIMENT, is held back until an experiment is available.
-        """
+    def _persist_workflow_events_to_storage(
+        self, experiment: Experiment | None
+    ) -> None:
+        # Hold back output until storage is created
         if experiment is None or not self._pending_workflow_events:
             return
         try:
             experiment.append_workflow_events(
                 event.model_dump_json() for event in self._pending_workflow_events
             )
         except Exception:
-            logger.exception("Failed to persist workflow log to storage")
+            logger.exception("Failed to persist workflow events to storage")
         self._pending_workflow_events = []
 
     def _evaluate_and_postprocess(

src/ert/workflow_runner.py

@@ -21,8 +21,6 @@
 
 @dataclass
 class WorkflowJobResult:
-    """The outcome of a single invocation of a workflow job."""
-
     name: str
     index: int
     arguments: list[str]
@@ -43,14 +41,6 @@ def status(self) -> str:
         return "success"
 
     def as_log_entry(self, workflow_name: str, hook: str | None = None) -> str:
-        """A multi-line rendering of this result for the ERT log.
-
-        Args:
-            workflow_name: The workflow the job was invoked from.
-            hook: The runtime the workflow was hooked to, if any. Workflows
-                started from the command line or the GUI's Run workflow tool
-                are not hooked to anything.
-        """
         header = "Workflow job"
         if hook is not None:
             header += f" {hook}"
@@ -160,14 +150,6 @@ def __init__(
         fixtures: WorkflowFixtures,
         hook: str | None = None,
     ) -> None:
-        """
-        Args:
-            workflow: The workflow to run.
-            fixtures: The fixtures made available to the workflow's jobs.
-            hook: The runtime the workflow was hooked to, used to identify it
-                in the log. Workflows started from the command line or the
-                GUI's Run workflow tool are not hooked to anything.
-        """
         self.__workflow = workflow
         self.fixtures = fixtures
         self._hook = hook
@@ -211,9 +193,7 @@ def run_blocking(self) -> None:
 
         for index, (job, args) in enumerate(self.__workflow):
             if self.__cancelled:
-                # The workflow was already cancelled before this job got a
-                # chance to start - record it as cancelled rather than
-                # silently dropping it.
+                # The workflow was cancelled before this job started
                 result = WorkflowJobResult(
                     name=job.name,
                     index=index,
@@ -232,10 +212,7 @@ def run_blocking(self) -> None:
             logger.info(f"Workflow job {jobrunner.name} starting")
             jobrunner.run(args, fixtures=self.fixtures)
             job_was_cancelled = self.__cancelled
-            # A job that was interrupted by a cancellation request did not
-            # complete, so it is reported as failed rather than succeeded -
-            # cooperatively cancelled internal jobs otherwise return
-            # normally and hasFailed() would stay False.
+
             failed = jobrunner.hasFailed() or job_was_cancelled
             self.__status[jobrunner.name] = {
                 "stdout": jobrunner.stdoutdata(),
@@ -253,8 +230,6 @@ def run_blocking(self) -> None:
             self.__job_results.append(result)
 
             extra = self._log_extra(result, execution_type=jobrunner.execution_type)
-            # Logged before acting on stop_on_fail, so the output of the job
-            # that aborted the workflow is in the log too.
             if failed and not job_was_cancelled:
                 logger.error(self._log_entry(result), extra=extra)
             else:

tests/ert/ui_tests/cli/test_cli.py

@@ -30,7 +30,7 @@
     ES_MDA_MODE,
     TEST_RUN_MODE,
 )
-from ert.run_models.event import RunModelWorkflowLogEvent
+from ert.run_models.event import WorkflowEvent
 from ert.sample_prior import sample_prior
 from ert.scheduler.driver import Driver
 from ert.scheduler.job import Job
@@ -546,7 +546,7 @@ def test_that_workflow_output_is_written_to_the_experiment_in_storage():
         (line,) = experiment.workflow_events_path.read_text(
             encoding="utf-8"
         ).splitlines()
-    event = RunModelWorkflowLogEvent.model_validate_json(line)
+    event = WorkflowEvent.model_validate_json(line)
     assert event.hook == "PRE_SIMULATION"
     assert event.workflow_name == "wfprint"
     assert event.job_name == "printjob"

tests/ert/unit_tests/run_models/test_base_run_model.py

@@ -38,7 +38,7 @@
 from ert.mode_definitions import TEST_RUN_MODE
 from ert.plugins import ErtRuntimePlugins
 from ert.run_models import create_model
-from ert.run_models.event import RunModelWorkflowLogEvent
+from ert.run_models.event import WorkflowEvent
 from ert.run_models.run_model import (
     RunModel,
     UserCancelled,
@@ -899,11 +899,11 @@ def _drain(status_queue):
 
 
 def _persisted_workflow_events(experiment):
-    """The workflow log events stored for an experiment, oldest first."""
+    """The workflow events persisted to storage for an experiment, oldest first."""
     if not experiment.workflow_events_path.exists():
         return []
     return [
-        RunModelWorkflowLogEvent.model_validate_json(line)
+        WorkflowEvent.model_validate_json(line)
         for line in experiment.workflow_events_path.read_text(
             encoding="utf-8"
         ).splitlines()
@@ -929,7 +929,7 @@ def _printing_workflow(tmp_path, name, script, *, stop_on_fail=False):
     )
 
 
-def test_that_run_workflows_sends_a_workflow_log_event_per_job(tmp_path, use_tmpdir):
+def test_that_run_workflows_sends_a_workflow_event_per_job(tmp_path, use_tmpdir):
     workflow = _printing_workflow(tmp_path, "hello", 'print("hello from workflow")')
     workflow.cmd_list.append(workflow.cmd_list[0])
     status_queue = SimpleQueue()
@@ -946,11 +946,11 @@ def test_that_run_workflows_sends_a_workflow_log_event_per_job(tmp_path, use_tmp
         ("HELLO", 1, "hello from workflow\n"),
     ]
     assert all(e.hook == "PRE_EXPERIMENT" for e in events)
-    assert all(e.run_id == brm._workflow_log_id for e in events)
+    assert all(e.run_id == brm._workflow_run_id for e in events)
     assert not any(e.failed for e in events)
 
 
-def test_that_a_workflow_log_event_is_sent_when_stop_on_fail_aborts_the_workflow(
+def test_that_a_workflow_event_is_sent_when_stop_on_fail_aborts_the_workflow(
     tmp_path, use_tmpdir
 ):
     workflow = _printing_workflow(
@@ -1009,7 +1009,7 @@ def test_that_a_cancelled_job_and_its_unstarted_siblings_carry_the_workflow_name
         ),
     ]
 
-    brm._send_workflow_log_events(
+    brm._send_workflow_events(
         workflow_runner=workflow_runner,
         hook=HookRuntime.PRE_SIMULATION,
         workflow_name="my_workflow",
@@ -1026,7 +1026,7 @@ def test_that_a_cancelled_job_and_its_unstarted_siblings_carry_the_workflow_name
     assert never_started_event.cancelled
 
 
-def test_that_workflow_log_events_from_an_update_hook_carry_the_iteration(
+def test_that_workflow_events_from_an_update_hook_carry_the_iteration(
     tmp_path, use_tmpdir
 ):
     workflow = _printing_workflow(tmp_path, "hello", 'print("hello")')
@@ -1119,7 +1119,7 @@ def test_that_pre_experiment_output_is_persisted_once_an_experiment_exists(
     ]
 
 
-def test_that_a_failure_to_persist_the_workflow_log_does_not_stop_the_experiment(
+def test_that_a_failure_to_persist_workflow_events_does_not_stop_the_experiment(
     tmp_path, use_tmpdir, caplog
 ):
     workflow = _printing_workflow(tmp_path, "hello", 'print("hello from workflow")')
@@ -1143,7 +1143,7 @@ def test_that_a_failure_to_persist_the_workflow_log_does_not_stop_the_experiment
             )
         )
 
-    assert "Failed to persist workflow log" in caplog.text
+    assert "Failed to persist workflow events to storage" in caplog.text
     assert _drain(status_queue), "the event should still be sent"
 
 
@@ -1181,14 +1181,14 @@ def test_that_starting_an_experiment_discards_workflow_output_from_the_previous_
     brm = create_run_model()
     brm._status_queue = SimpleQueue()
     brm._pending_workflow_events = [MagicMock()]
-    previous_log_id = brm._workflow_log_id
+    previous_log_id = brm._workflow_run_id
 
     brm.start_simulations_thread(
         EvaluatorServerConfig(use_token=False), rerun_failed_realizations=True
     )
 
     assert brm._pending_workflow_events == []
-    assert brm._workflow_log_id != previous_log_id
+    assert brm._workflow_run_id != previous_log_id
 
 
 def test_that_workflow_output_is_persisted_when_stop_on_fail_aborts_the_workflow(
@@ -1266,7 +1266,7 @@ def test_that_workflows_hooked_after_a_cancelled_one_still_appear_as_cancelled(
     """Regression test: when several workflows are hooked to the same
     runtime and cancellation happens while the first one is running, the
     workflows that come after it in the hook's list must still be reported
-    (as cancelled) rather than silently disappearing from the workflow log.
+    (as cancelled) rather than silently disappearing from the workflow events.
     """
     first = _printing_workflow(tmp_path, "first", 'print("first workflow")')
     second = _printing_workflow(tmp_path, "second", 'print("second workflow")')