Pause, Resume, and Human-in-the-Loop Flows

Sooner or later a pipeline needs a person in the middle. Someone has to approve a backfill that will rewrite six months of a table. Someone has to eyeball a reconciliation report before it goes to finance. Someone has to choose which of two remediation paths the job takes after it finds a data-quality break. The work isn’t fully automatable — not because the code can’t do it, but because a human is meant to make the call, and the pipeline should stop and ask.

Airflow has no first-class answer to this. What it has is a set of workarounds you bolt together: a PythonSensor that polls an external flag, an Airflow Variable a human flips through the UI, a task that blocks on a Slack webhook, or the honest-but-grim airflow tasks run ... --mark-success executed by hand. Every one of these lives outside the DAG’s own state model. The approval isn’t a flow state; it’s a side effect you invented, stored somewhere Airflow doesn’t know about, and the DAG’s record of “we waited for Priya to say yes” is, at best, a log line. This is the category of feature where the honest thing to say is: Prefect does something Airflow simply doesn’t. Waiting for a human is a state a flow run can be in, and resuming it is a first-class transition with the input the human provided attached to the run forever.

The Airflow version, and why it grates

Picture the approval gate concretely in Airflow. You want a DAG that computes how many rows a cleanup will delete, stops, waits for a human to approve, then deletes. The least-bad implementation is roughly:

# The Airflow shape — a sensor polling external state
from airflow.sdk import DAG, task
from airflow.providers.standard.sensors.python import PythonSensor
from airflow.models import Variable

@task
def count_doomed_rows() -> int:
    ...  # returns, say, 42_000

def approval_granted() -> bool:
    return Variable.get("prune_approved", default_var="false") == "true"

with DAG("prune_orders", schedule=None, catchup=False) as dag:
    count = count_doomed_rows()
    wait = PythonSensor(
        task_id="await_approval",
        python_callable=approval_granted,
        mode="reschedule",
        poke_interval=300,
        timeout=60 * 60 * 24,
    )
    # ... a delete task downstream, gated on `wait`
    count >> wait

Look at what you had to invent. The approval lives in a Variable — global mutable state with no connection to this run, so if two runs are waiting, one human flip approves both. The human doesn’t see the row count the sensor is guarding; the “42,000” is in a log somewhere and the approver approves blind. There’s no record of who approved or why attached to the run — the audit trail is “the variable was true when we poked.” And the reschedule-mode sensor still occupies a slot in the scheduler’s bookkeeping the whole time. You can make this work. You cannot make it clean, because Airflow’s state machine has no state that means “paused, waiting on a person, here is the schema of what I need from them.”

That sentence is exactly the Prefect feature.

pause_flow_run versus suspend_flow_run

Prefect gives you two ways to stop a flow mid-run and wait, and the difference between them is the whole design decision — get it wrong and you either burn infrastructure for a day or silently re-run work you didn’t mean to.

pause_flow_run freezes the flow in place. The Python process stays alive, holding every local variable in memory, and blocks on a poll loop waiting to be resumed. The flow run enters a Paused state; the worker (or the python my_flow.py process) is still there, still yours, still costing you whatever a running process costs.

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

@flow(log_prints=True)
def nightly_report():
    report = build_report()          # local variable, held in memory
    pause_flow_run(timeout=600)      # process blocks here, up to 10 minutes
    publish(report)                  # `report` is still right here on resume

suspend_flow_run is the stronger move. It doesn’t block — it exits. The flow run enters a Suspended state, the engine returns control, and the infrastructure running it can be torn down entirely: the worker is freed, the container stops, the Kubernetes job completes, you stop paying for idle compute. When someone resumes the run later, Prefect reschedules it from the beginning — a fresh process picks it up and re-executes the flow function from the top.

from prefect import flow
from prefect.flow_runs import suspend_flow_run

@flow(log_prints=True)
def nightly_report():
    report = build_report()          # runs now...
    suspend_flow_run(timeout=86400)  # process EXITS; infra torn down
    publish(report)                  # ...and build_report() runs AGAIN on resume

Read those two comments on the last line carefully, because that is the trap. With pause, the process never died, so report is the same object you built a moment ago. With suspend, resuming re-runs the flow from the start — build_report() executes again, and if it isn’t deterministic (or its results aren’t cached), the thing you publish after approval is not the thing that was computed before the wait. Suspension trades “hold the process” for “hold nothing,” and the price is that your flow has to survive being re-run from the top.

The way you pay that price is the caching you already met earlier in this series: mark the expensive, pre-suspension tasks so their results persist and get reused instead of recomputed.

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

@task(persist_result=True, cache_expiration=timedelta(days=2))
def build_report():
    ...  # heavy; on resume this returns the cached result, doesn't recompute

@flow
def nightly_report():
    report = build_report()          # first run: computes. resume: cache hit.
    suspend_flow_run(timeout=86400)
    publish(report)                  # same report both times, cheaply

So the rule of thumb writes itself. A brief gate — seconds to a few minutes, a human who’s already watching — use pause. The process is cheap to hold and you get consistency for free. A long wait — an approval that might sit until tomorrow morning — use suspend, free the infrastructure, and make sure the pre-suspension tasks are cached so the resume is consistent and cheap. Pausing a flow for eighteen hours means eighteen hours of paid idle worker; suspending it means zero, at the cost of one line of persist_result=True.

Resuming, and the timeout that bites

A stopped flow comes back three ways. In the UI, a Paused or Suspended run shows a Resume button (and, if it’s waiting on input, the form to fill — more on that next). From the CLI, prefect flow-run resume <flow-run-id>. And programmatically, from another flow, a script, or an automation:

from prefect.flow_runs import resume_flow_run

resume_flow_run("6f2c…")  # by flow run id

That last form is how you wire approvals to anything: a Slack slash-command handler, an internal admin page, a webhook — whatever your reviewers already use — calls resume_flow_run and the run comes back to life.

Now the edge that catches everyone: both functions take a timeout, and blowing past it fails the run — but their defaults are not the same, and neither is what you’d guess. suspend_flow_run defaults to timeout=None, which means no timeout at all: a suspended flow waits indefinitely, which is usually what you want from a human approval and is worth knowing before you go adding one “just in case.” pause_flow_run defaults to 3600 seconds — one hour — after which the run transitions to Failed, not back to running. (Older Prefect 2 material puts this default at five minutes; it isn’t, and hasn’t been for a while. Check it rather than trusting a tutorial, yours truly included.) An hour is generous for a machine and short for a human being who might be at lunch, so if your approver is a person, set the timeout to the real human timescale you expect rather than inheriting one:

pause_flow_run(timeout=1800)            # 30 minutes for a watching operator
suspend_flow_run(timeout=60 * 60 * 24)  # 24 hours for an async approval

Treat the timeout as a deadline for the human, not a formality. It’s the answer to “how long should this wait before we give up and call it abandoned?”

There’s one more prerequisite that trips people up and follows straight from how suspension works. Because suspend_flow_run exits the process, resuming means a new process has to pick the run up — and something has to schedule that new process. That something is a worker pulling from a work pool, which means a suspended run can only resume if the flow runs as part of a deployment. Suspend an interactive python prune.py run started from your terminal and there’s nothing left to reschedule it onto; the infrastructure is gone and no worker owns the run. pause_flow_run has no such requirement — the process is still alive, so even an ad-hoc local run can pause and resume. The practical reading: pause works anywhere, including at your desk; suspend is a production tool that assumes deployments and workers, which is exactly the setting where its “free the infrastructure” payoff matters.

Resuming can also carry data back in. When a run is waiting on input, the UI form is the usual path, but you can resume programmatically and supply the answers as a dict, which is how you drive approvals from Slack, an internal page, or an automation:

from prefect.flow_runs import resume_flow_run

resume_flow_run(
    "6f2c…",
    run_input={"approved": True, "reason": "Confirmed with data-gov, 2026-06-24"},
)

The dict is validated against the same RunInput model the flow declared, so a bad payload is rejected here just as it would be in the form — the schema is enforced no matter which door the answer comes through.

Typed human input with wait_for_input

Stopping is half of it. The better half is that when Prefect stops to ask a person a question, it can ask a structured question and get back validated, typed data — not a boolean flag scraped from a Variable, but a real object your code can branch on.

You describe what you need from the human as a RunInput, which is a Pydantic v2 model (subclassing RunInput instead of BaseModel), and hand it to pause_flow_run or suspend_flow_run through the wait_for_input argument. Prefect turns the model’s fields into a form in the UI, validates the human’s answers against the model, and returns you a populated instance.

from prefect import flow
from prefect.flow_runs import pause_flow_run
from prefect.input import RunInput

class Approval(RunInput):
    approved: bool
    reason: str = ""

@flow
async def gated_flow():
    decision = await pause_flow_run(wait_for_input=Approval)
    if decision.approved:
        ...

Two things to notice. First, the pause/suspend call returns the filled-in Approval instance — the value comes back to you where you asked for it, rather than landing in some side channel you then have to read. (You’ll see this written both ways in the wild. pause_flow_run is sync-compatible: call it from an ordinary def flow and it just blocks; await it from an async def flow and it dispatches to the async implementation. Async is a choice here, not a requirement — pick whichever your flow already is.) Second, that instance is typed: decision.approved is a real bool, validated by Pydantic before it ever reaches your code. If the reviewer somehow submits a non-boolean, the form rejects it; your flow only ever sees data that satisfies the schema. This is the thing the Airflow Variable can never give you — the input is shaped, validated, and bound to this run.

Defaults and the description make the form usable. RunInput subclasses get a with_initial_data helper that pre-fills fields and, importantly, sets the markdown blurb shown above the form — your chance to tell the approver what they’re approving:

decision = await pause_flow_run(
    wait_for_input=Approval.with_initial_data(
        description=f"About to delete **{doomed:,}** rows. Approve?",
        approved=False,   # form pre-checks to "no" — safe default
    ),
    timeout=1800,
)

Validation is just Pydantic, which means you can enforce policy in the model itself. Want to require a reason whenever someone rejects? A model validator makes the form refuse an empty rejection:

from pydantic import model_validator
from prefect.input import RunInput

class Approval(RunInput):
    approved: bool
    reason: str = ""

    @model_validator(mode="after")
    def reason_required_on_reject(self):
        if not self.approved and not self.reason.strip():
            raise ValueError("A reason is required when rejecting.")
        return self

The reviewer who unchecks “approved” and submits a blank reason gets a validation error in the form, not a broken flow. The rule lives with the data, once, instead of being re-checked in every flow that uses the gate.

A worked approval gate

Here is the whole pattern end to end — the ETL that stops for a human before a destructive delete, done properly. Because the wait might be long, it suspends (frees the worker); because suspension re-runs the flow from the top, the count is cached so the number the approver saw is the number that gets deleted.

from datetime import timedelta
from prefect import flow, task
from prefect.flow_runs import suspend_flow_run
from prefect.input import RunInput
from pydantic import model_validator


class DeleteApproval(RunInput):
    approved: bool
    reason: str = ""

    @model_validator(mode="after")
    def reason_required_on_reject(self):
        if not self.approved and not self.reason.strip():
            raise ValueError("A reason is required when rejecting.")
        return self


@task(persist_result=True, cache_expiration=timedelta(days=1))
def count_stale_orders(threshold_days: int) -> int:
    # Cached: on resume this returns the same count it computed pre-suspend,
    # so the approver's "42,000" is exactly what we delete.
    return _count_older_than(threshold_days)


@task
def delete_stale_orders(threshold_days: int) -> int:
    return _delete_older_than(threshold_days)


@flow(log_prints=True)
async def prune_stale_orders(threshold_days: int = 90, auto_below: int = 10_000):
    doomed = count_stale_orders(threshold_days)
    print(f"{doomed:,} orders older than {threshold_days} days")

    # Small deletes go through unattended; only big ones need a human.
    if doomed >= auto_below:
        decision = await suspend_flow_run(
            wait_for_input=DeleteApproval.with_initial_data(
                description=(
                    f"This will permanently delete **{doomed:,}** orders "
                    f"older than {threshold_days} days. Approve?"
                ),
                approved=False,
            ),
            timeout=60 * 60 * 24,  # a human has 24 hours to answer
        )
        if not decision.approved:
            print(f"Delete rejected: {decision.reason}")
            return {"deleted": 0, "rejected_reason": decision.reason}

    deleted = delete_stale_orders(threshold_days)
    print(f"Deleted {deleted:,} orders")
    return {"deleted": deleted}

Walk the run. The flow computes the count and prints it. If the count is under the threshold, no human is involved — routine cleanups don’t page anyone. If it’s over, the flow suspends: the process exits, the worker is freed, and the run sits in Suspended state showing an approval form seeded with the exact row count. A reviewer opens the run in the UI (or gets a Slack link from an automation on the Suspended state — the state-hooks post covers wiring that), reads “permanently delete 42,000 orders,” and either approves or rejects with a reason. On approve, Prefect reschedules the run; count_stale_orders is a cache hit so no recompute and no drift; the delete proceeds against the same number the human saw. On reject, the flow returns cleanly with the reason recorded. Either way the decision — who, when, approved or not, and why — is attached to this flow run in Prefect’s own state store, which is the audit trail the Airflow version never had.

Compare this to the sensor-and-Variable contraption from the top. Same outcome, but the approval is scoped to the run, the approver sees what they’re approving, the rejection reason is captured and validated, and no worker sits poking a flag for a day.

Beyond yes/no: choosing a path

Approval is the common case, but the input model is a full Pydantic model, so the human can hand back anything you can type — including a choice. When a data-quality check breaks, you might want an operator to pick the remediation rather than fail the run outright. Model the choice with an Enum (or a Literal), and Prefect renders it as a dropdown instead of a free-text box:

from enum import Enum
from prefect import flow, task
from prefect.flow_runs import suspend_flow_run
from prefect.input import RunInput


class Remediation(str, Enum):
    quarantine = "quarantine"   # move bad rows aside, keep loading the rest
    halt = "halt"               # stop the load, page the on-call
    ignore = "ignore"           # accept the rows as-is, log it


class QualityDecision(RunInput):
    action: Remediation = Remediation.halt   # safe default: stop
    note: str = ""


@flow(log_prints=True)
async def load_with_gate():
    bad = check_quality()                    # a task that counts violations
    if bad == 0:
        return finish_load()

    decision = await suspend_flow_run(
        wait_for_input=QualityDecision.with_initial_data(
            description=f"Found **{bad:,}** quality violations. How should we proceed?",
        ),
        timeout=60 * 60 * 4,
    )

    if decision.action is Remediation.halt:
        raise RuntimeError(f"Load halted by operator: {decision.note}")
    if decision.action is Remediation.quarantine:
        quarantine_bad_rows()
    return finish_load()                      # runs for quarantine and ignore

The operator gets a dropdown of exactly the three legal actions — they can’t fat-finger a typo into a policy decision — and the flow branches on a typed enum, not a parsed string. This is the general shape: the human’s judgment becomes structured data, the set of valid answers is enforced by the schema, and the branch reads like ordinary Python because that’s all it is.

Streaming input, briefly

wait_for_input is a single question with a single answer. Prefect also supports a conversation — a flow that receives inputs over time while it keeps running — through receive_input and send_input. This is the machinery behind interactive, long-lived flows (think a flow backing a chatbot, or a supervisor flow taking a stream of commands), and it’s more than most pipelines need, so treat this as a pointer rather than a deep dive.

A receiving flow loops over inputs as they arrive:

from prefect import flow
from prefect.input import receive_input

@flow
async def responder():
    async for name in receive_input(str, timeout=None, poll_interval=1):
        print(f"Hello, {name}!")

And any other flow (or process holding the run id) sends into it:

from prefect.input import send_input

await send_input("Ada", flow_run_id=responder_run_id)

receive_input yields each input as it lands — you can type it as a primitive like str or as a RunInput model for structured messages — and send_input pushes one in by flow-run id. The mental model: wait_for_input is a form the flow stops to fill once; receive_input/send_input is a mailbox the flow reads from while it runs. Reach for the mailbox only when you genuinely have an ongoing exchange; for approvals, gates, and choose-a-path prompts, the single wait_for_input is the right tool and the simpler one.

When not to reach for this

Human-in-the-loop is a power tool, and the failure mode is over-using it. A gate is only as reliable as the human behind it, and a human is the slowest, least available component in any pipeline. Put a pause_flow_run in a job that runs every fifteen minutes and you’ve built a system that fails every time nobody’s watching — which, at 3 a.m., is always. The timeout will fire, the run will fail, and you’ll have manufactured an outage out of a step that should have been automated.

So keep the gates where they earn their keep: high-stakes, low-frequency, genuinely-needs-judgment operations — the destructive delete, the six-month backfill, the report that carries legal weight, the remediation path a machine shouldn’t pick alone. For anything routine, encode the decision as a rule and let the flow make it — the auto_below threshold in the worked example is exactly this, letting small deletes through untouched so a person is only ever asked about the ones that matter. If you find yourself reaching for a human gate on a daily pipeline, the real fix is usually a better automated check, not a person clicking Approve at dawn.

Final thoughts

The reason Airflow never grew a clean version of this is that its state machine was built around a graph the scheduler parses ahead of time, and “waiting on a person” isn’t a shape you can draw at parse time — it’s a thing that happens, unpredictably, mid-run. So Airflow engineers approximate it from the outside with sensors and shared variables, and the approval ends up living everywhere except in the run’s own record. Prefect discovers the graph as the code runs, which means “stop here and ask a human” is just another thing the code can do, and Prefect made it a real state with real, typed, validated input bound to the run. The two knobs to keep straight are the ones this post turns on: pause when the process is cheap to hold and you want the in-memory value, suspend when the wait is long and you’ll cache your way to a consistent resume — and a timeout that reflects how long a human actually gets. Use it sparingly, on the operations that deserve a person’s judgment, and the audit question — who approved, what they saw, what they said, what happened next — finally has an answer that lives inside the flow instead of beside it.

Next: Artifacts and Logging