How to pause a run for approval

Call wait_for_input where the flow needs a decision. The run pauses with the question visible in the UI and the API, the engine is released, and the answer resumes it. Pausing needs a running server; offline, the question is asked on the terminal. A question is a message on the input topic: wait_for_input(prompt) is receive("input", prompt=prompt), and Send a message to a running flow covers other topics, timeouts, and state a run publishes while it waits.

Ask the question

from datetime import date
from cereyan import flow, task, wait_for_input, INPUTS

@task(cache=INPUTS, persist_result=True)
def prepare(day: date) -> int:
    return 1200

@task
def release(rows: int) -> str:
    return f"released {rows}"

@flow
def publish(day: date) -> str:
    rows = prepare(day)
    decision = wait_for_input(
        f"Release {rows} rows for {day}?",
        schema={"type": "object", "properties": {"approve": {"type": "boolean"}}, "required": ["approve"]},
    )
    return release(rows) if decision["approve"] else "held"

The first call to wait_for_input moves the run to Paused with the prompt and schema in its state details, records run.paused, and ends the attempt. The tasks before the question are checkpointed, so the resumed attempt does not redo their work; cache=INPUTS is no longer needed for that, though it still helps when the same inputs recur across runs.

Answer it

On the run page, the Details tab shows the question with a form built from the schema and a Resume button. Programmatically, against the Approval example, whose publish asks a second question after the first is answered:

run = served.client.run("publish", day="2026-03-01")
for question in ("Release", "Publish"):
    paused = served.wait_run(
        run["id"],
        until=lambda r: r["state"]["type"] == "Paused" and r["state"]["details"]["prompt"].startswith(question),
    )
    served.client.resume(run["id"], {"approve": True})
done = served.wait_run(run["id"])
assert done["state"]["type"] == "Completed"

POST /api/runs/{id}/resume with {"input": ...}, Client.resume, and the MCP resume_run tool do the same: each sends only the answer, and the server stores it against the question the run is waiting on. GET /api/runs/{id}/input returns that pending question and the answers given so far, and ?index= returns one answer. Resuming a run that is not paused answers 409.

What happens on resume

The answer is stored, run.resumed is recorded, and a new attempt of the same run is scheduled. It reruns the flow from the top: tasks completed before the pause return their checkpoints as Replayed task runs (and tasks with cache=INPUTS their cached results as Cached ones), wait_for_input returns the answer instead of pausing, and the rest of the flow executes. Any code between the top of the flow and the question that is not in a cached task runs again, so keep side effects inside tasks.

While it waits

A paused run counts as active, appears on the dashboard, can be cancelled, and holds no engine and no resources. Use a proactive rule to notice a run that waits too long:

from cereyan import App

app = App("approvals")

@app.rule(on="run.paused", flow="publish", unless="run.resumed", within=4 * 3600)
def nobody_answered(event, run):
    print(f"{run['name']} has been waiting for four hours")

Several questions

Each wait_for_input call in a flow pauses in turn, and a flow can ask, resume, and ask again. Questions are numbered in the order the body reaches them, and an answer is stored against the question it answered, so the second question waits for its own answer instead of receiving the first one. Two questions that use the same words are still two questions.

An answer is used only while the prompt at that position still matches the one it was given for. Edit the flow between a pause and a resume so that a different question now sits at that position, and the run asks it rather than handing over an answer meant for something else. A prompt built from a value that changes between executions — a timestamp, say — will therefore ask again, so keep prompts stable for the same question.

wait_for_input belongs in the flow body. Pausing works by raising out of the body, so a call from a task submitted with submit or map raises CereyanError instead of pausing the run.

Offline

Without a server, wait_for_input reads the answer from the terminal (as JSON when a schema is given) and raises CereyanError when there is no interactive terminal, so scripts under cron fail fast instead of hanging.

Related: Send a message to a running flow, Use cereyan with an AI agent for answering questions from an agent, Cache task results.