stagehand

Declare the DAG; the engine runs it

You declare what work needs doing and how the pieces depend on each other; the engine figures out how to run it. A node like map stamps out one task per item and wires its dependencies, so the scheduler runs everything that's ready — eval(cell_0) starts the moment train(cell_0) finishes, while train(cell_2) is still going. No barriers except where a node needs its whole upstream collection — that's reduce.

flow = Flow("runs", title="my sweep", concurrency=8)
trained = flow.map("train", cells, train_one)            # one task per cell
healthy = flow.filter("gate", trained, is_healthy)       # survivors stream on
evals   = flow.map("eval", healthy, eval_one)            # eval_i waits on train_i only
best    = flow.reduce("pick", evals, choose_best)        # the *only* barrier

async with live_dashboard(flow.runs_dir, title="my sweep"):
    await flow.run(stop_when=lambda s: s.done >= 100)   # early-exit optional

Fan out, retry, expand

best_of and with_retry are node policies — unit-fns you drop into map: best_of runs N attempts per item and keeps the best (objective score or async judge); with_retry re-runs an item that flunks its check — or raises — feeding feedback back in until it passes. expand handles dynamic fan-out: one upstream result becomes a data-dependent number of downstream tasks.

# N attempts per item, keep the best — shows up as an ordinary node
best = flow.map("solve", units, best_of(solve, n=4, score=lambda r: r["reward"]))

# retry each item with feedback until it parses
fixed = flow.map("fix", units, with_retry(solve, check=parses, max_attempts=3))

# train emits K checkpoints (K unknown until runtime) -> eval each
ckpts = flow.expand("ckpts", trained, lambda r: r["checkpoints"])
evals = flow.map("eval", ckpts, eval_one)

A handle is a placeholder, so you can't if on its value while building — put result-dependent control flow inside a node fn (a normal coroutine: plain await / if; raise to prune), or use filter / expand.

Compile the graph

Steps declare their I/O with plain type hints. flow.check() reads them and verifies the graph can run start to end — every dependency exists, no cycles, and each edge's producer type fits the consumer — before you spend any compute.

async def train(cfg: Config) -> Model: ...
async def evaluate(m: Model) -> Eval: ...

flow.check()                 # raises on a malformed / ill-typed graph
await flow.run(check=True)   # check first, then run

# flow check failed:
#   - map 'eval' <- 'train': expects str, got Model

Handle[T] is generic, so .result is typed in your editor. Gradual and pragmatic — unannotated is Any, subclasses / list[T] / Optional are handled. A linter, not a type system.

Coding agents as steps

An agent is just a step. agent(flow, prompt, …) spawns a headless coding agent and returns a structured AgentOutcome, so it composes with everything — best_of for best-of-N agents (judge picks the best patch), with_retry for retry-with-feedback, reduce to merge.

flow  = Flow("runs", concurrency=4)
patch = agent(flow, "fix the failing test in foo.py", isolation="worktree")
best  = flow.map("solve", issues, best_of(solve_agent, n=4, judge=pick_best))  # best-of-N agents
await flow.run()
print(patch.result.diff)

isolation="worktree" runs each agent in its own throwaway git worktree and captures the diff. Backends sit behind a seam so the core stays dependency-free: subprocess_backend (zero-dep claude -p) by default, or the recommended flightdeck_backend() for live stream capture to the dashboard (flightdeck imported lazily — an optional integration, not a dependency).

What you watch

Every unit of work writes a small JSON file as it runs; the dashboard reads that tree and renders one auto-refreshing page. A unit that flunks a gate goes red — the broken-harness failure mode shows up instead of silently poisoning the results.

Layers, each usable on its own

Quickstart

# add it
uv add git+https://github.com/dtch1997/stagehand

# or clone and run the worked example (fake compute, no network)
git clone https://github.com/dtch1997/stagehand && cd stagehand
make example     # runs the worked sweep → writes runs/status.html
make test        # unit tests

The single primitive underneath it all:

with monitor("cell_s0", total=256, path="runs/cell_s0/train.progress.json",
             parent="sweep") as m:
    for batch in batches:
        m.update(loss=train_step(batch))   # advance the ticker, record fields