diff --git a/lib/galaxy/exceptions/__init__.py b/lib/galaxy/exceptions/__init__.py
index 2448fd817960..5806ed157c21 100644
--- a/lib/galaxy/exceptions/__init__.py
+++ b/lib/galaxy/exceptions/__init__.py
@@ -151,6 +151,16 @@ class ToolInputsNotReadyException(MessageException):
     error_code = error_codes_by_name["TOOL_INPUTS_NOT_READY"]
 
 
+class ToolInputsNotOKException(MessageException):
+    def __init__(self, err_msg=None, type="info", *, src: str, id: int, **extra_error_info):
+        super().__init__(err_msg, type, **extra_error_info)
+        self.src = src
+        self.id = id
+
+    status_code = 400
+    error_code = error_codes_by_name["TOOL_INPUTS_NOT_OK"]
+
+
 class RealUserRequiredException(MessageException):
     status_code = 400
     error_code = error_codes_by_name["REAL_USER_REQUIRED"]
diff --git a/lib/galaxy/exceptions/error_codes.json b/lib/galaxy/exceptions/error_codes.json
index 7d8f521b8218..99464306acb6 100644
--- a/lib/galaxy/exceptions/error_codes.json
+++ b/lib/galaxy/exceptions/error_codes.json
@@ -94,6 +94,11 @@
         "code": 400016,
         "message": "Only real users can make this request."
     },
+    {
+        "name": "TOOL_INPUTS_NOT_OK",
+        "code": 400017,
+        "message": "Tool inputs not in required OK state."
+    },
     {
         "name": "USER_AUTHENTICATION_FAILED",
         "code": 401001,
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index 5a92fd960e4b..62bf8b5e217b 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -34,7 +34,10 @@
     exceptions,
     model,
 )
-from galaxy.exceptions import ToolInputsNotReadyException
+from galaxy.exceptions import (
+    ToolInputsNotOKException,
+    ToolInputsNotReadyException,
+)
 from galaxy.job_execution import output_collect
 from galaxy.metadata import get_metadata_compute_strategy
 from galaxy.model.base import transaction
@@ -3200,8 +3203,10 @@ def check_dataset_state(state):
 
             if self.require_dataset_ok:
                 if state != model.Dataset.states.OK:
-                    raise ValueError(
-                        f"Tool requires inputs to be in valid state, but dataset {input_dataset} is in state '{input_dataset.state}'"
+                    raise ToolInputsNotOKException(
+                        f"Tool requires inputs to be in valid state, but dataset {input_dataset} is in state '{input_dataset.state}'",
+                        src="hda",
+                        id=input_dataset.id,
                     )
 
         for input_dataset in input_datasets.values():
diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py
index 412c36bcefdd..c935aff13ecc 100644
--- a/lib/galaxy/tools/execute.py
+++ b/lib/galaxy/tools/execute.py
@@ -19,6 +19,7 @@
 from boltons.iterutils import remap
 
 from galaxy import model
+from galaxy.exceptions import ToolInputsNotOKException
 from galaxy.model.base import transaction
 from galaxy.model.dataset_collections.matching import MatchingCollections
 from galaxy.model.dataset_collections.structure import (
@@ -143,14 +144,17 @@ def execute_single_job(execution_slice, completed_job, skip=False):
     if check_inputs_ready:
         for params in execution_tracker.param_combinations:
             # This will throw an exception if the tool is not ready.
-            check_inputs_ready(
-                tool,
-                trans,
-                params,
-                history,
-                execution_cache=execution_cache,
-                collection_info=collection_info,
-            )
+            try:
+                check_inputs_ready(
+                    tool,
+                    trans,
+                    params,
+                    history,
+                    execution_cache=execution_cache,
+                    collection_info=collection_info,
+                )
+            except ToolInputsNotOKException as e:
+                execution_tracker.record_error(e)
 
     execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template)
     job_count = len(execution_tracker.param_combinations)
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index 36bdd374b172..805a8ba3f2f6 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -2314,6 +2314,16 @@ def callback(input, prefixed_name, **kwargs):
         if execution_tracker.execution_errors:
             # TODO: formalize into InvocationFailure ?
             message = f"Failed to create {len(execution_tracker.execution_errors)} job(s) for workflow step {step.order_index + 1}: {str(execution_tracker.execution_errors[0])}"
+            for error in execution_tracker.execution_errors:
+                # try first to raise a structured invocation error message
+                if isinstance(error, exceptions.ToolInputsNotOKException) and error.src == "hda":
+                    raise FailWorkflowEvaluation(
+                        why=InvocationFailureDatasetFailed(
+                            reason=FailureReason.dataset_failed,
+                            hda_id=error.id,
+                            workflow_step_id=step.id,
+                        )
+                    )
             raise exceptions.MessageException(message)
 
         return complete