Parallel steps and child runs

Parallel steps

ctx.gather runs awaitables concurrently, results in argument order. Each leg is journaled, so a resumed run re-runs only unfinished legs.

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. Two ctx.step calls with one id that run at the same time outside ctx.gather (asyncio.gather, a task group, create_task) raise ConcurrentStepIdError: they would share a journal key. If the body catches the error, the run still fails with it at its next suspension (a sleep, signal wait, interrupt, pause or continue_as_new) or when the body returns, because a replay cannot be relied on to overlap the two calls the same way; steps after the caught refusal run live until then. return_exceptions=True returns exceptions as results instead of raising the first.

Direct calls (an agent’s tool, sub-agent and model calls, and the drop-in provider adapters) take no step id, so a resumed run matches each to its record by its inputs:

• A replayed call never receives a result or failure that was recorded for different inputs.
• A direct tool call that raised is journaled as its failure, so a body that caught it meets the same failure on replay and the tool is not run again (the failure comes back as DirectToolCallError around the tool’s own error). Only a failure that a resume in any process raises again as it was is journaled: an error of a class from Python’s builtins module or a public pyrula module (ValueError, RuntimeError, KeyError, TypeError, TimeoutError) whose args and attributes come back with the same types, with no __cause__, and for ImportError, AttributeError and NameError the same name, path and obj. Its traceback and implicit __context__ (an error raised inside an except block) are not kept. Every other tool failure is run again on replay: an exception class from the rest of the standard library (json.JSONDecodeError, subprocess.CalledProcessError, socket.gaierror, ssl.SSLError), an app’s own or a library’s error (an httpx or SDK error), an exception group, an error carrying a cause (a tool timeout), and one with an enum, tuple or int-keyed attribute. A failure that is the last thing journaled before the run ended is made again when the run is retried; one the body caught and re-raised after other journaled calls is served again on every requeue, as a failed ctx.step is. A call with an id= journals no failure, and a refusal (a cancel, a pause, a case stop, a blocked write) is never one. A tool called from a plain thread or through Agent.stream() binds no run and journals nothing. The failure of a sub-agent call, of a tool called inside a sub-agent, and of a model call made with ctx.llm is not journaled: a resume runs it again. A sub-agent’s tool calls count against the same name-and-index slots as the body’s, so a sub-agent that takes a different path on a re-run can be served a call the body made or refuse the run as changed input (replay_input_changed).
• A body that runs its calls one after another replays call for call, and a call whose inputs changed refuses before it runs. A body that now makes the same calls in a different order replays, and each call gets the result of its own inputs.
• Inside ctx.gather, each leg counts its own calls, so any interleaving and any resume point replays.
• Under asyncio.gather, a task group or parallel graph nodes, a region whose calls all finished replays whatever order its calls start in. Calls of one name with identical inputs are interchangeable: they get the recorded results in recorded order, in the order they start on replay. A resume that lands inside a region, with some calls unfinished, may instead refuse with REPLAY_INPUT_CHANGED rather than guess which call a record belongs to. That happens when a call was unfinished at the resume point because the process died or the call was cut off (see “Waiting beside other tasks” below); such a call runs again. When the result matters, a sampled model reply say, run the calls with ctx.gather or give each an id=.
• A tool call held for approval keeps the index it was asked for. The question records which call it holds, so a call that was denied, or answered and not yet started, gets its index back whatever order the calls start in on replay, and a sibling recorded after it is served its own record (a denied call no longer leaves an empty index for another call to run at). A call whose inputs differ from a denied approval’s at the same position is refused with REPLAY_INPUT_CHANGED before that call runs, and nobody is asked for it. A question that was not denied (still open, or approved with the call not started) only keeps its index for its call: a call with no record that starts first takes the next free index. So a code change at an approved position whose call has not started, made while the run was parked, no longer refuses up front: the changed call runs, and the run then ends in a version-skew error because the approved call never came back. Questions are keyed by tool and index (tool_approval:refund#0), so two different gated tools never share a key. A journal written before a question recorded its call has no index to keep: its empty index is run live, as before.

An id= names the call’s journal key whatever the scope, so give each call its own: a second call of one name with an id= that an earlier call is still running under (in a ctx.gather leg as well) raises ConcurrentCallIdError (from pyrula.contracts.errors, a ConcurrentStepIdError) and ends the run even if the body catches it. Using the same id= again after the first call has finished is not refused while the run is live, but an id= should still name one call: a resume replays by name and id= and keeps only the last record, so a reused id with different inputs refuses the resume with REPLAY_INPUT_CHANGED.

Bounding fan-out

Every leg starts at once by default. max_concurrency= caps legs in flight:

@workflow(name="summarize_documents")
async def summarize_documents(ctx, docs: list[str]) -> list[str]:
    return await ctx.gather(
        *[ctx.step(f"summarize:{d}", summarize, d) for d in docs],
        max_concurrency=8,
    )

Legs start in argument order. The bound only delays starts; result order, step ids and replay don’t change. The first error cancels the rest, queued legs included. A leg holds its slot across all its steps, so the cap counts legs. Int >= 1 or None. (par_map is a separate process-based helper for CPU-bound work, sized with workers=.)

A leg can’t wait. ctx.sleep, ctx.sleep_until, ctx.wait_for_signal and ctx.interrupt inside a leg raise SuspendInGatherLeg (even with return_exceptions=True): a run has one suspension slot. Wait before or after the gather. ctx.gather is the form to use when legs run side effects beside a question. Tasks you start with plain asyncio (asyncio.gather, a task group, create_task) may wait, with the limits below.

Waiting beside other tasks

The first wait to open holds the run’s one slot, and a task that reaches a wait or a sleep meanwhile waits for the first to finish opening. If the first wait stands, the task ends the attempt without writing anything and makes its call on the attempt after the wait is answered. If it ended at once (a signal already buffered, a deadline already past), the task takes the slot in the same attempt.

A call another task has already started is not cut off when the run parks. The attempt waits for every ctx.step, tool call and model call (ctx.llm, with the tools it runs) that a task has in flight, up to the attempt’s timeout, so each records its outcome and runs once, not again after the answer. The question is saved before the wait, so it can be answered meanwhile, and journaled again behind those outcomes, so it is still the last thing the run recorded. A task that reaches a new call after the run has decided to park (a step, a tool call, a model call, ctx.emit, ctx.now(), ctx.uuid(), ctx.random(), ctx.checkpoint(), ctx.patched(), ctx.run_child or ctx.send_signal) does not start it: the call is made on the attempt after the answer. A call that is still running at the timeout is cancelled with a warning that names its step id, and it runs again when the run resumes. A call that only finishes once the question is answered therefore holds the attempt, and its worker slot, until the timeout.

Limits:

• A task group cancels its other tasks when one of them ends the attempt, so a call it had in flight is cut off before it can record an outcome and runs again after the answer. Calls that must run once belong in ctx.gather or asyncio.gather.
• A call made from a thread (ctx.to_thread, a tool body that calls back into the run) is not waited for.
• A model stream you stop reading before its last event holds the attempt until the timeout only while it is stopped inside the model call and you keep a reference to it; read it to the end, aclose() it, or let it go, and the attempt stops waiting.
• A sub-agent called as a tool and in flight when the run parks stops at its next model call and resumes after the answer. Its finished calls replay, so no provider call repeats, but its outcome is recorded after the answer, not before the question.
• These are neither held nor waited for, so one in another task can be journaled behind the question: the pending update a wait applies before it opens, ctx.deprecate_patch, ctx.stream and ctx.commit in a sibling task, ctx.mcp_tools(), the writes a case replay serves, and a ctx.run_child or ctx.send_signal that is already running when the run parks.
• The wait is part of the worker’s attempt. A body driven through Agent.stream() or AgentRunner directly, without a worker, does not wait.

Blocking calls in threads

await ctx.to_thread(fn, *args, **kwargs) runs a blocking function on a worker thread, and await ctx.map_threads(fn, items, max_workers=8) runs it once per item and returns the results in item order. The thread carries the run’s context, which a plain threading.Thread, loop.run_in_executor or ThreadPoolExecutor does not: a model call that the run journals reaches the run’s loop from the thread, and one made without the context runs live again on every resume. When a pyrula.agents.compat call finds no run while a run is bound in the same process, pyrula logs a warning naming the call site, once per site, that points at these two helpers. The sync pyrula.agents.compat.OpenAI and Anthropic clients journal from these threads; see Sync code and threads.

answers = await ctx.map_threads(ask, questions, max_workers=4)

Each call runs under its own replay scope (thread:<n>, map:<n>:<i>), so concurrent threads never race one call index. Start to_thread calls in a deterministic order. Inside a ctx.step or tool body a call is a plain thread and takes no number, because a replayed step does not run its body.

max_workers is an int >= 1 and defaults to 8: the worker threads are one pool shared by every run in the process, and an unbounded map would queue all its items ahead of other runs’ sync steps. Raise it for more fan-out (the pool holds min(32, CPUs + 4) threads); None removes the bound. When an item raises, items still queued never start. Items already running cannot be stopped, so they are waited for and the error of the lowest-indexed failed item propagates after the slowest of them finishes. Cancelling the caller returns at once, and a thread still running then finishes on its own and keeps journaling its calls; if the run’s attempt has ended its next journaled call raises pyrula.contracts.errors.RunAttemptEnded, a stop that is not an Exception; let it propagate. A thread that waits on a blocked run loop past the liveness bound raises pyrula.contracts.errors.LoopUnresponsive.

Child runs

ctx.run_child starts another registered workflow as its own run and waits for the 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 must be running. The parent polls (poll_interval, 0.5 s) until it finishes.
• Failed child: ChildWorkflowError. timeout bounds the wait.
• The child id derives from the parent’s, so a replayed parent re-attaches to the same child.
• Composes with gather: await ctx.gather(ctx.run_child("a"), ctx.run_child("b")).

run_child holds the parent’s worker while it polls; it doesn’t suspend the parent.

What happens to a child when its parent ends

A child is open from when the parent journals it until the parent collects its result. If the parent ends meanwhile (completes, fails, cancelled, terminated), the worker applies parent_close before writing the parent’s terminal event:

parent_close	The open child
"cancel" (default)	Is asked to cancel, so its cleanup runs. Needs a worker.
"terminate"	Is ended at once, without its worker or cleanup.
"abandon"	Keeps running.

await ctx.run_child("notify", {"order": order}, parent_close="abandon")

Applies however the parent ends, so a failed gather leg doesn’t strand its siblings. A terminated parent also terminates its cancel children; abandon children are never touched. Grandchildren close the same way. The child’s terminal event carries detail.cause (parent run id, workflow name, how it ended) and requested_by="parent".

Children close before the parent’s terminal is written, so a crash in between just repeats the close on replay. Runs from ctx.spawn_workflow (agents) aren’t children and are never closed this way. A run the store ends itself (requeue budget exhausted) has no worker to close its children.

A failing Kafka-consumed workflow is requeued under the same run id, so its children stay open across retries. They close when the consumer gives up the record (on_error dlq or skip after retries) and the parent’s journal has ended. With on_error="fail" nothing closes.

Long workflows in bounded chunks: ctx.continue_as_new, see Replay and recovery. Agents fan out with ctx.gather_subagents, which takes (agent, kwargs) pairs.