From a2320deab05579a413495a786d5bf866b7b643a1 Mon Sep 17 00:00:00 2001
From: nate nowack <thrast36@gmail.com>
Date: Tue, 27 Aug 2024 17:02:31 -0500
Subject: [PATCH] ensure `on_failure` hook runs upon parameter validation error
 (#15109)

---
 src/prefect/flow_engine.py |  1 +
 tests/test_flows.py        | 14 ++++++++++++++
 2 files changed, 15 insertions(+)

diff --git a/src/prefect/flow_engine.py b/src/prefect/flow_engine.py
index d67d83e00648..4b0c0d538949 100644
--- a/src/prefect/flow_engine.py
+++ b/src/prefect/flow_engine.py
@@ -205,6 +205,7 @@ def begin_run(self) -> State:
                     result_factory=run_coro_as_sync(ResultFactory.from_flow(self.flow)),
                 )
                 self.short_circuit = True
+                self.call_hooks()
 
         new_state = Running()
         state = self.set_state(new_state)
diff --git a/tests/test_flows.py b/tests/test_flows.py
index aec1b8421ddf..8bff75e285f1 100644
--- a/tests/test_flows.py
+++ b/tests/test_flows.py
@@ -3183,6 +3183,20 @@ def my_flow():
         assert state.type == StateType.FAILED
         assert my_mock.call_args_list == [call(), call()]
 
+    def test_on_failure_hooks_run_on_bad_parameters(self):
+        my_mock = MagicMock()
+
+        def failure_hook(flow, flow_run, state):
+            my_mock("failure_hook")
+
+        @flow(on_failure=[failure_hook])
+        def my_flow(x: int):
+            pass
+
+        state = my_flow(x="x", return_state=True)
+        assert state.type == StateType.FAILED
+        assert my_mock.call_args_list == [call("failure_hook")]
+
 
 class TestFlowHooksOnCancellation:
     def test_noniterable_hook_raises(self):