Store workflow output

docs/ert/reference/workflows/complete_workflows.rst

@@ -122,6 +122,14 @@ The :code:`status` field is one of :code:`success`, :code:`failed` or
 jobs that were stopped because the workflow was cancelled are logged at
 :code:`INFO` level.
 
+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 output of a
+workflow can be shown again later; it is not meant to be read directly.
+Output from hooks that ran before the experiment is created, such as
+:code:`PRE_EXPERIMENT`, is held back and written once the storage is created.
+
 .. _runpath-file-workflows:
 
 Locating the realisations: <RUNPATH_FILE>

src/ert/run_models/event.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from datetime import datetime
 from pathlib import Path
 from typing import Annotated, Any, Literal
 from uuid import UUID
@@ -20,6 +21,7 @@
     StartEvent,
     WarningEvent,
 )
+from ert.workflow_runner import WorkflowJobStatus
 
 logger = logging.getLogger(__name__)
 
@@ -93,6 +95,23 @@ def write_as_csv(self, output_path: Path | None) -> None:
             self.data.to_csv("Report", output_path / str(self.run_id))
 
 
+class WorkflowEvent(BaseModel, extra="forbid"):
+    """The output of a single workflow job invocation."""
+
+    event_type: Literal["WorkflowEvent"] = "WorkflowEvent"
+    run_id: UUID
+    hook: str
+    workflow_name: str
+    job_name: str
+    job_index: int
+    arguments: list[str]
+    stdout: str
+    stderr: str
+    status: WorkflowJobStatus
+    timestamp: datetime
+    iteration: int | None = None
+
+
 class RunPathCreationEvent(BaseModel, extra="forbid"):
     pass
 
@@ -129,6 +148,7 @@ class RunPathCreatedEvent(RunPathCreationEvent):
     | SnapshotUpdateEvent
     | StartEvent
     | WarningEvent
+    | WorkflowEvent
     | EnsembleEvaluationWarning
     | StartingTotalRunPathCreationEvent
     | FinishedTotalRunPathCreationEvent

src/ert/run_models/run_model.py

@@ -39,11 +39,13 @@
     ConfigValidationError,
     DesignMatrix,
     HookedWorkflowFixtures,
+    HookRuntime,
     ModelConfig,
     ParameterConfig,
     PostSimulationFixtures,
     PreSimulationFixtures,
     QueueConfig,
+    Workflow,
     create_workflow_fixtures_from_hooked,
 )
 from ert.config.queue_config import KnownQueueOptionsAdapter
@@ -74,7 +76,7 @@
 from ert.trace import tracer
 from ert.utils import log_duration
 from ert.warnings import PostExperimentWarning, capture_specific_warning
-from ert.workflow_runner import WorkflowRunner
+from ert.workflow_runner import WorkflowJobStatus, WorkflowRunner
 
 from ._create_run_path import create_run_path
 from .event import (
@@ -83,6 +85,7 @@
     SnapshotUpdateEvent,
     StartEvent,
     StatusEvents,
+    WorkflowEvent,
 )
 
 if TYPE_CHECKING:
@@ -180,6 +183,8 @@ class RunModel(RunModelConfig, ABC):
     _start_iteration: int = PrivateAttr(default=0)
     _max_parallelism_violation: ParallelismViolation = ParallelismViolation()
     _workflow_runner: WorkflowRunner | None = PrivateAttr(default=None)
+    _workflow_run_id: uuid.UUID = PrivateAttr(default_factory=uuid.uuid4)
+    _pending_workflow_events: list[WorkflowEvent] = PrivateAttr(default_factory=list)
 
     def __init__(
         self,
@@ -380,6 +385,8 @@ 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_run_id = uuid.uuid4()
+        self._pending_workflow_events = []
         try:  # ruff: ignore[too-many-statements-in-try-clause]
             self.send_event(StartEvent(timestamp=start_timestamp))
             with (
@@ -831,24 +838,110 @@ def run_workflows(
         self,
         fixtures: HookedWorkflowFixtures,
     ) -> None:
-        for workflow in self.hooked_workflows[fixtures.hook]:
-            workflow_runner = WorkflowRunner(
-                workflow=workflow,
-                fixtures=create_workflow_fixtures_from_hooked(fixtures),
-                hook=str(fixtures.hook),
-            )
-            self._workflow_runner = workflow_runner
-            try:
+        ensemble = getattr(fixtures, "ensemble", None)
+        experiment = ensemble.experiment if ensemble is not None else None
+        iteration = ensemble.iteration if ensemble is not None else None
+        try:
+            for workflow in self.hooked_workflows[fixtures.hook]:
                 if self._end_event.is_set():
-                    workflow_runner.cancel()
-                    raise UserCancelled("Experiment cancelled by user during workflows")
+                    # Cancel all remaining workflows
+                    self._send_cancelled_workflow_events(
+                        workflow=workflow,
+                        hook=fixtures.hook,
+                        iteration=iteration,
+                    )
+                    continue
 
-                workflow_runner.run_blocking()
-            finally:
-                self._workflow_runner = None
+                workflow_runner = WorkflowRunner(
+                    workflow=workflow,
+                    fixtures=create_workflow_fixtures_from_hooked(fixtures),
+                    hook=str(fixtures.hook),
+                )
+                self._workflow_runner = workflow_runner
+                try:
+                    workflow_runner.run_blocking()
+                finally:
+                    self._workflow_runner = None
+                    self._send_workflow_events(
+                        workflow_runner=workflow_runner,
+                        hook=fixtures.hook,
+                        workflow_name=workflow.name,
+                        iteration=iteration,
+                    )
+        finally:
+            self._persist_workflow_events_to_storage(experiment)
 
-            if self._end_event.is_set():
-                raise UserCancelled("Experiment cancelled by user during workflows")
+        if self._end_event.is_set():
+            raise UserCancelled("Experiment cancelled by user during workflows")
+
+    def _send_workflow_events(
+        self,
+        workflow_runner: WorkflowRunner,
+        hook: HookRuntime,
+        workflow_name: str,
+        iteration: int | None,
+    ) -> None:
+        events = [
+            WorkflowEvent(
+                run_id=self._workflow_run_id,
+                hook=str(hook),
+                workflow_name=workflow_name,
+                job_name=result.name,
+                job_index=result.index,
+                arguments=result.arguments,
+                stdout=result.stdout,
+                stderr=result.stderr,
+                status=result.status,
+                timestamp=result.timestamp,
+                iteration=iteration,
+            )
+            for result in workflow_runner.workflow_job_results()
+        ]
+        for event in events:
+            self.send_event(event)
+        self._pending_workflow_events.extend(events)
+
+    def _send_cancelled_workflow_events(
+        self,
+        workflow: Workflow,
+        hook: HookRuntime,
+        iteration: int | None,
+    ) -> None:
+        # Report jobs not started due to cancellation
+        now = datetime.datetime.now(tz=datetime.UTC)
+        events = [
+            WorkflowEvent(
+                run_id=self._workflow_run_id,
+                hook=str(hook),
+                workflow_name=workflow.name,
+                job_name=job.name,
+                job_index=index,
+                arguments=[str(arg) for arg in args],
+                stdout="",
+                stderr="",
+                status=WorkflowJobStatus.CANCELLED,
+                timestamp=now,
+                iteration=iteration,
+            )
+            for index, (job, args) in enumerate(workflow)
+        ]
+        for event in events:
+            self.send_event(event)
+        self._pending_workflow_events.extend(events)
+
+    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 events to storage")
+        self._pending_workflow_events = []
 
     def _evaluate_and_postprocess(
         self,

src/ert/storage/local_experiment.py

@@ -5,7 +5,7 @@
 import json
 import logging
 import shutil
-from collections.abc import Generator
+from collections.abc import Generator, Iterable
 from datetime import UTC, datetime
 from enum import StrEnum, auto
 from functools import cached_property
@@ -916,6 +916,23 @@ def export_everest_opt_results_to_csv(self) -> Path:
     def status_snapshot_path(self, iteration: int) -> Path:
         return self._path / "status" / f"iteration-{iteration}.json"
 
+    @property
+    def workflow_events_path(self) -> Path:
+        """The events describing the workflows run for this experiment.
+
+        One JSON-serialized workflow log event per line, used to repopulate
+        the workflow view when an experiment is opened again later.
+        """
+        return self._path / "workflow_events.jsonl"
+
+    @require_write
+    def append_workflow_events(self, lines: Iterable[str]) -> None:
+        """Append already JSON-serialized workflow log events, one per line."""
+        events_path = self.workflow_events_path
+        events_path.parent.mkdir(parents=True, exist_ok=True)
+        with events_path.open("a", encoding="utf-8") as fout:
+            fout.writelines(f"{line}\n" for line in lines)
+
     def write_status_snapshot(self, iteration: int, data: bytes) -> None:
         snapshot_path = self.status_snapshot_path(iteration)
         snapshot_path.parent.mkdir(parents=True, exist_ok=True)

tests/ert/ui_tests/cli/test_cli.py

@@ -30,6 +30,7 @@
     ES_MDA_MODE,
     TEST_RUN_MODE,
 )
+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
@@ -511,6 +512,49 @@ def test_that_stop_on_fail_workflow_jobs_stop_ert(
             run_cli(TEST_RUN_MODE, "--disable-monitoring", "poly.ert")
 
 
+@pytest.mark.usefixtures("copy_poly_case")
+def test_that_workflow_output_is_written_to_experiment_in_storage():
+    Path("print_job").write_text("EXECUTABLE print_script.sh\n", encoding="utf-8")
+    Path("print_script.sh").write_text(
+        dedent(
+            """\
+                #!/bin/bash
+                echo hello from the workflow
+                echo problem from the workflow >&2
+            """
+        ),
+        encoding="utf-8",
+    )
+    Path("print_script.sh").chmod(os.stat("print_script.sh").st_mode | 0o111)
+    Path("print_workflow").write_text("printjob\n", encoding="utf-8")
+
+    with Path("poly.ert").open(mode="a", encoding="utf-8") as fh:
+        fh.write(
+            dedent(
+                """
+                   LOAD_WORKFLOW_JOB print_job printjob
+                   LOAD_WORKFLOW print_workflow wfprint
+                   HOOK_WORKFLOW wfprint PRE_SIMULATION
+                """
+            )
+        )
+
+    run_cli(TEST_RUN_MODE, "--disable-monitoring", "poly.ert")
+
+    with open_storage("storage", "r") as storage:
+        (experiment,) = storage.experiments
+        (line,) = experiment.workflow_events_path.read_text(
+            encoding="utf-8"
+        ).splitlines()
+    event = WorkflowEvent.model_validate_json(line)
+    assert event.hook == "PRE_SIMULATION"
+    assert event.workflow_name == "wfprint"
+    assert event.job_name == "printjob"
+    assert event.job_index == 0
+    assert event.stdout == "hello from the workflow\n"
+    assert event.stderr == "problem from the workflow\n"
+
+
 @pytest.fixture(name="mock_cli_run")
 def fixture_mock_cli_run(monkeypatch):
     end_event = Mock()