@flow turns a plain Python function into a durable, journaled run: every @task/@agent call it makes is recorded as it happens, so a crash — or a deliberate pause waiting on a human — can be picked up again with resume(run_id), in the same process or a brand-new one, without re-running (or re-paying for) anything that already finished.
import composeai as compose
@compose.task(retries=2, timeout=30, name="fetch_page")
def fetch_page(url: str) -> str:
return f"contents of {url}"@task is usable bare (@compose.task) or with keyword arguments:
retries(default0) — on anyException, retry the task body up to this many times before letting the exception propagate. Each attempt's error type and message are recorded on the task's trace span.timeout(seconds, defaultNone) — bounds one execution on a dedicated daemon thread. A timeout raisesTaskTimeoutErrorimmediately and is never retried (retrying an already-abandoned, still-running thread would just pile up more abandoned threads); there is no safe way to stop the abandoned thread, so treat a timed-out task as failed and move on. The abandoned thread also immediately loses journal write access, so it can't corrupt a flow that has already moved on without it.name— the journal key prefix for this task (defaults to the function's__name__). Task names must be unique per process; a duplicate raisesConfigErrorunlessreplace=True.replace(defaultFalse) — re-bind an existing task name instead of raising. Only affects steps not yet journaled — an already-journaled step's stored value still replays unchanged regardless of what the task's body now does.
A Task object is directly callable whether or not a flow is active: outside a flow it just runs inside a plain trace span; inside one, each call auto-journals as a step.
@compose.flow
def research(topic: str) -> str:
sources = fetch_sources(topic) # journaled step
summaries = compose.map(summarize, sources) # parallel, journaled steps
draft = editor(summaries) # a whole agent run = one step
return draftEvery @task call (and @agent call, and compose.map item, and nested @flow call) made from an active flow body is journaled to the durable store keyed by call order — f"{name}#{n}", with n assigned in flow-body order, never completion order, so concurrent dispatch (compose.map, parallel tool calls) still gets deterministic keys. Each name gets its own counter, so two calls to fetch_sources inside one flow journal under fetch_sources#1 and fetch_sources#2.
flow_obj(topic) is sugar for flow_obj.run(topic).output; .run() always starts a new durable run and returns the full Run (.output, .usage, .trace, .status, .pending). A flow also has .stream(...), same shape as an agent's.
Journaled steps replay instantly on resume() — only the unfinished tail actually executes. Here's the flow from above, saved to a module so it can be imported fresh in a second process:
# research_flow.py
import composeai as compose
@compose.task
def fetch_sources(topic: str) -> list[str]:
return [f"https://example.com/{topic}"]
@compose.task
def summarize(source: str) -> str:
return f"summary of {source}"
@compose.agent(model="anthropic/claude-sonnet-5")
def editor(summaries: list[str]) -> str:
"""Turn source summaries into a short draft."""
return compose.prompt(f"Draft a short paragraph from: {summaries}")
@compose.task
def publish(draft: str) -> str:
return f"published: {draft}"
@compose.flow
def research(topic: str) -> str:
sources = fetch_sources(topic)
summaries = compose.map(summarize, sources)
draft = editor(summaries)
if compose.approve("publish", payload={"draft": draft}):
return publish(draft)
return f"kept as draft: {draft}"Process A starts the run. It pauses on the unanswered approve("publish") gate (more on that below) and the process can simply exit:
# process_a.py
from research_flow import research
run = research.run("quantum computing")
run.status # "paused"
run.id # save this run_id somewhere durable
run.pending # Interrupt(id='publish', kind='approval', question=None, payload={'draft': ...})Process B — hours later, a completely fresh interpreter — imports the same module (so the @flow re-registers under its name) and resumes by run_id:
# process_b.py
from research_flow import research # noqa: F401 -- import registers the @flow
from composeai import resume
run = resume("01K3F8G7QZR3XJ8N4V0T5W2Y1B", answers={"publish": True})
run.status # "completed"
run.output # "published: ..."On this second run, fetch_sources, map(summarize), and editor all replay from the journal — editor's LLM call is not re-made and not re-billed — and only publish (never reached before the pause) actually executes:
└─ ▶ research [3ms]
├─ • fetch_sources [0ms] (replayed)
├─ ⇉ map(summarize) [2ms]
│ ├─ • summarize [0ms] (replayed)
│ └─ • summarize [0ms] (replayed)
├─ ◆ editor [0ms] (replayed)
└─ • publish [0ms]
Replay works by re-running the flow body and substituting journaled step results, in call order, wherever they occur. That only produces the right answer if the flow body is a deterministic function of its journaled step results — side effects, randomness, and clock reads belong inside @task/@agent calls, never directly in the flow body. Nothing detects a violation of this contract; a body that breaks it just won't replay correctly.
Wall-clock reads and randomness are common enough to need a dedicated escape hatch: compose.now() and compose.random() may be called directly in a flow body. Each call journals one value in flow-body order (keys now#1, now#2, ... / random#1, random#2, ...) and replays that exact value verbatim on every resume, so a flow can branch on "what time is it" or "flip a coin" without violating determinism:
import composeai as compose
@compose.flow
def maybe_offer_discount(customer_id: str) -> str:
ts = compose.now() # datetime, timezone-aware UTC -- journaled once
if compose.random() < 0.1: # float in [0, 1) -- journaled once
return f"{customer_id} gets a discount as of {ts.isoformat()}"
return f"{customer_id}: no discount"Outside an active flow, compose.now() is exactly datetime.now(timezone.utc) and compose.random() is exactly random.random() — no journaling, no special behavior.
When to reach for now()/random() vs. a @task: use now()/random() for a single, cheap value the flow body itself needs to branch on (a timestamp to stamp onto output, a coin flip for an A/B split). Reach for @task instead when there's real work involved — network calls, file I/O, anything that should get retries/timeout, or that legitimately takes long enough to want its own span. now()/random() have no retry/timeout knobs of their own; they're a single journaled draw, not a unit of work.
def resume(
run_id: str,
answers: dict[str, Any] | None = None,
*,
budget: Budget | None = None,
allow_code_change: bool = False,
) -> Run: ...The one entry point for continuing any durable run — a @flow or a standalone @agent — in this process or a fresh one.
answers— journaled under the interrupt's id so the pausedapprove()/ask_human()/approval-gated tool call finds its answer already there when the body re-executes:resume(run_id, answers={"publish": True, "style": "formal"}). A bare tool name (e.g."my_tool") resolves automatically to the fulltool:{name}:{call_id}interrupt id when exactly one pending interrupt matches; an ambiguous or unmatched shorthand raisesConfigErrornaming the pending ids. Answers are journaled first-write-wins, so a second, differentresume()call for the same interrupt can never silently overwrite the first.budget— overrides the run's storedBudgetfor this attempt and every later one. This is a plain last-write-wins column update, not journaled (the journal is first-write-wins, which would otherwise freeze the very first override forever). Prior attempts' real spend still counts against the new cap — you can't dodge a budget by crashing and resuming with a bigger one applied only going forward.None(the default) keeps whatever budget was already stored; there's no way to clear a budget via resume.allow_code_change(defaultFalse) — see the fingerprint check below.
Behavior by run state:
- Missing run —
ConfigError. - Standalone
@agentrun (kind == "agent") — routed to the agent loop: restores the saved conversation and continues it, or re-pauses on the next unanswered interrupt. - Flow not registered in this process — its defining module was never imported, so its
@flowdecoration never ran:ConfigErrornaming the fix (import the module before callingresume(), as in Process B above). - Already
"completed"— returns the stored output as a completedRunwithout re-executing anything. - Source changed since the run started (fingerprint mismatch —
@flowfingerprints its source via a hash ofinspect.getsource(fn)at decoration time) —ResumeMismatchError, unlessallow_code_change=True: the journal may no longer match the new code's call sequence, so this is an explicit opt-in to that risk. - Otherwise (
"running","paused", or"failed") — re-executes the flow body with the samerun_idand sametrace_id, so resumed spans join the original trace; journaled steps replay, the rest actually runs.
from composeai import resume, Budget
# answer a pending approval
run = resume(run_id, answers={"publish": True})
# raise the cap for this attempt and every later one, even though nothing paused
run = resume(run_id, budget=Budget(usd=5.0))
# explicitly accept the risk of resuming against changed flow source
run = resume(run_id, allow_code_change=True)approve(id, payload=None) -> bool and ask_human(id, question, payload=None) -> Any are the two primitives. Both look the answer up in the active run's journal first — a hit returns it (coerced to bool for approve); a miss pauses the run. Pausing is not an error: the process may exit right after it, and a paused span is marked "paused" in the trace, never "error".
import composeai as compose
@compose.flow
def publish_flow(draft: str) -> str:
if compose.approve("publish", payload={"draft": draft}):
return f"published: {draft}"
return "kept as draft"Interrupts are named, never positional — id is a string you pick, and it's what the answer is journaled under, so resuming is idempotent and order-independent regardless of which interrupt a given resume happens to answer.
The same mechanism gates tool calls: @compose.tool(requires_approval=True) pauses the agent loop mid-turn on an unanswered call, under the reserved interrupt id tool:{tool_name}:{call_id}:
@compose.tool(requires_approval=True)
def send_email(to: str, body: str) -> str:
"""Send an email.
Args:
to: Recipient address.
body: Email body.
"""
return f"sent to {to}"Resuming with answers={"send_email": True} (the bare tool name resolves to the full interrupt id, same shorthand as above) lets the call run; answers={"send_email": False} denies it — the model sees a tool result of "denied by user" (an error result) and carries on rather than the run failing outright.
Pause/resume works identically whether the pause originated directly in a flow body, inside a nested @task, or by propagating up from an @agent's tool call — and across process boundaries exactly like the crash-resume example above: run.id is all a second process needs to pick a paused run back up.
Calling a @flow from inside another active @flow's body is a nested flow call: it's journaled as one step of the enclosing flow, the same way a nested @agent call is, rather than starting a brand-new durable run row of its own.
import composeai as compose
@compose.flow
def sub_flow(x: int) -> int:
return x * 2
@compose.flow
def outer_flow(x: int) -> int:
doubled = sub_flow(x) # one journaled step of outer_flow, not a separate run
return doubled + 1
result = outer_flow(5) # 11Without this, resuming outer_flow would re-execute sub_flow's entire body — and every @task/@agent call inside it, including paid LLM calls — from scratch on every attempt, even after it had already completed once. On a journal hit, sub_flow's body never runs again at all; on a miss, it runs for real inside its own nested span (same trace, no new run_id) and its output is journaled as the outer flow's step value. A pause raised from inside sub_flow's body isn't caught at the call site — it propagates up to whichever flow (outermost or otherwise) is actually running resume()/.run(), exactly like a pause inside a nested @task would.
agents covers the @agent idiom used for editor-style steps inside a flow; composition covers pipe/aggregate/map, all usable (and journaled) inside a flow body; budgets covers Budget and cumulative spend across resume() attempts; observability covers compose trace on a paused run and the exact resume(...) call it prints for you.