Parallel steps and child runs

Parallel steps

ctx.gather runs awaitables concurrently and returns their results in argument order, like asyncio.gather. Each leg is journaled, so a resumed run re-runs only the legs that hadn’t finished.

from pyrula.workflows import workflow

@workflow(name="quote")
async def quote(ctx, skus: list[str]) -> int:
    prices = await ctx.gather(*[ctx.step(f"price:{s}", fetch_price, s) for s in skus])
    return sum(prices)

Give each step a distinct id. Pass return_exceptions=True to get exceptions back as results instead of the first one raising.

Child runs

ctx.run_child starts another registered workflow as its own run and waits for its result. The child has its own journal, retries and version, and inherits the parent’s owner.

@workflow(name="ship")
async def ship(ctx, order: str) -> str:
    return await ctx.step("label", lambda: make_label(order))

@workflow(name="order")
async def order(ctx, order: str) -> dict:
    total = await ctx.step("total", lambda: price(order))
    label = await ctx.run_child("ship", {"order": order}, timeout=60)
    return {"total": total, "label": label}

• A worker serving the child’s name has to be running; the parent polls (poll_interval, default 0.5 s) until the child finishes.
• A failed child raises ChildWorkflowError in the parent. timeout bounds the wait.
• The child id is derived from the parent’s, so a replayed parent attaches to the same child instead of starting another.
• Children compose with gather: await ctx.gather(ctx.run_child("a"), ctx.run_child("b")).

To run a long workflow in bounded chunks, see ctx.continue_as_new in Replay and recovery.

Agents fan out with ctx.gather_subagents, which takes (agent, kwargs) pairs.