diff --git a/docs/ert/reference/workflows/complete_workflows.rst b/docs/ert/reference/workflows/complete_workflows.rst
index 0f3e2583166..c266fc6b113 100644
--- a/docs/ert/reference/workflows/complete_workflows.rst
+++ b/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:
 
diff --git a/src/ert/config/_capture_output.py b/src/ert/config/_capture_output.py
new file mode 100644
index 00000000000..5542578c76b
--- /dev/null
+++ b/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)
diff --git a/src/ert/config/ert_config.py b/src/ert/config/ert_config.py
index 0c96cc015cf..129075104ac 100644
--- a/src/ert/config/ert_config.py
+++ b/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:
diff --git a/src/ert/config/ert_script.py b/src/ert/config/ert_script.py
index 9e9ea921f2a..2e344eaba1d 100644
--- a/src/ert/config/ert_script.py
+++ b/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
diff --git a/src/ert/config/external_ert_script.py b/src/ert/config/external_ert_script.py
index 238acf2d73f..030a7fbc3a1 100644
--- a/src/ert/config/external_ert_script.py
+++ b/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()
diff --git a/src/ert/config/workflow.py b/src/ert/config/workflow.py
index 485ba4da8d5..93b676f608f 100644
--- a/src/ert/config/workflow.py
+++ b/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,6 +110,7 @@ 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,
@@ -107,7 +118,11 @@ def from_file(
             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)],
         )
 
diff --git a/src/ert/run_models/run_model.py b/src/ert/run_models/run_model.py
index 1e3e3627438..510bd40973d 100644
--- a/src/ert/run_models/run_model.py
+++ b/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:
diff --git a/src/ert/workflow_runner.py b/src/ert/workflow_runner.py
index af2ce5ffaef..d8e6961cf13 100644
--- a/src/ert/workflow_runner.py
+++ b/src/ert/workflow_runner.py
@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+import datetime
 import logging
 import types
 from concurrent import futures
 from concurrent.futures import Future
+from dataclasses import dataclass, field
+from enum import StrEnum
 from typing import Any, Self
 
 from ert import ErtScript
@@ -17,6 +20,25 @@
 )
 
 
+class WorkflowJobStatus(StrEnum):
+    SUCCESS = "success"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+@dataclass
+class WorkflowJobResult:
+    name: str
+    index: int
+    arguments: list[str]
+    stdout: str
+    stderr: str
+    status: WorkflowJobStatus
+    timestamp: datetime.datetime = field(
+        default_factory=lambda: datetime.datetime.now(tz=datetime.UTC)
+    )
+
+
 class WorkflowJobRunner:
     def __init__(self, workflow_job: WorkflowJob) -> None:
         self.job = workflow_job
@@ -33,36 +55,36 @@ def run(
             arguments = []
         fixtures = {} if fixtures is None else fixtures
         self.__running = True
-        if self.job.min_args and len(arguments) < self.job.min_args:
-            raise ValueError(
-                f"The job: {self.job.name} requires at least "
-                f"{self.job.min_args} arguments, {len(arguments)} given."
-            )
-
-        if self.job.max_args and self.job.max_args < len(arguments):
-            raise ValueError(
-                f"The job: {self.job.name} can only have "
-                f"{self.job.max_args} arguments, {len(arguments)} given."
+        try:
+            if self.job.min_args and len(arguments) < self.job.min_args:
+                raise ValueError(
+                    f"The job: {self.job.name} requires at least "
+                    f"{self.job.min_args} arguments, {len(arguments)} given."
+                )
+
+            if self.job.max_args and self.job.max_args < len(arguments):
+                raise ValueError(
+                    f"The job: {self.job.name} can only have "
+                    f"{self.job.max_args} arguments, {len(arguments)} given."
+                )
+
+            if isinstance(self.job, BaseErtScriptWorkflow):
+                ert_script_class = self.job.load_ert_script_class()
+                self.__script = ert_script_class()
+                # We let stop on fail either from class or config take precedence
+                self.stop_on_fail = self.job.stop_on_fail or self.__script.stop_on_fail
+
+            else:
+                self.__script = ExternalErtScript(
+                    self.job.executable,  # type: ignore
+                )
+                self.stop_on_fail = self.job.stop_on_fail
+
+            return self.__script.initializeAndRun(
+                self.job.argument_types(), arguments, fixtures
             )
-
-        if isinstance(self.job, BaseErtScriptWorkflow):
-            ert_script_class = self.job.load_ert_script_class()
-            self.__script = ert_script_class()
-            # We let stop on fail either from class or config take precedence
-            self.stop_on_fail = self.job.stop_on_fail or self.__script.stop_on_fail
-
-        else:
-            self.__script = ExternalErtScript(
-                self.job.executable,  # type: ignore
-            )
-            self.stop_on_fail = self.job.stop_on_fail
-
-        result = self.__script.initializeAndRun(
-            self.job.argument_types(), arguments, fixtures
-        )
-        self.__running = False
-
-        return result
+        finally:
+            self.__running = False
 
     @property
     def name(self) -> str:
@@ -107,9 +129,11 @@ def __init__(
         self,
         workflow: Workflow,
         fixtures: WorkflowFixtures,
+        hook: str | None = None,
     ) -> None:
         self.__workflow = workflow
         self.fixtures = fixtures
+        self._hook = hook
 
         self.__workflow_result: bool | None = None
         self._workflow_executor = futures.ThreadPoolExecutor(max_workers=1)
@@ -119,6 +143,7 @@ def __init__(
         self.__cancelled = False
         self.__current_job: WorkflowJobRunner | None = None
         self.__status: dict[str, dict[str, Any]] = {}
+        self.__job_results: list[WorkflowJobResult] = []
 
     def __enter__(self) -> Self:
         self.run()
@@ -144,48 +169,102 @@ def run_blocking(self) -> None:
 
         # Reset status
         self.__status = {}
+        self.__job_results = []
         self.__running = True
 
-        for job, args in self.__workflow:
+        for index, (job, args) in enumerate(self.__workflow):
+            if self.__cancelled:
+                # The workflow was cancelled before this job started
+                result = WorkflowJobResult(
+                    name=job.name,
+                    index=index,
+                    arguments=[str(arg) for arg in args],
+                    stdout="",
+                    stderr="",
+                    status=WorkflowJobStatus.CANCELLED,
+                )
+                self.__job_results.append(result)
+                logger.info(self._log_entry(result), extra=self._log_extra(result))
+                continue
+
             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(),
-                }
-
-                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,
-                    )
+            logger.info(
+                f"Workflow job starting; {self._job_description(jobrunner.name, index)}"
+            )
+            jobrunner.run(args, fixtures=self.fixtures)
+
+            if self.__cancelled:
+                status = WorkflowJobStatus.CANCELLED
+            elif jobrunner.hasFailed():
+                status = WorkflowJobStatus.FAILED
+            else:
+                status = WorkflowJobStatus.SUCCESS
+
+            self.__status[jobrunner.name] = {
+                "stdout": jobrunner.stdoutdata(),
+                "stderr": jobrunner.stderrdata(),
+                "completed": status is WorkflowJobStatus.SUCCESS,
+            }
+            result = WorkflowJobResult(
+                name=jobrunner.name,
+                index=index,
+                arguments=[str(arg) for arg in args],
+                stdout=jobrunner.stdoutdata(),
+                stderr=jobrunner.stderrdata(),
+                status=status,
+            )
+            self.__job_results.append(result)
+
+            extra = self._log_extra(result, execution_type=jobrunner.execution_type)
+            if status is WorkflowJobStatus.FAILED:
+                logger.error(self._log_entry(result), extra=extra)
+            else:
+                logger.info(self._log_entry(result), extra=extra)
+
+            if jobrunner.hasFailed() and jobrunner.stop_on_fail:
+                self.__running = False
+                raise RuntimeError(
+                    f"Workflow job {result.name} failed with error: {result.stderr}"
+                )
 
         self.__current_job = None
         self.__running = False
         self.__workflow_result = True
 
+    def _job_description(self, job_name: str, index: int) -> str:
+        """Identify a job invocation the same way in every workflow log line."""
+        return (
+            f"hook={self._hook} workflow={self.__workflow.name} job={job_name}#{index}"
+        )
+
+    def _log_entry(self, result: WorkflowJobResult) -> str:
+        description = self._job_description(result.name, result.index)
+        sections = [f"Workflow job result; {description} status={result.status}"]
+        if result.arguments:
+            sections.append(f"--- arguments ---\n{' '.join(result.arguments)}")
+        if result.stdout:
+            sections.append(f"--- stdout ---\n{result.stdout.rstrip('\n')}")
+        if result.stderr:
+            sections.append(f"--- stderr ---\n{result.stderr.rstrip('\n')}")
+        return "\n".join(sections)
+
+    def _log_extra(
+        self, result: WorkflowJobResult, execution_type: str | None = None
+    ) -> dict[str, Any]:
+        extra: dict[str, Any] = {
+            "class": "WORKFLOW_JOB",
+            "job_name": result.name,
+            "workflow_name": self.__workflow.name,
+            "arguments": " ".join(result.arguments),
+            "status": result.status,
+        }
+        if self._hook is not None:
+            extra["hook"] = self._hook
+        if execution_type is not None:
+            extra["execution_type"] = execution_type
+        return extra
+
     def isRunning(self) -> bool:
         if self.__running:
             return True
@@ -223,3 +302,11 @@ def workflowResult(self) -> bool | None:
 
     def workflowReport(self) -> dict[str, dict[str, Any]]:
         return self.__status
+
+    def workflow_job_results(self) -> list[WorkflowJobResult]:
+        """One entry per job invocation, in the order the jobs were run.
+
+        Unlike workflowReport(), which is keyed by job name, this keeps the
+        output of every invocation when the same job is run more than once.
+        """
+        return self.__job_results
diff --git a/tests/ert/ui_tests/gui/test_workflow_tool.py b/tests/ert/ui_tests/gui/test_workflow_tool.py
index 35afc3c011c..f2ec75d4142 100644
--- a/tests/ert/ui_tests/gui/test_workflow_tool.py
+++ b/tests/ert/ui_tests/gui/test_workflow_tool.py
@@ -1,3 +1,4 @@
+import logging
 from collections.abc import Generator
 from contextlib import contextmanager
 from pathlib import Path
@@ -79,7 +80,9 @@ def close_all():
     assert Path(".ert_runpath_list").is_file()
 
 
-def test_run_workflow_with_no_ensemble_selected(qtbot, tmp_path, capsys, monkeypatch):
+def test_run_workflow_with_no_ensemble_selected(
+    qtbot, tmp_path, capsys, caplog, monkeypatch
+):
     monkeypatch.chdir(tmp_path)
     (tmp_path / "config.ert").write_text(
         dedent("""
@@ -121,6 +124,9 @@ def close_all():
         qtbot.mouseClick(workflow_widget.run_button, Qt.MouseButton.LeftButton)
 
     QTimer.singleShot(1000, handle_run_workflow_tool)
-    gui.workflows_tool.trigger()
+    with caplog.at_level(logging.INFO, logger="ert.workflow_runner"):
+        gui.workflows_tool.trigger()
     assert capsys.readouterr().out == "Hello world\n"
+    assert "workflow=print_workflow job=PRINT#0 status=success" in caplog.text
+    assert "--- stdout ---\nHello world" in caplog.text
     gui.close()
diff --git a/tests/ert/unit_tests/cli/test_cli_workflow.py b/tests/ert/unit_tests/cli/test_cli_workflow.py
index 3c2b96d60c4..b2ef8fa1345 100644
--- a/tests/ert/unit_tests/cli/test_cli_workflow.py
+++ b/tests/ert/unit_tests/cli/test_cli_workflow.py
@@ -1,3 +1,4 @@
+import logging
 from argparse import Namespace
 from pathlib import Path
 
@@ -20,3 +21,29 @@ def test_executing_workflow(storage):
     args = Namespace(name="test_wf")
     execute_workflow(rc, storage, args.name)
     assert Path("test_workflow_output.csv").is_file()
+
+
+@pytest.mark.usefixtures("copy_poly_case")
+def test_that_output_of_workflow_run_from_cli_is_in_ert_log(storage, caplog):
+    Path("print_job").write_text("EXECUTABLE print_script.sh\n", encoding="utf-8")
+    print_script = Path("print_script.sh")
+    print_script.write_text(
+        "#!/bin/bash\necho hello from the cli workflow\n", encoding="utf-8"
+    )
+    print_script.chmod(print_script.stat().st_mode | 0o111)
+    Path("print_workflow").write_text("printjob\n", encoding="utf-8")
+
+    config_file = "poly.ert"
+    with Path(config_file).open("a", encoding="utf-8") as file_handle:
+        file_handle.write(
+            "LOAD_WORKFLOW_JOB print_job printjob\n"
+            "LOAD_WORKFLOW print_workflow wfprint\n"
+        )
+
+    rc = ErtConfig.with_plugins(get_site_plugins()).from_file(config_file)
+
+    with caplog.at_level(logging.INFO, logger="ert.workflow_runner"):
+        execute_workflow(rc, storage, "wfprint")
+
+    assert "workflow=wfprint job=printjob#0 status=success" in caplog.text
+    assert "--- stdout ---\nhello from the cli workflow" in caplog.text
diff --git a/tests/ert/unit_tests/config/test_ert_config.py b/tests/ert/unit_tests/config/test_ert_config.py
index b1956d7f7a4..e1b1c05629e 100644
--- a/tests/ert/unit_tests/config/test_ert_config.py
+++ b/tests/ert/unit_tests/config/test_ert_config.py
@@ -3227,3 +3227,43 @@ def test_that_log_shape_registry_logs_count_of_shapes(caplog):
         shape_registry.register(CircleShapeConfig(north=i, east=i, radius=i))
     log_shape_registry(shape_registry)
     assert "Count of shapes in ShapeRegistry: {'CircleShapeConfig': 10}" in caplog.text
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_workflow_takes_name_given_to_load_workflow():
+    Path("WFJOB").write_text("EXECUTABLE echo\n", encoding="utf-8")
+    Path("wf_file").write_text("WFJOB hello\n", encoding="utf-8")
+    Path("test.ert").write_text(
+        dedent(
+            """
+            NUM_REALIZATIONS 1
+            LOAD_WORKFLOW_JOB WFJOB
+            LOAD_WORKFLOW wf_file my_workflow_name
+            """
+        ),
+        encoding="utf-8",
+    )
+
+    ert_config = ErtConfig.from_file("test.ert")
+
+    assert ert_config.workflows["my_workflow_name"].name == "my_workflow_name"
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_workflow_without_explicit_name_is_named_after_its_file():
+    Path("WFJOB").write_text("EXECUTABLE echo\n", encoding="utf-8")
+    Path("wf_file").write_text("WFJOB hello\n", encoding="utf-8")
+    Path("test.ert").write_text(
+        dedent(
+            """
+            NUM_REALIZATIONS 1
+            LOAD_WORKFLOW_JOB WFJOB
+            LOAD_WORKFLOW wf_file
+            """
+        ),
+        encoding="utf-8",
+    )
+
+    ert_config = ErtConfig.from_file("test.ert")
+
+    assert ert_config.workflows["wf_file"].name == "wf_file"
diff --git a/tests/ert/unit_tests/workflow_runner/test_ert_script.py b/tests/ert/unit_tests/workflow_runner/test_ert_script.py
index c9596afdd9e..83fd551a244 100644
--- a/tests/ert/unit_tests/workflow_runner/test_ert_script.py
+++ b/tests/ert/unit_tests/workflow_runner/test_ert_script.py
@@ -1,4 +1,5 @@
 import sys
+import threading
 from pathlib import Path
 
 import pytest
@@ -80,3 +81,163 @@ def run(self, *arg):
     failing = FailingScript()
     failing.initializeAndRun([], [])
     assert failing.hasFailed()
+
+
+def test_that_stdout_and_stderr_printed_by_ert_script_are_captured():
+    class PrintingScript(ErtScript):
+        def run(self):
+            print("to stdout")
+            print("to stderr", file=sys.stderr)
+
+    script = PrintingScript()
+    script.initializeAndRun([], [])
+
+    assert script.stdoutdata == "to stdout\n"
+    assert script.stderrdata == "to stderr\n"
+
+
+def test_that_output_printed_before_ert_script_raises_is_captured():
+    class PrintingAndFailingScript(ErtScript):
+        def run(self):
+            print("printed before failing")
+            raise ValueError("boom")
+
+    script = PrintingAndFailingScript()
+    script.initializeAndRun([], [])
+
+    assert script.hasFailed()
+    assert script.stdoutdata == "printed before failing\n"
+
+
+def test_that_stack_trace_of_failing_script_is_appended_to_captured_stderr():
+    class PrintingAndFailingScript(ErtScript):
+        def run(self):
+            print("printed to stderr", file=sys.stderr)
+            raise ValueError("boom")
+
+    script = PrintingAndFailingScript()
+    script.initializeAndRun([], [])
+
+    assert script.stderrdata.startswith("printed to stderr\n")
+    assert "ValueError: boom" in script.stderrdata
+
+
+def test_that_stderr_without_trailing_newline_is_separated_from_stack_trace():
+    class PrintingAndFailingScript(ErtScript):
+        def run(self):
+            print("partial", end="", file=sys.stderr)
+            raise ValueError("boom")
+
+    script = PrintingAndFailingScript()
+    script.initializeAndRun([], [])
+
+    assert script.stderrdata.startswith("partial\nboom\n")
+
+
+def test_that_output_captured_from_ert_script_is_still_written_to_stdout(capsys):
+    class PrintingScript(ErtScript):
+        def run(self):
+            print("to stdout")
+            print("to stderr", file=sys.stderr)
+
+    PrintingScript().initializeAndRun([], [])
+
+    captured = capsys.readouterr()
+    assert captured.out == "to stdout\n"
+    assert captured.err == "to stderr\n"
+
+
+def _join(thread: threading.Thread) -> None:
+    thread.join(timeout=10)
+    assert not thread.is_alive(), (
+        f"{thread.name} did not finish; it would leave sys.stdout captured"
+    )
+
+
+def test_that_output_written_by_another_thread_is_left_out_of_capture():
+    job_may_finish = threading.Event()
+    other_thread_has_printed = threading.Event()
+
+    class SlowScript(ErtScript):
+        def run(self):
+            print("from the job")
+            other_thread_has_printed.wait(timeout=10)
+            job_may_finish.wait(timeout=10)
+
+    def print_from_another_thread():
+        print("from an unrelated thread")
+        other_thread_has_printed.set()
+
+    script = SlowScript()
+    job = threading.Thread(target=script.initializeAndRun, args=([], []))
+    job.start()
+    other_thread = threading.Thread(target=print_from_another_thread)
+    other_thread.start()
+    _join(other_thread)
+    assert other_thread_has_printed.is_set(), "the unrelated thread never printed"
+    job_may_finish.set()
+    _join(job)
+
+    assert "from the job" in script.stdoutdata
+    assert "from an unrelated thread" not in script.stdoutdata
+
+
+def test_that_concurrent_scripts_only_capture_their_own_output():
+    both_are_running = threading.Barrier(2, timeout=10)
+
+    class PrintingScript(ErtScript):
+        def run(self, message):
+            both_are_running.wait()
+            print(message)
+            both_are_running.wait()
+
+    first, second = PrintingScript(), PrintingScript()
+    threads = [
+        threading.Thread(target=script.initializeAndRun, args=([str], [message]))
+        for script, message in ((first, "first"), (second, "second"))
+    ]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        _join(thread)
+
+    assert first.stdoutdata.strip() == "first"
+    assert second.stdoutdata.strip() == "second"
+
+
+def test_that_original_streams_are_restored_after_capturing():
+    class PrintingScript(ErtScript):
+        def run(self):
+            print("hello")
+
+    stdout, stderr = sys.stdout, sys.stderr
+    PrintingScript().initializeAndRun([], [])
+
+    assert sys.stdout is stdout
+    assert sys.stderr is stderr
+
+
+@pytest.mark.parametrize(
+    "attribute", ["fileno", "isatty", "encoding", "errors", "buffer", "line_buffering"]
+)
+def test_that_captured_stdout_exposes_same_attributes_as_real_one(
+    attribute,
+):
+    def look_up(stream):
+        """The attribute value, or the error raised, so both can be compared."""
+        try:
+            value = getattr(stream, attribute)
+            return value() if callable(value) else value
+        except Exception as e:
+            return type(e)
+
+    seen = {}
+
+    class InspectingScript(ErtScript):
+        def run(self):
+            seen["value"] = look_up(sys.stdout)
+
+    expected = look_up(sys.stdout)
+    InspectingScript().initializeAndRun([], [])
+
+    assert seen["value"] == expected
diff --git a/tests/ert/unit_tests/workflow_runner/test_workflow_runner.py b/tests/ert/unit_tests/workflow_runner/test_workflow_runner.py
index 6cb5430917c..d18c9a04e40 100644
--- a/tests/ert/unit_tests/workflow_runner/test_workflow_runner.py
+++ b/tests/ert/unit_tests/workflow_runner/test_workflow_runner.py
@@ -1,3 +1,4 @@
+import logging
 from pathlib import Path
 from textwrap import dedent
 from unittest.mock import patch
@@ -10,7 +11,7 @@
     UserInstalledErtScriptWorkflow,
     workflow_job_from_file,
 )
-from ert.workflow_runner import WorkflowJobRunner, WorkflowRunner
+from ert.workflow_runner import WorkflowJobRunner, WorkflowJobStatus, WorkflowRunner
 from tests.ert.utils import wait_until
 
 from .workflow_common import WorkflowCommon
@@ -81,6 +82,37 @@ def test_error_handling_external_job():
     assert runner.stderrdata().startswith("Traceback")
 
 
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_stdout_printed_before_external_job_fails_is_captured():
+    WorkflowCommon.createExternalDumpJob()
+
+    job = workflow_job_from_file(
+        name="DUMP", config_file="dump_failing_job", origin="user"
+    )
+
+    runner = WorkflowJobRunner(job)
+    runner.run([])
+
+    assert runner.stdoutdata() == "Hello Failing\n"
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_failing_external_job_reports_its_exit_code_without_ert_stack_trace():
+    WorkflowCommon.createExternalDumpJob()
+
+    job = workflow_job_from_file(
+        name="DUMP", config_file="dump_failing_job", origin="user"
+    )
+
+    runner = WorkflowJobRunner(job)
+    runner.run([])
+
+    stderr = runner.stderrdata()
+    assert stderr.endswith("dump_failing.py failed with exit code 1")
+    assert "ert_script.py" not in stderr
+    assert "external_ert_script.py" not in stderr
+
+
 @pytest.mark.usefixtures("use_tmpdir")
 @pytest.mark.filterwarnings("ignore:.*Deprecated keywords, SCRIPT and INTERNAL")
 def test_run_internal_script():
@@ -178,6 +210,121 @@ def test_workflow_run():
     assert Path("dump2").read_text(encoding="utf-8") == "dump_text_2"
 
 
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_job_results_contain_one_entry_per_job_invocation():
+    WorkflowCommon.createExternalDumpJob()
+
+    dump_job = workflow_job_from_file("dump_job", name="DUMP", origin="user")
+    workflow = Workflow.from_file(
+        "dump_workflow", {"<PARAM>": "text"}, {"DUMP": dump_job}
+    )
+
+    runner = WorkflowRunner(workflow, fixtures={})
+    runner.run_blocking()
+
+    results = runner.workflow_job_results()
+    assert [(result.name, result.index, result.arguments) for result in results] == [
+        ("DUMP", 0, ["dump1", "dump_text_1"]),
+        ("DUMP", 1, ["dump2", "dump_text_2"]),
+    ]
+    assert [result.stdout for result in results] == ["Hello World\n", "Hello World\n"]
+    assert all(result.status is WorkflowJobStatus.SUCCESS for result in results)
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_output_of_workflow_job_is_written_to_ert_log(caplog):
+    WorkflowCommon.createExternalDumpJob()
+
+    dump_job = workflow_job_from_file("dump_job", name="DUMP", origin="user")
+    workflow = Workflow.from_file(
+        "dump_workflow", {"<PARAM>": "text"}, {"DUMP": dump_job}
+    )
+
+    with caplog.at_level(logging.INFO, logger="ert.workflow_runner"):
+        WorkflowRunner(workflow, fixtures={}).run_blocking()
+
+    assert "workflow=dump_workflow job=DUMP#0 status=success" in caplog.text
+    assert "--- arguments ---\ndump1 dump_text_1" in caplog.text
+    assert "--- stdout ---\nHello World" in caplog.text
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_hook_workflow_was_run_from_is_named_in_ert_log(caplog):
+    WorkflowCommon.createExternalDumpJob()
+
+    dump_job = workflow_job_from_file("dump_job", name="DUMP", origin="user")
+    workflow = Workflow.from_file(
+        "dump_workflow", {"<PARAM>": "text"}, {"DUMP": dump_job}
+    )
+
+    with caplog.at_level(logging.INFO, logger="ert.workflow_runner"):
+        WorkflowRunner(workflow, fixtures={}, hook="POST_EXPERIMENT").run_blocking()
+
+    assert (
+        "Workflow job result; hook=POST_EXPERIMENT workflow=dump_workflow"
+        in caplog.text
+    )
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_workflow_run_outside_hook_is_logged_with_hook_none(caplog):
+    WorkflowCommon.createExternalDumpJob()
+
+    dump_job = workflow_job_from_file("dump_job", name="DUMP", origin="user")
+    workflow = Workflow.from_file(
+        "dump_workflow", {"<PARAM>": "text"}, {"DUMP": dump_job}
+    )
+
+    with caplog.at_level(logging.INFO, logger="ert.workflow_runner"):
+        WorkflowRunner(workflow, fixtures={}).run_blocking()
+
+    assert "Workflow job result; hook=None workflow=dump_workflow" in caplog.text
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_job_is_logged_as_starting_before_it_is_logged_as_finished(caplog):
+    WorkflowCommon.createExternalDumpJob()
+
+    dump_job = workflow_job_from_file("dump_job", name="DUMP", origin="user")
+    workflow = Workflow.from_file(
+        "dump_workflow", {"<PARAM>": "text"}, {"DUMP": dump_job}
+    )
+
+    with caplog.at_level(logging.INFO, logger="ert.workflow_runner"):
+        WorkflowRunner(workflow, fixtures={}, hook="PRE_SIMULATION").run_blocking()
+
+    started = (
+        "Workflow job starting; hook=PRE_SIMULATION workflow=dump_workflow job=DUMP#0"
+    )
+    finished = (
+        "Workflow job result; hook=PRE_SIMULATION"
+        " workflow=dump_workflow job=DUMP#0 status=success"
+    )
+    assert started in caplog.text
+    assert finished in caplog.text
+    assert caplog.text.index(started) < caplog.text.index(finished)
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_output_of_job_that_stops_workflow_is_still_logged(caplog):
+    WorkflowCommon.createExternalDumpJob()
+    with Path("dump_failing_job").open("a", encoding="utf-8") as f:
+        f.write("STOP_ON_FAIL True")
+    Path("dump_failing_workflow").write_text("DUMP", encoding="utf-8")
+
+    dump_job = workflow_job_from_file("dump_failing_job", name="DUMP", origin="user")
+    workflow = Workflow.from_file("dump_failing_workflow", {}, {"DUMP": dump_job})
+
+    with (
+        caplog.at_level(logging.INFO, logger="ert.workflow_runner"),
+        pytest.raises(RuntimeError, match="failed with error"),
+    ):
+        WorkflowRunner(workflow, fixtures={}).run_blocking()
+
+    assert "status=failed" in caplog.text
+    assert "--- stdout ---\nHello Failing" in caplog.text
+
+
 @pytest.mark.slow
 @pytest.mark.usefixtures("use_tmpdir")
 @pytest.mark.filterwarnings("ignore:.*Deprecated keywords, SCRIPT and INTERNAL")
@@ -215,6 +362,17 @@ 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.workflow_job_results()
+    }
+    assert results[0].status is WorkflowJobStatus.SUCCESS
+    # The job that was interrupted by cancellation is reported as cancelled,
+    # not as failed.
+    assert results[1].status is WorkflowJobStatus.CANCELLED
+    # The remaining job never got a chance to start, so it is reported as
+    # cancelled rather than silently omitted.
+    assert results[2].status is WorkflowJobStatus.CANCELLED
+
 
 @pytest.mark.slow
 @pytest.mark.usefixtures("use_tmpdir")
@@ -344,3 +502,18 @@ def test_workflow_stops_with_stopping_job():
 
     # Expect no error raised
     WorkflowRunner(workflow, fixtures={}).run_blocking()
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+@pytest.mark.filterwarnings("ignore:.*Deprecated keywords, SCRIPT and INTERNAL")
+def test_that_job_runner_stops_reporting_it_is_running_when_arguments_are_rejected():
+    WorkflowCommon.createErtScriptsJob()
+    job = workflow_job_from_file(
+        name="SUBTRACT", config_file="subtract_script_job", origin="user"
+    )
+    runner = WorkflowJobRunner(job)
+
+    with pytest.raises(ValueError, match="requires at least 2 arguments"):
+        runner.run([1])
+
+    assert not runner.isRunning()