Pause and resume

Workflows don’t always run to completion in one shot. An approval gate, a manual review step, or a hand-off to an external system can require the workflow to wait for minutes, hours, or days before continuing. Flux’s pause() primitive suspends execution at a named checkpoint and persists everything, so the workflow picks up exactly where it left off when you resume it.

The pause() task

pause() lives in flux.tasks. Call it inside a workflow to suspend execution at a named point:

from flux import ExecutionContext
from flux.task import task
from flux.tasks import pause, sleep
from flux.workflow import workflow
from datetime import timedelta


@task
async def process_data():
    await sleep(timedelta(seconds=2))
    return "Data processed"


@workflow
async def approval_workflow(ctx: ExecutionContext):
    result = await process_data()

    # Suspend here until resumed
    await pause("wait_for_approval")

    return result + " and approved"

When the workflow reaches await pause("wait_for_approval"), Flux:

1. Emits a WORKFLOW_PAUSED event and persists it.
2. Raises PauseRequested internally, which the runtime catches cleanly.
3. Returns control to the caller. workflow.run() returns a context in state PAUSED.

To resume, call .run() again with the same execution_id:

# First run — stops at the pause point
ctx = approval_workflow.run()
print(ctx.state)            # PAUSED

# Resume from where it left off
ctx = approval_workflow.run(execution_id=ctx.execution_id)
print(ctx.state)            # COMPLETED
print(ctx.output)           # "Data processed and approved"

Flux replays all previously completed tasks deterministically (their outputs are cached in the event log) and re-runs only the remaining work. Nothing executes twice.

Checking pause state

After .run() returns, inspect ctx.is_paused to branch on whether the workflow suspended:

ctx = approval_workflow.run()

if ctx.is_paused:
    print(f"Workflow {ctx.execution_id} is waiting at a pause point.")
    # Store ctx.execution_id for later resumption

Passing input on resume

pause() returns the value provided at resume time, so you can collect decisions or data from an operator without storing them out-of-band:

from flux import ExecutionContext
from flux.task import task
from flux.tasks import pause
from flux.workflow import workflow


@task
async def initial_task():
    return {"stage": "initial", "data": [1, 2, 3]}


@task
async def process_with_input(initial_data, user_input):
    multiplier = 1
    if user_input and isinstance(user_input, dict):
        multiplier = user_input.get("multiplier", 1)
    return {
        "stage": "processed",
        "result": sum(initial_data["data"]) * multiplier,
    }


@workflow
async def review_workflow(ctx: ExecutionContext):
    initial_result = await initial_task()

    # pause() returns whatever input is provided at resume time
    user_input = await pause("waiting_for_user_input")

    return await process_with_input(initial_result, user_input)

Resuming from another process (SDK client)

When the workflow ran on a server and another process needs to hand it input — an approver service, a webhook handler, a notebook — use FluxClient. The client talks to the server’s secured HTTP API; it never touches the runtime in-process:

import asyncio
from flux.client import FluxClient

async def approve(execution_id: str):
    async with FluxClient("http://localhost:8000") as flux:
        result = await flux.resume_execution_sync(
            workflow_ref="review_workflow",
            execution_id=execution_id,
            input_data={"multiplier": 5, "comment": "Approved"},
        )
        print(result["output"])  # {"stage": "processed", "result": 30}

asyncio.run(approve("abc123"))

resume_execution_sync blocks until the workflow completes or hits the next pause. resume_execution is the fire-and-forget variant — returns immediately with the execution metadata, and a worker picks up the resume. Both accept input_data as the value pause() will return inside the workflow; pass None (or omit it) when the pause point doesn’t need input.

Resuming in the same process (inline)

For inline scripts and tests — the SDK form used in the snippets above — call workflow.resume(execution_id, input):

# First run — pauses waiting for input
ctx = review_workflow.run()

# Resume with input
ctx = review_workflow.resume(ctx.execution_id, {"multiplier": 5, "comment": "Approved"})
print(ctx.output)           # {"stage": "processed", "result": 30}

When no input is provided on resume, pause() returns None.

Multiple pause points

A workflow can contain as many pause points as the process requires. Each call to pause() takes a distinct name that appears in the event log:

from flux import ExecutionContext
from flux.task import task
from flux.tasks import pause
from flux.workflow import workflow


@task
async def init_process():
    return {"stage": "init"}


@task
async def load_data(state):
    return {**state, "stage": "data_loaded", "records": 1000}


@workflow
async def multi_stage_workflow(ctx: ExecutionContext):
    state = await init_process()

    await pause("verify_setup")         # Stop 1: operator verifies config

    state = await load_data(state)

    await pause("validate_data")        # Stop 2: operator validates loaded data

    return {**state, "stage": "complete"}

Each .run() call advances the workflow past one pause point:

ctx = multi_stage_workflow.run()
# ctx.is_paused == True, stopped at "verify_setup"

ctx = multi_stage_workflow.run(execution_id=ctx.execution_id)
# ctx.is_paused == True, stopped at "validate_data"

ctx = multi_stage_workflow.run(execution_id=ctx.execution_id)
# ctx.is_paused == False, ctx.state == "COMPLETED"

You can also generate pause point names dynamically, which is useful when the number of checkpoints isn’t known until runtime:

@workflow
async def chunked_workflow(ctx: ExecutionContext):
    total_chunks = 3
    for chunk_id in range(1, total_chunks + 1):
        state = await process_chunk(state, chunk_id)
        if chunk_id < total_chunks:
            await pause(f"monitor_progress_{chunk_id}")
    return state

Resume via the CLI

In a distributed deployment (server + worker), resume through the CLI rather than the SDK:

flux workflow resume <workflow-name> <execution-id> 'null'

To pass input on resume, supply a JSON value as the third argument:

flux workflow resume review_workflow abc123 '{"multiplier": 5, "comment": "Approved"}'

The --mode flag controls how the CLI waits:

flux workflow resume review_workflow abc123 'null' --mode sync

How replay preserves correctness

When a paused workflow resumes, Flux replays the event log from the beginning. Every completed task’s output is read from the log; the task body does not run again. Only the tasks after the pause point execute fresh, which means:

• Task side effects (network calls, writes) do not repeat.
• Non-deterministic operations (uuid4, now, random values) return their original results.
• The workflow sees a consistent view of state across the pause boundary.

For more detail on the replay mechanism, see Replays and re-execution.

Reference

• Reference: flux workflow resume
• Human approvals — pause() is the right call for waiting for input; when the use case is specifically “a human must approve this task before it runs,” use @task.with_options(requires_approval=...) instead, which adds a durable approval row, RBAC verbs, and flux execution approve|reject CLI commands.