Record when a workflow job was cancelled, not just failed

A cooperatively-cancelled internal job returns normally, so hasFailed()
stays False and the job was indistinguishable from one that ran to
completion. WorkflowJobResult and RunModelWorkflowLogEvent now carry a
'cancelled' flag alongside 'failed', and the persisted log entry status
reflects it.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: erlenlh

src/ert/run_models/event.py

@@ -106,11 +106,17 @@ class RunModelWorkflowLogEvent(BaseModel, extra="forbid"):
     stdout: str
     stderr: str
     failed: bool
+    cancelled: bool = False
     timestamp: datetime
     iteration: int | None = None
 
     def as_log_entry(self) -> str:
-        status = "failed" if self.failed else "success"
+        if self.cancelled:
+            status = "cancelled"
+        elif self.failed:
+            status = "failed"
+        else:
+            status = "success"
         header = (
             f"=== {self.timestamp.isoformat(timespec='seconds')} {self.hook} "
             f"workflow={self.workflow_name} job={self.job_name}#{self.job_index} "

src/ert/run_models/run_model.py

@@ -887,6 +887,7 @@ def _send_workflow_log_events(
                 stdout=result.stdout,
                 stderr=result.stderr,
                 failed=result.failed,
+                cancelled=result.cancelled,
                 timestamp=result.timestamp,
                 iteration=iteration,
             )

src/ert/workflow_runner.py

@@ -29,6 +29,7 @@ class WorkflowJobResult:
     stdout: str
     stderr: str
     failed: bool
+    cancelled: bool = False
     timestamp: datetime.datetime = field(
         default_factory=lambda: datetime.datetime.now(tz=datetime.UTC)
     )
@@ -166,51 +167,75 @@ def run_blocking(self) -> None:
         self.__running = True
 
         for index, (job, args) in enumerate(self.__workflow):
-            jobrunner = WorkflowJobRunner(job)
-            self.__current_job = jobrunner
-            if not self.__cancelled:
-                logger.info(f"Workflow job {jobrunner.name} starting")
-                jobrunner.run(args, fixtures=self.fixtures)
-                self.__status[jobrunner.name] = {
-                    "stdout": jobrunner.stdoutdata(),
-                    "stderr": jobrunner.stderrdata(),
-                    "completed": not jobrunner.hasFailed(),
-                }
+            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.
                 self.__job_results.append(
                     WorkflowJobResult(
-                        name=jobrunner.name,
+                        name=job.name,
                         index=index,
                         arguments=[str(arg) for arg in args],
-                        stdout=jobrunner.stdoutdata(),
-                        stderr=jobrunner.stderrdata(),
-                        failed=jobrunner.hasFailed(),
+                        stdout="",
+                        stderr="",
+                        failed=False,
+                        cancelled=True,
                     )
                 )
+                continue
+
+            jobrunner = WorkflowJobRunner(job)
+            self.__current_job = jobrunner
+            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(),
+                "stderr": jobrunner.stderrdata(),
+                "completed": not failed,
+            }
+            self.__job_results.append(
+                WorkflowJobResult(
+                    name=jobrunner.name,
+                    index=index,
+                    arguments=[str(arg) for arg in args],
+                    stdout=jobrunner.stdoutdata(),
+                    stderr=jobrunner.stderrdata(),
+                    failed=failed,
+                )
+            )
 
-                info = {
-                    "class": "WORKFLOW_JOB",
-                    "job_name": jobrunner.name,
-                    "arguments": " ".join(args),
-                    "stdout": jobrunner.stdoutdata(),
-                    "stderr": jobrunner.stderrdata(),
-                    "execution_type": jobrunner.execution_type,
-                }
-
-                if jobrunner.hasFailed():
-                    if jobrunner.stop_on_fail:
-                        self.__running = False
-                        raise RuntimeError(
-                            f"Workflow job {info['job_name']}"
-                            f" failed with error: {info['stderr']}"
-                        )
-
-                    logger.error(f"Workflow job {jobrunner.name} failed", extra=info)
-                else:
-                    logger.info(
-                        f"Workflow job {jobrunner.name} completed successfully",
-                        extra=info,
+            info = {
+                "class": "WORKFLOW_JOB",
+                "job_name": jobrunner.name,
+                "arguments": " ".join(args),
+                "stdout": jobrunner.stdoutdata(),
+                "stderr": jobrunner.stderrdata(),
+                "execution_type": jobrunner.execution_type,
+            }
+
+            if job_was_cancelled:
+                logger.info(f"Workflow job {jobrunner.name} was cancelled", extra=info)
+            elif jobrunner.hasFailed():
+                if jobrunner.stop_on_fail:
+                    self.__running = False
+                    raise RuntimeError(
+                        f"Workflow job {info['job_name']}"
+                        f" failed with error: {info['stderr']}"
                     )
 
+                logger.error(f"Workflow job {jobrunner.name} failed", extra=info)
+            else:
+                logger.info(
+                    f"Workflow job {jobrunner.name} completed successfully",
+                    extra=info,
+                )
+
         self.__current_job = None
         self.__running = False
         self.__workflow_result = True

tests/ert/unit_tests/run_models/test_base_run_model.py

@@ -43,6 +43,7 @@
     UserCancelled,
 )
 from ert.warnings import PostExperimentWarning
+from ert.workflow_runner import WorkflowJobResult
 
 
 @pytest.fixture(autouse=True)
@@ -959,6 +960,59 @@ def test_that_a_workflow_log_event_is_sent_when_stop_on_fail_aborts_the_workflow
     assert event.stdout == "printed before failing\n"
 
 
+def test_that_a_cancelled_job_and_its_unstarted_siblings_carry_the_workflow_name(
+    use_tmpdir,
+):
+    """Regression test: a job interrupted by cancellation is reported as
+    failed (not merely cancelled), the jobs after it that never got a
+    chance to start are reported as cancelled, and both kinds of result
+    still carry the originating workflow's name once turned into events.
+    """
+    status_queue = SimpleQueue()
+    brm = create_run_model(
+        hooked_workflows={},
+        status_queue=status_queue,
+    )
+
+    workflow_runner = MagicMock()
+    workflow_runner.workflowJobResults.return_value = [
+        WorkflowJobResult(
+            name="INTERRUPTED",
+            index=0,
+            arguments=[],
+            stdout="partial output",
+            stderr="",
+            failed=True,
+            cancelled=False,
+        ),
+        WorkflowJobResult(
+            name="NEVER_STARTED",
+            index=1,
+            arguments=[],
+            stdout="",
+            stderr="",
+            failed=False,
+            cancelled=True,
+        ),
+    ]
+
+    brm._send_workflow_log_events(
+        workflow_runner=workflow_runner,
+        hook=HookRuntime.PRE_SIMULATION,
+        workflow_name="my_workflow",
+        iteration=0,
+    )
+
+    interrupted_event, never_started_event = _drain(status_queue)
+    assert interrupted_event.workflow_name == "my_workflow"
+    assert interrupted_event.failed
+    assert not interrupted_event.cancelled
+
+    assert never_started_event.workflow_name == "my_workflow"
+    assert not never_started_event.failed
+    assert never_started_event.cancelled
+
+
 def test_that_workflow_log_events_from_an_update_hook_carry_the_iteration(
     tmp_path, use_tmpdir
 ):

tests/ert/unit_tests/run_models/test_workflow_log_event.py

@@ -53,6 +53,12 @@ def test_that_workflow_log_entry_header_states_hook_workflow_job_and_status():
     )
 
 
+def test_that_a_cancelled_job_is_reported_as_cancelled_even_if_not_failed():
+    entry = _event(cancelled=True, failed=False).as_log_entry()
+
+    assert "status=cancelled" in entry.splitlines()[0]
+
+
 def test_that_workflow_log_entries_are_separated_by_a_blank_line():
     first = _event(job_name="FIRST", stdout="first\n").as_log_entry()
     second = _event(job_name="SECOND", stdout="second\n").as_log_entry()

tests/ert/unit_tests/workflow_runner/test_workflow_runner.py

@@ -228,6 +228,7 @@ def test_that_job_results_contain_one_entry_per_job_invocation():
     ]
     assert [result.stdout for result in results] == ["Hello World\n", "Hello World\n"]
     assert not any(result.failed for result in results)
+    assert not any(result.cancelled for result in results)
 
 
 @pytest.mark.slow
@@ -267,6 +268,18 @@ def test_workflow_thread_cancel_ert_script():
     assert not Path("wait_cancelled_2").exists()
     assert not Path("wait_finished_2").exists()
 
+    results = {result.index: result for result in workflow_runner.workflowJobResults()}
+    assert results[0].cancelled is False
+    assert results[0].failed is False
+    # The job that was interrupted by cancellation did not complete, so it
+    # is reported as failed rather than as a separate "cancelled" status.
+    assert results[1].cancelled is False
+    assert results[1].failed is True
+    # The remaining job never got a chance to start, so it is reported as
+    # cancelled rather than silently omitted.
+    assert results[2].cancelled is True
+    assert results[2].failed is False
+
 
 @pytest.mark.slow
 @pytest.mark.usefixtures("use_tmpdir")