Short answer: this is expected behavior. On a resumed run, calling

suspend_flow_run()

again at the same code location will not exit unless you give it a new key. Prefect de-duplicates suspends by a “pause key”; if the same key is used twice in a run, the orchestrator rejects the second suspend and your process continues. Why it happens -

suspend_flow_run()

pauses by raising a

Pause

exception so the worker can exit immediately. - Prefect tracks each suspend with a pause key. By default, a key is auto-generated per call site. After resume, your flow re-starts from the top so the same call site produces the same default key, which is considered “already fired,” so no

Pause

is raised and the process does not exit. - Result: the function returns and your code keeps running. Fix: pass a unique key for each logical suspension Give each suspension a distinct

key

so the orchestrator treats them as new events. Example

from prefect import flow
from prefect.flow_runs import suspend_flow_run

@flow
async def my_flow():
    # First suspend
    await suspend_flow_run(key="stage-1", timeout=3600)

    # ... later in the flow, after the run is resumed ...

    # Second suspend – must use a different key
    await suspend_flow_run(key="stage-2", timeout=3600)

    # You can derive keys from parameters, loop counters, etc.

Common gotchas that also prevent exit - Catching the

Pause

exception. Don’t catch it; let it propagate so the worker can end the process.

# BAD: do not swallow Pause
  try:
      await suspend_flow_run()
  except Exception:  # will catch Pause too
      pass

- Using

pause_flow_run()

instead of

suspend_flow_run()

.

pause_flow_run()

blocks in the current process; it does not exit. - Suspending from a subflow or without a deployment.

suspend_flow_run()

only works for top-level deployed flows; otherwise it will be rejected/aborted. References - suspend_flow_run API - Interactive workflows guide - Orchestration logic for pause keys (duplicate suspends are rejected): core_policy.py If that doesn’t resolve it, can you share: - The snippet where you call

suspend_flow_run()

both times - Whether it’s a deployed top-level flow (not a subflow) - Your worker type/runner and whether there’s any try/except around the call I can help tailor the keying strategy or spot any exception handling that’s swallowing the pause.

Thanks for the snippet — two likely causes of “second suspend doesn’t exit” here: 1) You must await it

suspend_flow_run

is async. If you don’t

await

it, it won’t raise the

Pause

exception that cleanly exits the process. 2) Use a new key for each suspend Prefect de-duplicates suspends by a “pause key.” On resume, the flow starts again from the top; if the same suspend call fires with the same key, the orchestrator rejects it and your code just continues. Give each logical suspend a unique

key

. Try this pattern:

from prefect import flow, task
from prefect.flow_runs import suspend_flow_run

@task
def task1():
    ...

@task
def task2():
    ...

@task
def task3():
    ...

@flow
async def work_flow():
    await task1()
    await suspend_flow_run(key="after-task1", timeout=3600)

    await task2()
    await suspend_flow_run(key="after-task2", timeout=3600)

    await task3()

Checklist to ensure the process exits: - Await the call:

await suspend_flow_run(...)

- Use unique keys per suspend (e.g., “after-task1”, “after-task2”) - Do not catch the

Pause

exception (avoid broad

except Exception:

around it) - Only suspend from a top-level deployed flow (not a subflow, not ad-hoc script runs) - Don’t call it inside a

finally

block that might swallow the exception Docs and references: - suspend_flow_run API - Interactive workflows guide If this still doesn’t exit on the second call, can you share: - Whether this flow is a deployed top-level flow (not a subflow) - Any try/except around the suspend calls - Your worker type and how you’re running the flow (e.g., via

prefect deploy

+ worker)

Great question. In Prefect 3.x, both pause and suspend have a timeout, and the outcome is similar but the mechanics differ slightly. What happens on timeout - pause_flow_run(): - The flow process is actively polling. If the timeout elapses without a resume,

pause_flow_run()

raises

FlowPauseTimeout

in your flow code and the flow run transitions to FAILED with a message like “Flow run was paused and never resumed.” - suspend_flow_run(): - The process has already exited. A server-side service detects that the pause timeout expired and sets the flow run to FAILED with a message like “The flow was suspended and never resumed.” There is no exception in your code because your process is gone. Resume after timeout - If someone tries to resume a paused/suspended run after the timeout, the orchestrator rejects the transition and the run remains/lands in FAILED. wait_for_input behavior - If you used `wait_for_input=...`: - pause_flow_run(): the client-side timeout triggers first; it raises

FlowPauseTimeout

rather than returning input. - suspend_flow_run(): the server fails the run on timeout; input is no longer relevant. Practical notes - The timeout is stored as an absolute timestamp on the paused state (

state_details.pause_timeout

). - For suspend specifically, ensure you set a timeout that’s realistic for the external event you’re waiting on. References - Interactive workflows: guide - API reference: pause_flow_run, suspend_flow_run - Orchestration timeout handling: resume rule, fail expired pauses service

There isn’t a built-in “resume on timeout” option on

pause_flow_run()

or

suspend_flow_run()

in Prefect 3.x. By default, when the timeout elapses they fail. You have three ways to emulate “resume after N seconds”: Option A — Catch the pause timeout and continue - Works with

pause_flow_run()

only (since it stays in-process). - Wrap the call and continue on `FlowPauseTimeout`:

from prefect import flow
from prefect.flow_runs import pause_flow_run
from prefect.exceptions import FlowPauseTimeout
from prefect.logging import get_run_logger

@flow
def my_flow():
    logger = get_run_logger()
    try:
        pause_flow_run(timeout=300)  # 5 min
    except FlowPauseTimeout:
        logger.warning("Pause timed out; continuing automatically")
    # Continue the flow here (acts like “resume on timeout”)

Option B — Prefect Automations to resume after a delay (recommended) - Works for both

pause_flow_run()

and

suspend_flow_run()

. - Create an automation that resumes any run that remains paused for your window:

# auto-resume.yaml
name: Auto-Resume Paused Flows
enabled: true
trigger:
  type: event
  posture: Proactive
  expect:
    - prefect.flow-run.Paused
  within: 300         # seconds to wait before acting
  threshold: 1
  for_each:
    - prefect.resource.id
actions:
  - type: resume-flow-run

- Apply it:

prefect automation create --from-file auto-resume.yaml

- Docs: Automations, Resume action source: ResumeFlowRun Notes: - For

suspend_flow_run()

, use a long

timeout

(e.g., hours) so the automation resumes the run before the pause expiration service fails it. Automations resume immediately when the “paused for N seconds” condition is met. - If a resume happens after the pause’s

pause_timeout

, the orchestrator will reject it and set the run to FAILED; so ensure your automation delay < pause timeout. Option C — A scheduled helper flow that calls

resume_flow_run

- Useful for custom logic or selecting specific runs.

from prefect import flow, task
from prefect.flow_runs import resume_flow_run
from prefect.client.orchestration import get_client

@flow
async def auto_resume_job():
    async with get_client() as client:
        runs = await client.read_flow_runs(flow_run_filter={"state": {"type": {"any_": ["PAUSED"]}}})
        for fr in runs:
            # Your selection criteria (e.g., paused > 5 min)
            await resume_flow_run(fr.id)

Key constraints to remember - There’s no built-in “resume_on_timeout” flag. - Don’t let the pause timeout elapse if you intend to resume: set

timeout

long enough, and have the automation resume sooner than that. - For multiple suspends in one run, pass unique

key

values to each

suspend_flow_run(...)

call. References - Interactive workflows: guide - API: pause_flow_run, suspend_flow_run, resume_flow_run

Because suspend reschedules the run. When you resume a suspended flow run, Prefect starts a fresh process and re-executes the flow from the top. Your code reaches the first

suspend_flow_run()

again, and whether it suspends a second time depends on the pause key for that call. Key behavior - suspend_flow_run() puts the run in a Suspended (Paused with reschedule=True) state and raises a Pause signal to exit the process. On resume, the worker starts the flow again from the beginning. - Prefect uses a pause key to make each pause/suspend idempotent. If the orchestrator has already recorded a pause with that key for the current attempt, it will reject a second pause with the same key and the function returns immediately (no exit). - Default keys: - Suspending the current run: a counter-based key is auto-generated per call site (e.g., "0", "1", ...). - Suspending another run (passing

flow_run_id

): a random UUID is used by default. What you’re seeing - After resume, the new process hits the first

suspend_flow_run()

again. If it uses a new key, it will suspend again; if it reuses the same key that the orchestrator still considers “already fired” for this attempt, the pause is rejected and the call returns immediately, so the process continues. How to control it - If you want the first suspend to happen only once ever for this flow run: give it a stable custom key and Prefect will reject it on subsequent resumes.

await suspend_flow_run(key="after-task1", timeout=3600)

- If you want to suspend again on the next resume: give it a different key each time (e.g., increment a version, include a parameter, or place the call in a later code path so the default counter advances).

await suspend_flow_run(key="after-task1-v2", timeout=3600)
  await suspend_flow_run(key="after-task2", timeout=3600)

Tips - Always await:

await suspend_flow_run(...)

so the Pause signal is raised. - Don’t catch the Pause signal; let it propagate so the worker exits. - Use unique keys for multiple suspensions; use stable keys to avoid re-suspending at the same spot. References - suspend_flow_run API - Orchestration idempotency (pause keys): HandleFlowPausesAndSuspensions - Interactive workflows guide: docs

Short answer: don’t “handle”

suspend_flow_run()

inside `asyncio.run`—let it raise and exit. On resume, Prefect restarts your flow process from the top and your code continues automatically. If you’re running the flow as a plain script, you can catch the signal just to avoid an ugly traceback, but you should not swallow it when running under a worker. How to use it depending on how you run the flow A) Deployed flow (recommended for suspend) - Requirement:

suspend_flow_run()

only works for top‑level deployed flows (not subflows, not ad‑hoc scripts). - You do not need to handle anything on resume. The worker will start a new process, your flow re-enters from the top, and the

suspend_flow_run(...)

call will be skipped on the second pass if it uses the same pause key. - Use a stable key and (optionally)

wait_for_input

so you can continue with input on resume:

from prefect import flow
from prefect.flow_runs import suspend_flow_run

@flow
async def my_flow():
    # First pass: this suspends and exits the process
    # Second pass (after resume): orchestrator rejects duplicate key; function returns,
    # and if wait_for_input was provided, it loads and returns the input.
    approval = await suspend_flow_run(
        key="approve-step",
        wait_for_input=str,
        timeout=86400
    )
    # After resume, execution resumes here; 'approval' is set if provided
    ...

- Important: Always pass a stable

key

for each logical suspension. Without it, the default counter resets in a new process and you may suspend again unintentionally. B) Local script with asyncio.run (for testing) - If you still call the flow directly (not via a worker),

suspend_flow_run()

will raise a

Pause

to exit. You can catch it at the very top just to exit cleanly:

import asyncio, sys
from prefect.exceptions import Pause
from myflow import my_flow

if __name__ == "__main__":
    try:
        asyncio.run(my_flow())
    except Pause:
        print("Flow suspended; exiting.")
        # For local testing, exit cleanly; under a worker, do not catch this
        sys.exit(0)

- Note: Suspending without a deployment will be rejected; use a deployment for real suspend/resume behavior. For true local interactive behavior, use

pause_flow_run()

instead—it blocks in-process and simply returns after resume:

from prefect import flow
from prefect.flow_runs import pause_flow_run

@flow
def my_flow():
    user_val = pause_flow_run(wait_for_input=str, timeout=300)
    # continues here in the same process after resume

Key tips - Always await:

await suspend_flow_run(...)

. - Don’t catch

Pause

inside the flow; let it propagate so the worker exits. Only catch it at the very top of a local script if you want a clean exit message. - Use stable keys to avoid re-suspending at the same spot on resume; use different keys for subsequent suspends. - Optional: use

wait_for_input=...

to receive typed input on resume. References - suspend_flow_run API: docs - Interactive workflows guide: guide - Orchestration idempotency (pause keys): source

Short answer: yes, if your goal is to avoid rescheduling (and the “flow restarts from the top” behavior),

pause_flow_run()

is the simpler choice. Trade-offs - pause_flow_run(): - Blocks in the current process and continues right after the call on resume. - No reschedule, so you won’t re-hit earlier code or worry about duplicate suspend keys at the same site. - Works locally and in subflows; no deployment required. - Downside: the process stays alive (consumes a worker slot). - suspend_flow_run(): - Exits the process and frees the worker; on resume, Prefect re-runs the flow from the top. - To avoid re-suspending at the same spot, pass a stable

key

per logical suspend; the orchestrator will reject a duplicate key and continue. - Requires a top-level deployed flow (not a subflow). Quick patterns

# Avoid rescheduling; continue right after resume
approval = await pause_flow_run(wait_for_input=str, timeout=3600)

# If you need to free the worker, use suspend with stable keys
await suspend_flow_run(key="after-task1", timeout=86400)
# Later in the flow, use a different key
await suspend_flow_run(key="after-task2", timeout=86400)

Docs: - pause_flow_run - suspend_flow_run - Interactive workflows guide

Short answer: the run stays PAUSED. If a worker/process dies while

pause_flow_run()

is blocking, nothing auto-recovers it. When someone resumes, Prefect will attempt to continue the run, but there’s no in-process code left to continue because that process died. The run will remain PAUSED until either: - You manually resume again after a new worker picks it up (not typical for pause), or - The pause timeout expires and a server service marks it FAILED. Details -

pause_flow_run()

blocks in the current process and polls the API. If that process dies, the polling stops and there’s no auto-reschedule. The flow run’s state in the API stays PAUSED. - The “Foreman” only marks workers offline; it does not transition paused flow runs. - A background service fails expired pauses: when

pause_timeout

is exceeded, it sets the run to FAILED with a message like “The flow was paused and never resumed.” -

resume_flow_run()

on a paused run (reschedule=False) would normally set the run to RUNNING to let the same process continue. If the original process is gone, there’s nothing to continue. You’ll see the state transition but effectively no execution resumes, and it can get stuck. Recommendation - Prefer

suspend_flow_run()

for production if you need resilience across worker restarts. Suspend exits the process and, on resume, the run is SCHEDULED so any worker can pick it up cleanly. - If you must use pause, keep timeouts modest so the run doesn’t linger indefinitely after crashes, and consider an Automation to alert/fail or even switch to a resume pattern. References -

pause_flow_run

implementation and in-process polling: source - Resume orchestration rules and behavior: source - Service that fails expired pauses: source - Foreman (worker heartbeat) service: source

Short answer: not in the crash/timeout cases you’re asking about. - Flow/task on_failure hooks exist in Prefect 3.x (via

@flow(on_failure=[...])

,

@flow.on_failure

, and the task equivalents). They run in-process when the engine sets a Failed state during normal execution. - If a worker dies while

pause_flow_run()

is polling, the process is gone—no hooks can run. - If a pause/suspend times out, the server marks the run Failed from the API side. That failure happens outside your process, so on_failure hooks do not run. What to use instead - Use Automations for notifications or recovery on server-side failures and timeouts; they trigger on events like

prefect.flow-run.Failed

, even if the worker is dead. - Automations docs: concepts, how-to - Keep on_failure hooks for in-process cleanup and local side effects when your code raises and the engine records the Failed state. References - Flow hooks API (source): flows.py - Task hooks API (source): tasks.py - Hook invocation point (engine): flow_engine.py - Pause timeout orchestration: core_policy.py