Log workflow output

docs/ert/reference/workflows/complete_workflows.rst

@@ -88,8 +88,28 @@ Observe that the workflows being 'hooked in' with the
 :code:`HOOK_WORKFLOW` must be loaded with the :code:`LOAD_WORKFLOW`
 keyword.
 
-Workflow logs with output from job execution can be found in the terminal where ERT was
-started.
+Workflow output
+---------------
+
+Output from workflow jobs is written to the ERT log.
+Every job invocation gets an output that the job wrote to stdout and stderr::
+
+    2026-07-30 10:19:33,041 - ert.workflow_runner - MainThread - INFO - Workflow job starting; hook=PRE_SIMULATION workflow=my_workflow job=MY_JOB#0
+    2026-07-30 10:19:33,052 - ert.workflow_runner - MainThread - INFO - Workflow job result; hook=PRE_SIMULATION workflow=my_workflow job=MY_JOB#0 status=success
+    --- arguments ---
+    first_argument second_argument
+    --- stdout ---
+    Hello from the workflow
+
+This covers every way a workflow can be started: hooked in with
+:code:`HOOK_WORKFLOW`, run with :code:`ert workflow`, or started from the *Run
+workflow* tool in the GUI. Workflows that were not started from a hook, such as
+a manual run, are logged with :code:`hook=None`.
+
+The :code:`status` field is one of :code:`success`, :code:`failed` or
+:code:`cancelled`. Only :code:`failed` is logged at :code:`ERROR` level;
+jobs that were stopped because the workflow was cancelled are logged at
+:code:`INFO` level.
 
 .. _runpath-file-workflows:
 

src/ert/config/_capture_output.py

@@ -0,0 +1,137 @@
+"""Capturing what workflow jobs write to ``sys.stdout``/``sys.stderr``.
+
+:func:`contextlib.redirect_stdout` is deliberately not used here. It swaps
+``sys.stdout`` process-wide and hands the stream to a single writer, whereas
+capturing workflow job output has to
+
+* pass writes on to the stream it replaced, so output still reaches the
+  terminal as the job runs rather than only appearing once it is over,
+* keep each thread's writes apart, since workflow runners each run their jobs
+  on a thread of their own and a job must not pick up output another thread
+  happened to write, and
+* survive overlapping captures, restoring the original stream only once the
+  last capture is done and only if nothing else has replaced it since.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import sys
+import threading
+from collections.abc import Iterator
+from typing import Any, TextIO, override
+
+
+class _CaptureProxy(io.TextIOBase):
+    """Stands in for ``sys.stdout``/``sys.stderr`` while workflow jobs run.
+
+    Internal jobs print straight to ``sys.stdout``/``sys.stderr``, so replacing
+    those streams is the only way to get hold of what they write. Everything
+    written is passed on to the stream this proxy replaced, so output still
+    reaches the terminal.
+
+    Capture is per-thread: a thread collects only what it writes itself, so
+    workflow jobs running concurrently do not pick up each other's output.
+    Threads that are not capturing are unaffected beyond the forwarding.
+    """
+
+    def __init__(self, stream: TextIO) -> None:
+        super().__init__()
+        self.wrapped_stream = stream
+        self._local = threading.local()
+        self._active_captures = 0
+
+    @property
+    def _buffers(self) -> list[io.StringIO]:
+        buffers: list[io.StringIO] | None = getattr(self._local, "buffers", None)
+        if buffers is None:
+            buffers = []
+            self._local.buffers = buffers
+        return buffers
+
+    @contextlib.contextmanager
+    def capture(self) -> Iterator[io.StringIO]:
+        """Record what the calling thread writes for the duration of the block."""
+        buffer = io.StringIO()
+        buffers = self._buffers
+        buffers.append(buffer)
+        try:
+            yield buffer
+        finally:
+            buffers.remove(buffer)
+
+    @override
+    def write(self, s: str, /) -> int:
+        for buffer in self._buffers:
+            buffer.write(s)
+        return self.wrapped_stream.write(s)
+
+    @override
+    def flush(self) -> None:
+        self.wrapped_stream.flush()
+
+    @override
+    def close(self) -> None:
+        """Closing is ignored, as the wrapped stream outlives the capture."""
+
+    @override
+    def fileno(self) -> int:
+        return self.wrapped_stream.fileno()
+
+    @override
+    def isatty(self) -> bool:
+        return self.wrapped_stream.isatty()
+
+    @override
+    def writable(self) -> bool:
+        # IOBase defaults to False, so this one is load-bearing. readable() and
+        # seekable() are left to IOBase, which already reports False.
+        return True
+
+    @property
+    @override
+    def encoding(self) -> str:  # type: ignore[override]
+        # TextIOBase defines encoding, errors and newlines as descriptors
+        # returning None, so __getattr__ is never consulted for them.
+        return getattr(self.wrapped_stream, "encoding", "utf-8")
+
+    @property
+    @override
+    def errors(self) -> str | None:  # type: ignore[override]
+        return getattr(self.wrapped_stream, "errors", None)
+
+    @property
+    @override
+    def newlines(self) -> Any:  # type: ignore[override]
+        return getattr(self.wrapped_stream, "newlines", None)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self.__dict__["wrapped_stream"], name)
+
+
+_capture_lock = threading.Lock()
+
+
+@contextlib.contextmanager
+def capturing(stream_name: str) -> Iterator[io.StringIO]:
+    # Record what the calling thread writes to ``sys.<stream_name>``
+    proxy: _CaptureProxy
+    with _capture_lock:
+        stream = getattr(sys, stream_name)
+        proxy = stream if isinstance(stream, _CaptureProxy) else _CaptureProxy(stream)
+        proxy._active_captures += 1
+        setattr(sys, stream_name, proxy)
+    try:
+        with proxy.capture() as buffer:
+            yield buffer
+    finally:
+        with _capture_lock:
+            # Captures running at the same time share one proxy, so only the
+            # last one to finish puts the original stream back. The identity
+            # check keeps us from doing so if something else has replaced
+            # sys.<stream_name> in the meantime, as restoring would then throw
+            # away their stream rather than ours.
+            proxy._active_captures -= 1
+            if proxy._active_captures == 0 and getattr(sys, stream_name) is proxy:
+                setattr(sys, stream_name, proxy.wrapped_stream)

src/ert/config/ert_config.py

@@ -487,6 +487,7 @@ def create_and_hook_workflows(
                 work[0],
                 substitutions,
                 workflow_jobs,
+                name=filename,
             )
             workflows[filename] = workflow
             if existed:

src/ert/config/ert_script.py

@@ -9,6 +9,7 @@
 from types import MappingProxyType, ModuleType
 from typing import Any
 
+from ._capture_output import capturing
 from .workflow_fixtures import (
     WorkflowFixtures,
     all_hooked_workflow_fixtures,
@@ -17,6 +18,14 @@
 logger = logging.getLogger(__name__)
 
 
+class ExternalScriptError(RuntimeError):
+    """Raised when an external workflow job exits with a non-zero exit code.
+
+    Reported without a stack trace, since it would only
+    show ert internals and could be confusing
+    """
+
+
 class ErtScript:
     """
     ErtScript is the abstract baseclass for workflow jobs and
@@ -113,7 +122,7 @@ def initializeAndRun(
                         f"Mixture of fixtures and positional arguments, err: {e}"
                     )
 
-            return self.run(*arguments)
+            return self._run_capturing_output(arguments)
         except AttributeError as e:
             error_msg = str(e)
             if not hasattr(self, "run"):
@@ -139,6 +148,10 @@ def initializeAndRun(
                 f"User warning in workflow script {self.__class__.__name__}: {uw}"
             )
             return uw.args[0]
+        except ExternalScriptError as e:
+            self.output_stack_trace(error=str(e))
+            logger.error(f"Workflow job failed: {e!s}")
+            return None
         except BaseException as e:
             full_trace = "".join(traceback.format_exception(*sys.exc_info()))
             self.output_stack_trace(f"{e!s}\n{full_trace}")
@@ -150,6 +163,14 @@ def initializeAndRun(
         finally:
             self.cleanup()
 
+    def _run_capturing_output(self, arguments: list[Any]) -> Any:
+        with capturing("stdout") as stdout, capturing("stderr") as stderr:
+            try:
+                return self.run(*arguments)
+            finally:
+                self._stdoutdata = self.stdoutdata + stdout.getvalue()
+                self._stderrdata = self.stderrdata + stderr.getvalue()
+
     # Need to have unique modules in case of identical object naming in scripts
     __module_count = 0
 
@@ -179,7 +200,10 @@ def output_stack_trace(self, error: str = "") -> None:
             f"error while running:\n{str(stack_trace).strip()}\n"
         )
 
-        self._stderrdata = error
+        existing_stderr = self.stderrdata
+        if existing_stderr and not existing_stderr.endswith("\n"):
+            existing_stderr += "\n"
+        self._stderrdata = existing_stderr + error
         self.__failed = True
 
     @staticmethod

src/ert/config/external_ert_script.py

@@ -5,7 +5,7 @@
 from subprocess import PIPE, Popen
 from typing import Any
 
-from .ert_script import ErtScript
+from .ert_script import ErtScript, ExternalScriptError
 
 
 class ExternalErtScript(ErtScript):
@@ -25,13 +25,15 @@ def run(self, *args: Any) -> None:
         # The job will complete before stdout and stderr is returned
         stdoutdata, stderrdata = self.__job.communicate()
 
-        self._stdoutdata = codecs.decode(stdoutdata, "utf8", "replace")
-        self._stderrdata = codecs.decode(stderrdata, "utf8", "replace")
-
-        sys.stdout.write(self._stdoutdata)
+        # Written to the current stdout/stderr, which ErtScript captures into
+        # self.stdoutdata/self.stderrdata while the script is running.
+        sys.stdout.write(codecs.decode(stdoutdata, "utf8", "replace"))
+        sys.stderr.write(codecs.decode(stderrdata, "utf8", "replace"))
 
         if self.__job.returncode != 0:
-            raise RuntimeError(self._stderrdata)
+            raise ExternalScriptError(
+                f"{self.__executable} failed with exit code {self.__job.returncode}"
+            )
 
     def cancel(self) -> Any:
         super().cancel()

src/ert/config/workflow.py

@@ -2,7 +2,10 @@
 
 import os
 from collections.abc import Iterator
-from typing import Any
+from pathlib import Path
+from typing import Any, Self
+
+from pydantic import model_validator
 
 from ert.base_model_context import BaseModelWithContextSupport
 
@@ -13,6 +16,7 @@
 
 class Workflow(BaseModelWithContextSupport):
     src_file: str
+    name: str = ""
     cmd_list: list[tuple[WorkflowJob, Any]]
 
     def __len__(self) -> int:
@@ -24,6 +28,12 @@ def __getitem__(self, index: int) -> tuple[WorkflowJob, Any]:
     def __iter__(self) -> Iterator[tuple[WorkflowJob, Any]]:  # type: ignore
         return iter(self.cmd_list)
 
+    @model_validator(mode="after")
+    def _default_name_to_file_name(self) -> Self:
+        if not self.name:
+            self.name = Path(self.src_file).name
+        return self
+
     @staticmethod
     def validate_workflow_job(
         job_name: str,
@@ -100,14 +110,19 @@ def from_file(
         src_file: str,
         context: dict[str, str] | None,
         job_dict: dict[str, WorkflowJob],
+        name: str | None = None,
     ) -> Workflow:
         cmd_list = cls._parse_command_list(
             src_file=src_file,
             context=list(map(list, context.items())) if context else [],
             job_dict=job_dict,
         )
 
-        return cls(src_file=src_file, cmd_list=cmd_list)
+        return cls(
+            src_file=src_file,
+            name=name or Path(src_file).name,
+            cmd_list=cmd_list,
+        )
 
     @classmethod
     def from_instructions(
@@ -120,6 +135,7 @@ def from_instructions(
         job = cls.validate_workflow_job(job_name, args, job_dict)
         return cls(
             src_file=workflow_name,
+            name=workflow_name,
             cmd_list=[(job, args)],
         )
 

src/ert/run_models/run_model.py

@@ -835,6 +835,7 @@ def run_workflows(
             workflow_runner = WorkflowRunner(
                 workflow=workflow,
                 fixtures=create_workflow_fixtures_from_hooked(fixtures),
+                hook=str(fixtures.hook),
             )
             self._workflow_runner = workflow_runner
             try: