Add Resuming flow runs to BypassCancellingFlowRunsWithNoInfra orc…

…hestration policy (#13299)

src/prefect/server/orchestration/core_policy.py

@@ -1018,8 +1018,10 @@ class BypassCancellingFlowRunsWithNoInfra(BaseOrchestrationRule):
  exiting the flow and tearing down infra.
 
  The `Cancelling` state is used to clean up infrastructure. If there is not infrastructure
- to clean up, we can transition directly to `Cancelled`. Runs that are `AwaitingRetry` are
- a `Scheduled` state that may have associated infrastructure.
+ to clean up, we can transition directly to `Cancelled`. Runs that are `Resuming` are in a
+ `Scheduled` state that were previously `Suspended` and do not yet have infrastructure.
+
+ Runs that are `AwaitingRetry` are a `Scheduled` state that may have associated infrastructure.
  """
 
  FROM_STATES = {StateType.SCHEDULED, StateType.PAUSED}
@@ -1034,6 +1036,7 @@ async def before_transition(
  if (
  initial_state.type == states.StateType.SCHEDULED
  and not context.run.infrastructure_pid
+ or initial_state.name == "Resuming"
  ):
  await self.reject_transition(
  state=states.Cancelled(),

tests/server/orchestration/test_core_policy.py

@@ -2994,6 +2994,35 @@ async def test_rejects_cancelling_suspended_flow_and_sets_to_cancelled(
  assert ctx.response_status == SetStateStatus.REJECT
  assert ctx.validated_state_type == states.StateType.CANCELLED
 
+ async def test_rejects_cancelling_resuming_flow_and_sets_to_cancelled(
+ self,
+ session,
+ initialize_orchestration,
+ ):
+ """Suspended flows should skip the cancelling state and be set immediately to cancelled
+ because they don't have infra to shut down.
+ """
+
+ intended_transition = (states.StateType.SCHEDULED, states.StateType.CANCELLING)
+
+ ctx = await initialize_orchestration(
+ session,
+ "flow",
+ *intended_transition,
+ initial_state_name="Resuming",
+ )
+
+ # Resuming flows have infra pids
+ ctx.run.infrastructure_pid = "my-pid-42"
+
+ async with BypassCancellingFlowRunsWithNoInfra(
+ ctx, *intended_transition
+ ) as ctx:
+ await ctx.validate_proposed_state()
+
+ assert ctx.response_status == SetStateStatus.REJECT
+ assert ctx.validated_state_type == states.StateType.CANCELLED
+
  async def test_accepts_cancelling_flow_run_with_pid(
  self,
  session,