Update task engine to increment run_count when entering running state

src/prefect/task_engine.py

@@ -692,8 +692,11 @@ async def wait_until_ready(self):
         if scheduled_time := self.state.state_details.scheduled_time:
             sleep_time = (scheduled_time - pendulum.now("utc")).total_seconds()
             await anyio.sleep(sleep_time if sleep_time > 0 else 0)
+            new_state = Retrying() if self.state.name == "AwaitingRetry" else Running()
+            if self.state.name == "AwaitingRetry":
+                self.task_run.run_count += 1
             self.set_state(
-                Retrying() if self.state.name == "AwaitingRetry" else Running(),
+                new_state,
                 force=True,
             )
 
@@ -1199,8 +1202,11 @@ async def wait_until_ready(self):
         if scheduled_time := self.state.state_details.scheduled_time:
             sleep_time = (scheduled_time - pendulum.now("utc")).total_seconds()
             await anyio.sleep(sleep_time if sleep_time > 0 else 0)
+            new_state = Retrying() if self.state.name == "AwaitingRetry" else Running()
+            if self.state.name == "AwaitingRetry":
+                self.task_run.run_count += 1
             await self.set_state(
-                Retrying() if self.state.name == "AwaitingRetry" else Running(),
+                new_state,
                 force=True,
             )
 

tests/test_tasks.py

@@ -1245,7 +1245,8 @@ async def test_task_retries_receive_latest_task_run_in_context(
         run_counts = []
         start_times = []
 
-        @task(retries=3)
+        # Added retry_delay_seconds as a regression check for https://github.com/PrefectHQ/prefect/issues/15422
+        @task(retries=3, retry_delay_seconds=1)
         def flaky_function():
             ctx = TaskRunContext.get()
             state_names.append(ctx.task_run.state_name)