# Parallel steps and child runs

> Run steps concurrently with ctx.gather, and start other workflows with ctx.run_child.

Source: https://docs.pyrula.io/workflows/guides/parallel-and-children/

## Parallel steps

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

```python
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:

```python
@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](https://docs.pyrula.io/agents/integrations/durable-vs-not.md#sync-code-and-threads).

```python
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.

```python
@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. |

```python
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](https://docs.pyrula.io/workflows/concepts/replay-and-recovery.md). Agents fan out with
[`ctx.gather_subagents`](https://docs.pyrula.io/agents/authoring/context.md), which takes `(agent, kwargs)` pairs.