Human approvals

Pause a task until a human operator approves or rejects it. Declare requires_approval on @task.with_options(...) — bool or a predicate — and the engine emits a durable approval row that operators decide via CLI or HTTP.

Some steps in a workflow should not run unattended. A production deploy, a refund above a threshold, a destructive cleanup — those are the moments where a human should look at the inputs and say yes, go or no, stop. Flux exposes this as a primitive on every task: @task.with_options(requires_approval=...) makes the task pause before its body runs and wait for an operator’s decision.

The same primitive is what powers agent tool approval; this page covers the underlying mechanism and the non-agent use cases.

Gating a task with requires_approval=True

from flux import ExecutionContext, task, workflow


@task
async def build_artifact(environment: str) -> str:
    return f"artifact-{environment}-1.0.0"


@task
async def run_smoke_tests(artifact: str) -> bool:
    return True


@task.with_options(requires_approval=True)
async def deploy_to_environment(*, environment: str, artifact: str) -> str:
    return f"deployed {artifact} to {environment}"


@workflow
async def deploy_workflow(ctx: ExecutionContext[dict]):
    raw = ctx.input or {}
    environment = raw.get("environment", "staging")

    artifact = await build_artifact(environment)
    if not await run_smoke_tests(artifact):
        return {"status": "failed", "stage": "smoke_tests"}

    deployment = await deploy_to_environment(environment=environment, artifact=artifact)
    return {"status": "ok", "deployment": deployment}

Run it inline:

ctx = deploy_workflow.run({"environment": "prod"})
print(ctx.is_paused)        # True — waiting on the deploy approval
print(ctx.execution_id)

The workflow runs build_artifact and run_smoke_tests normally, then pauses at deploy_to_environment before its body executes.

Conditional approval (predicates)

A boolean is fine for “always gate this task.” For “only gate in certain conditions,” pass a callable that returns bool (or an awaitable bool). The predicate receives the same arguments as the task:

@task.with_options(
    requires_approval=lambda environment, **_: environment == "prod",
)
async def deploy_to_environment(*, environment: str, artifact: str) -> str:
    return f"deployed {artifact} to {environment}"

Now staging deploys go through unattended; only environment="prod" pauses. The predicate is evaluated once, before the task body runs — its result is recorded in the approval row and never re-evaluated.

Async predicates work too:

@task.with_options(
    requires_approval=async_check_compliance,   # async def → awaitable[bool]
)
async def issue_refund(amount: float, customer: str) -> None: ...

If the predicate raises, the task call fails. Predicate exceptions are not silently mapped to approve or reject.

Acting as the approver

CLI

List pending approvals:

flux execution approvals

Useful filters: --execution <id>, --workflow <ns>/<name>, --task <name>, --age 1h, --status all|approved|rejected|cancelled, --limit N, --json.

Approve a specific task call:

flux execution approve <execution_id> <task_call_id> --reason "tests green"

Add --always to make the approval a standing grant that also covers every later gate on the same task within this execution:

flux execution approve <execution_id> <task_call_id> --always

Reject it:

flux execution reject <execution_id> <task_call_id> --reason "deploying off-hours"

flux execution show <execution_id> lists the execution’s pending approvals on stderr (stdout stays clean for json.loads). flux workflow status <workflow> <execution_id> appends a Blocked on N approval(s) line on stderr when the execution is paused on one or more gates.

HTTP API

POST /executions/{execution_id}/approvals/{task_call_id}/approve
Content-Type: application/json

{"reason": "tests green", "always": false}
POST /executions/{execution_id}/approvals/{task_call_id}/reject
Content-Type: application/json

{"reason": "deploying off-hours"}

always: true on approve creates a standing grant covering later gates on the same task within the execution; it is ignored on reject. 200 returns the post-decision row, including its scope (call for a plain decision, execution for a standing grant — the same field appears in the listing endpoints below). 409 returns {"error": "already_decided", "current_status": ..., "decided_at": ...} — useful when multiple operators might race on the same gate; only the first decision is recorded, and the response does not leak the winning approver’s identity.

Listing endpoints mirror the CLI:

GET /approvals[?status=&execution_id=&workflow_namespace=&workflow_name=&task_name=&age_min=&limit=&offset=]
GET /executions/{execution_id}/approvals
GET /executions/{execution_id}/approvals/{task_call_id}

Permissions

Deciding an approval requires the workflow:{namespace}:{name}:task:{task_name}:approve verb. The built-in operator role has it via workflow:*:*:task:*:approve. A user with workflow:*:*:read can see a pending approval but cannot decide it without the approve verb. Wire your role grants accordingly.

Standing approvals

By default an approval covers exactly one task call. When the same gated task runs many times in one execution — an agent’s shell tool, a deploy step in a loop, retry attempts — approve with a standing grant to cover every later gate on the same task name within that execution:

flux execution approve <execution_id> <task_call_id> --always

Or over HTTP, pass "always": true in the approve body.

Semantics:

For a runnable end-to-end example, see Standing grant.

When a task is rejected

The engine raises ApprovalRejected at the call site:

from flux import ExecutionContext, task, workflow
from flux.approvals import ApprovalRejected


@task.with_options(requires_approval=True)
async def deploy_to_environment(*, environment: str, artifact: str) -> str: ...


@workflow
async def deploy_workflow(ctx: ExecutionContext[dict]):
    raw = ctx.input or {}
    try:
        deployment = await deploy_to_environment(environment=raw["environment"], artifact=raw["artifact"])
    except ApprovalRejected as e:
        # e.reason, e.approver_subject, e.approver_provider are available
        return {"status": "rejected", "by": e.approver_subject, "reason": e.reason}
    return {"status": "ok", "deployment": deployment}

Replay and durability

The approval row is the durable record. On replay or worker reclaim:

A non-deterministic predicate (time, random, external state) cannot flip the verdict between runs because the row is looked up before the predicate is consulted on subsequent passes. The one narrow window is a worker crash between predicate evaluation and the next event flush — in that case the predicate can re-run on reclaim. This is the same restart-race contract that applies to ordinary task bodies.

A rejection emits TASK_FAILED with the ApprovalRejected exception persisted via the task’s configured output storage. On replay, the event log short-circuits the call and re-raises the same exception, so the workflow body sees identical behavior on the original run and every replay.

Retries trigger a fresh approval

Each retry attempt is a fresh task call: the predicate re-evaluates and a new approval is required. The previous attempt may have had partial side effects, and the approver should reconsider. “Approve once, run forever” would be a footgun — which is why covering later gates (including retry attempts) takes the explicit standing grant, never a plain approval.

A task suspended at a retry-attempt approval gate resumes at the correct attempt once decided: the retry history in the event log is durable, so the engine re-enters the retry chain where it left off instead of re-running the original attempt and duplicating its side effects.

Cancellation

If a workflow is cancelled while paused on an approval, all pending approval rows for that execution transition to cancelled. They do not emit TASK_REJECTED — rejection implies an approver acted; cancellation does not. Cancellation handling at the workflow level takes over from there.

Limitations

Next steps