Long-running loops

Long autonomous processes — the kind you kick off and let run for minutes or hours — tend to reuse the same handful of skeletons. This chapter documents them as recipes built from the primitives, not as new engine features. The through-line is one discipline: a work node writes an artifact, and a separate gate or router decides what happens next. Keep those apart and your run stays debuggable, and a resume after a crash stays predictable.

Recipe: optimize / improve

Draft, score, redraft until good enough. This is exactly what iterate_until_converged does — one work function returning a Candidate, and the loop owns best-so-far, patience, and the stop decision.

flowchart LR
    START --> step
    step -- again --> step
    step -- stop --> END

Reach for this when the whole task is "keep improving one artifact against one score". When you instead need an explicit accept / revise / escalate decision inside a bigger graph, use a quality gate.

Recipe: research → plan → implement → review

The workhorse shape for a gated pipeline. Two quality gates guard the risky transitions: a plan gate makes sure the plan is sound before any work happens, and a review gate checks the result.

flowchart LR
    research --> plan --> plan_gate{plan ok?}
    plan_gate -- no --> replan --> plan
    plan_gate -- yes --> implement --> review_gate{review ok?}
    review_gate -- no --> revise --> review_gate
    review_gate -- yes --> done
    review_gate -- exhausted --> escalate

Two rules make this safe to run unattended:

• plan and review are the natural human-gate or hard-evaluator points. Escalate there, not from the middle of implement.
• implement should be narrow and idempotent, so a resume after a crash re-runs it safely. Wrap it with skip_if_done so an identical plan does not rebuild.

examples/research_plan_implement_review.py is a runnable version.

Recipe: critic / adversarial loop

Two roles, one artifact. A builder produces work; a critic returns structured findings but never edits. The builder fixes; the critic re-checks. Stop when the critic signs off or a step budget is spent.

flowchart LR
    build --> critic{findings?}
    critic -- issues --> build
    critic -- lgtm --> END

Model the critic as an evaluator returning a GateResult whose reasons carry the findings and whose feedback tells the builder what to fix. The builder reads state["gate_results"][...]["feedback"] on the next pass (the gate stores each verdict as a plain dict). This keeps "judge" and "fix" in separate nodes — the same discipline as every other recipe here.

Recipe: fan-out / map-reduce with a gate on the aggregate

Run several independent branches — different hypotheses, files, or test suites — then merge and gate the combined result.

flowchart TD
    START --> a & b & c
    a --> merge
    b --> merge
    c --> merge
    merge --> gate{aggregate ok?}

Two things matter here. First, reducers do the merging: give the aggregate channel an append or merge reducer so the parallel updates fold together deterministically instead of racing. Second, a super-step is all-or-nothing — if any branch raises, the whole step is discarded and no checkpoint is written, so a resume never sees a half-merged aggregate. Bound the fan-out with compile(max_node_concurrency=...) if the branches hit a shared resource.

Recipe: dual-state search with a planner and a promotion boundary

A search or optimization loop separates proposing from judging. One structured channel holds the whole search, and a single planner is the only writer of the proposals. Several workers run concurrently, each producing one candidate. A single deterministic verifier is the promotion boundary: it checks candidates, overwrites the current best whole, and emits events the router reacts to.

flowchart LR
    START --> planner --> dispatch{runnable?}
    dispatch -- fan out --> workers
    workers --> verifier
    verifier -- improved --> planner
    verifier -- done --> END

The reason this needs by_key: the one search channel holds fields that fold differently — intents and candidates and verified grow with append, events is replaced with last, incumbent is overwritten whole with last. A single merge cannot express that; by_key gives each field its own reducer, so concurrent workers appending candidates never clobber the planner's intents.

Two subtleties this recipe makes you confront, both consequences of reducers, not bugs:

• An append channel does not clear when you write [] to it — append(current, []) adds nothing. To drain one, have the consumer work off what it has not yet handled (key by id) rather than relying on a clear.
• intents accumulates across rounds, so dispatch must compute a runnable subset (proposed minus already verified) instead of fanning out the raw list. This is the frontier idea: runnable work is a function of state, not the whole backlog.

examples/dual_state_search.py is a runnable, offline version — a planner proposing strategies, workers scoring them concurrently, and a verifier promoting the best. The domain types (intents, candidates, verified, incumbent) live in the consumer; the core provides only the by_key channel they fold through.

Recipe: plan-as-data, the planner emits a task DAG

The previous recipe fans out a flat set of independent experiments. The next step up is work with dependencies — a task that cannot start until others finish. The engine graph is static, so you cannot add a node per task at runtime. Instead the planner emits the dependency graph as data, and add_plan_executor runs it by readiness: a task runs once all its prerequisites are done, independent tasks fan out concurrently, and the loop continues until nothing is runnable.

flowchart LR
    planner --> dispatch{runnable?}
    dispatch -- fan out --> slots
    slots --> collect{more?}
    collect -- yes --> dispatch
    collect -- no --> planner
    dispatch -- none --> planner

The plan is data in a by_key channel, so it is checkpointed and a resume recomputes which tasks are runnable from durable state — no node is added to the graph at runtime, and a new task is new data, not new structure.

from agentflow.prebuilt import PlanState, Task, TaskResult, add_plan_executor, plan_complete

class BuildState(PlanState):          # the plan channel comes from the mixin
    seeded: Annotated[bool, last]

async def run_step(task, state, ctx) -> TaskResult:
    return TaskResult(task.id, output=await do_work(task))

async def planner(state, ctx):
    if state.get("seeded"):
        return {}
    tasks = [
        Task("fetch", "step"),
        Task("compile", "step", prerequisites=("fetch",)),
        Task("test", "step", prerequisites=("compile",)),
        Task("lint", "step", prerequisites=("compile",)),   # runs with test
        Task("package", "step", prerequisites=("test", "lint")),
    ]
    return {"plan": {"tasks": tasks}, "seeded": True}

g = Graph(BuildState)
g.add_node("planner", planner)
add_plan_executor(g, handlers={"step": run_step}, on_idle="report", max_fanout=4)
g.add_edge(START, "planner")
g.add_edge("planner", "dispatch")

A handler is async (task, state, ctx) -> TaskResult, chosen by the task's kind from the handlers pool. The executor creates a dispatch node, a fixed pool of max_fanout run slots (so at most that many tasks run concurrently per super-step), and a collect node, wired to loop until the plan has no runnable task. on_idle is where control goes then — typically back to the planner to observe results and propose more, or to a node that ends the run.

Three properties worth stating plainly:

• The runnable frontier is derived from state every super-step, never stored. That is what makes a resume need no special path: a task mid-flight at a crash is simply re-dispatched, since its id is not yet in done_ids.
• A failed task counts as finished but blocks its dependents — they never become runnable. Use plan_stalled(state) at on_idle to tell "nothing left to do" from "stuck on a failed prerequisite or a cycle", and plan_complete(state) for the clean-finish case.
• Execution is at-least-once, like the rest of the SDK. task.id is the stable key across a resume, so pass it to IdempotentOp or a downstream dedup token to make an effect exactly-once.
• The plan is stored as plain dicts, not Task/TaskResult objects, so it is JSON-serializable and survives a durable checkpointer. A planner may append ergonomic Task(...) objects (the channel coerces them), but read results back as dicts: state["plan"]["results"][i]["output"].

examples/plan_executor_demo.py is a runnable, offline version — the five-task build DAG above, with test and lint running concurrently. This is the primitive the optimizer tutorial builds toward: a planner that emits a plan and re-optimizes it as results come in, with the engine running it durably as it goes.

Recipe: budget-aware stopping

An autonomous run needs a ceiling — a router that stops not only on quality but on cost. Keep the budget where it belongs: a count of attempts is a decision input that must survive a resume, so it lives in state; wall-clock time is non-deterministic, so keep it in RunMetrics or your own hook rather than in the checkpoint. Putting a timestamp in state means a resume would "replay" a time from the past and mislead the router.

from typing import Annotated
from agentflow import add

class ReviewState(GateState):
    attempts: Annotated[int, add]      # survives resume — safe to route on

def route(state):
    result = state["gate_results"].get("review")
    if result and result["passed"]:
        return "done"
    if state.get("attempts", 0) >= 10:   # hard budget ceiling
        return "escalate"
    return "revise"

A quality gate already enforces a per-gate max_attempts; this recipe is for a budget that spans several nodes or the whole run.

Token and cost budget

When the proposer is an LLM, each round costs tokens — and a count of attempts does not capture that a few expensive turns can blow the budget. Unlike wall-clock time, token spend is safe to keep in state: it only accumulates, so a resume continues from what was already spent rather than replaying. The budget helpers make this a mixin plus two calls.

from agentflow.prebuilt import BudgetState, record_usage, over_budget

class SearchState(BudgetState):          # adds the summed `budget` channel
    ...

async def propose(state, ctx):
    turn = await call_llm(...)           # a TurnEnd carrying .usage
    return {**result, **record_usage(turn.usage)}   # fold its tokens into budget

def route(state):
    if over_budget(state, max_tokens=100_000):   # or max_cost_usd=...
        return "stop"
    return "continue"

record_usage adds a turn's tokens (and cost_usd, when a backend or you supply pricing) into the budget channel; over_budget is the router's predicate. The stop is a routing decision at the next super-step boundary, so a run can overshoot by at most the turn in flight — the deliberate trade for a budget that is durable and needs no engine support. A backend that reports no token counts (usage is None) contributes zero, so the heuristic/offline path simply never trips the cap. For pure observation without stopping, attach a TokenMeter hook, which sums the same usage but cannot (by the hook contract) halt the run.

Recipe: idempotency, skip work already done

Long loops re-run the same steps — after a crash, on resume, or on another iteration. Expensive, deterministic work should run once per distinct input. skip_if_done is a recipe over the Store: hash the node's input, serve a cached result if one exists under that hash, otherwise run and cache.

from agentflow import MemoryStore
from agentflow.prebuilt import artifact_key, skip_if_done

store = MemoryStore()   # or PostgresStore for a durable, shared cache

async def expensive(state, ctx):
    return {"result": await do_costly_thing(state["input"])}

node = skip_if_done(
    expensive, store,
    namespace=("cache", "expensive"),
    key_from=lambda s: artifact_key(s["input"]),
)
g.add_node("expensive", node)

artifact_key is a stable SHA-256 of any JSON-serializable value (dict order does not matter), so it keys by what the input is rather than when it ran. Because the cache is a Store, a PostgresStore shares it across processes and a ttl expires stale entries. Only wrap deterministic work whose result depends solely on the hashed input — a node with side effects you always want is a poor fit.

Side effects and delivery semantics

There is a boundary worth stating plainly, because it is easy to misread "atomic super-step" as more than it is. A super-step is atomic over state: if any node in the step raises, the step's state updates are discarded and no checkpoint is written. That transaction covers the checkpointer and nothing else. A side effect a node already performed — a file written, an HTTP POST sent, a row inserted in another system — is not rolled back, because the engine has no transaction over it. On a resume the node runs again and the effect can happen twice.

This is not a defect to fix. It is the delivery contract of every durable executor that does not wrap your external systems in a distributed transaction: checkpoints give at-least-once execution; exactly-once holds only if the effect itself is idempotent. There is an irreducible window between "the effect happened" and "the checkpoint recorded it" — a crash in that gap always risks a repeat.

So the fix lives at the effect, and AgentFlow gives you IdempotentOp to make it explicit. Unlike skip_if_done, which wraps a whole node to avoid recomputing harmless work, IdempotentOp guards one specific effect inside a node with a claim → run → commit marker in a Store:

from agentflow.prebuilt import IdempotentOp, effect_key

op = IdempotentOp(store, ("effects", "send_email"))

async def notify(state, ctx):
    key = effect_key("send_email", state["message_id"])

    async def send():
        # pass the SAME key downstream as the provider's idempotency token
        return await email_api.send(state["to"], state["body"], idempotency_key=key)

    receipt = await op.run(key, send, on_incomplete="error")
    return {"receipt": receipt}

The state machine per key is held in one store entry so a plain get reads it in one shot:

Prior state	op.run does
absent	write in_flight, run the effect, write done(result), return it
done	return the stored result — the effect does not run again
in_flight	a prior attempt did not reach done; dispatch on on_incomplete

on_incomplete is the honest part: the SDK cannot know whether an interrupted effect actually landed, so you choose per effect.

on_incomplete	Meaning	Use when
"error" (default)	raise IncompleteEffectError	the effect is not safe to repeat — let a human or gate decide
"rerun"	run the effect again	the effect is genuinely idempotent
"skip"	assume it completed, return None	a duplicate is worse than a miss

"rerun" carries a concurrency caveat: it overwrites the in_flight marker unconditionally, so if that marker belongs to a winner that is still running (not a dead attempt), the rerun executes concurrently with it — two simultaneous runs. The atomic claim guarantees a single runner only for the first attempt; "rerun"'s safety rests entirely on the effect being idempotent and deduplicated downstream by the key, not on the claim.

A done marker only appears when the effect returned cleanly, so this reliably catches a hard crash mid-effect. But an exception is trickier — and it is a second, separate window.

When fn raises, the effect may still have happened: a TimeoutError reading the response of a POST that already went through is an exception after delivery. on_error decides what the marker is left as:

on_error	Meaning	Use when
"keep" (default)	leave in_flight so the next attempt hits on_incomplete	the exception does not prove the effect was skipped (the safe default)
"release"	drop the marker; the retry re-runs the effect	the exception reliably means "not done" (e.g. a connect error before any request left)

The consequence to internalize: on_incomplete="error" protects the crash window, but with on_error="release" it does not protect the lost-response window — the marker is already gone, so the retry simply runs the effect again.

The claim itself is atomic: it uses Store.put(..., if_absent=True), an atomic conditional create. Two workers racing the same key give exactly one winner (who runs the effect) and one StoreConflict (who re-reads and defers to the winner's marker), so the "both saw absent and both ran" race is closed — within a process for MemoryStore, and across processes for PostgresStore via a single conditional INSERT.

That closes the duplicate-starter race, but not the two windows above: the winner can still crash or lose its response inside the effect. So IdempotentOp narrows the duplicate window and makes every ambiguous case an explicit decision — it does not, and cannot, guarantee exactly-once at the orchestrator layer alone. What finally collapses duplicates is idempotency at the effect itself; use the key IdempotentOp gives you as the token that carries it there:

Effect	Idempotency technique
HTTP API	send an idempotency key / dedup token the server honors
File write	write a temp file, then atomic rename to a stable destination path; the retry must produce the same path and the same content for the replace to be idempotent
Database	INSERT ... ON CONFLICT DO NOTHING, or a unique constraint
Message/queue	a dedup id the broker or consumer deduplicates on

Reach for skip_if_done to save work; reach for IdempotentOp to protect an effect. examples/idempotent_effect.py is a runnable walk-through: an effect whose response is lost, the retry that IdempotentOp stops, and a downstream store that deduplicates on the same key.

Recipe: watch and respond

Some loops wait on something outside the graph — a pull request, a CI build, a deployment, a ticket. You don't want to burn model turns (or hold a worker) while nothing is happening. A watch node polls the external system and parks between polls: it calls ctx.wait(wake_at), which suspends the run until the time passes, then polls again.

flowchart LR
    START --> wait{watch}
    wait -- idle --> wait
    wait -- activity --> respond
    respond --> wait
    wait -- terminal --> END

Build it from watch_node and a conditional edge. The watcher reports one of three outcomes per poll — idle (park and poll again), activity (something to react to), terminal (final state, exit):

from agentflow.prebuilt import CommandWatcher, watch_node, route_watch

g.add_node("wait", watch_node(CommandWatcher("gh pr view $PR --json state -q .state"),
                              poll_interval=60))
g.add_node("respond", respond)   # reads state["watch_result"]["payload"]
g.add_conditional_edges("wait", route_watch, {"activity": "respond", "terminal": END})
g.add_edge("respond", "wait")

CommandWatcher runs any program that prints the small contract JSON ({"outcome", "payload", "cursor"}), so GitHub, CI, or a ticket system is just your script — no provider code lives in the SDK. examples/watch_until_done.py is a runnable, offline version.

Parked vs sleeping — the honest difference

ctx.wait is a timer-based sibling of the human-in-the-loop interrupt: it suspends the run and records when to wake. What that costs depends on how you run the graph:

• Through the control plane: a parked run is WAITING in the queue with a wake_at. The worker is released — zero cost while parked — and the run is re-claimed only once wake_at passes. It survives a process restart. This is the real "parked without cost".
• Inline, via app.run_until_done(...): there is no worker to release, so the method simply sleeps until wake_at and resumes. It works and is handy for development, but the process must stay alive for the whole wait. It is not free the way the control plane is.

Two limits to keep honest: polling is not push — an event is seen at the next poll, so poll_interval trades freshness for cost — and in the control plane the wake granularity is bounded by how often workers claim. A watcher's cursor is persisted across the park (it rides the wait checkpoint), so a watcher must resume from its cursor, not from in-memory state.

The one rule behind all of these

Do not mix "the work" and "the decision to continue" in one node. The work node writes an artifact; a gate or router reads state and decides. That separation is what makes each super-step atomic, each checkpoint meaningful, and each resume predictable — the properties a run needs to survive being left alone for an hour.

Next: the quality gates chapter for the gate contract in full, or the Q&A agent tutorial for an end-to-end agent.