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)
     )
@@ -171,6 +172,7 @@ def run_blocking(self) -> None:
             if not self.__cancelled:
                 logger.info(f"Workflow job {jobrunner.name} starting")
                 jobrunner.run(args, fixtures=self.fixtures)
+                job_was_cancelled = self.__cancelled
                 self.__status[jobrunner.name] = {
                     "stdout": jobrunner.stdoutdata(),
                     "stderr": jobrunner.stderrdata(),
@@ -184,6 +186,7 @@ def run_blocking(self) -> None:
                         stdout=jobrunner.stdoutdata(),
                         stderr=jobrunner.stderrdata(),
                         failed=jobrunner.hasFailed(),
+                        cancelled=job_was_cancelled,
                     )
                 )
 
@@ -205,6 +208,10 @@ def run_blocking(self) -> None:
                         )
 
                     logger.error(f"Workflow job {jobrunner.name} failed", extra=info)
+                elif job_was_cancelled:
+                    logger.info(
+                        f"Workflow job {jobrunner.name} was cancelled", extra=info
+                    )
                 else:
                     logger.info(
                         f"Workflow job {jobrunner.name} completed successfully",

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,12 @@ 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
+    assert results[1].cancelled is True
+    assert results[1].failed is False
+
 
 @pytest.mark.slow
 @pytest.mark.usefixtures("use_tmpdir")