# Run steps in parallel

> Run several tools on the same input at the same time with a fanout step, and choose how their results come back.

This guide runs several tools on the same input at the same time, with a fanout step, and chooses how their results come back.

A `fanout` step runs several tools at once, on the same input, and collects their outputs into one.

## Before you start

- **A pack with the tools** from [Write a flow](https://docs.kindgi.com/v0.1.8/guides/flows/write-a-flow.md), such as `acme.get-order`, on a runtime such as `kindgi dev`, with Kindgi 0.1's CLI.

## Fan out to several tools

Ask your coding agent: "I want a flow that scores an order's risk and quotes its shipping at the same time." Kindgi's skills in your pack tell it how ([How it knows Kindgi](https://docs.kindgi.com/v0.1.8/start/coding-agents.md)). For reference, the tools and the flow look like this. This guide calls the flow `acme.assess-order`, and it uses two new read-only tools:

- **TypeScript**

  `tools/score-risk/index.ts`

  ```ts
  import { defineTool } from '@kindgi/sdk/define';
  import type { ToolId } from '@kindgi/sdk/types';
  import { z } from 'zod';

  const defined = defineTool({
    id: 'acme.score-risk' as ToolId,
    description: 'Scores the fraud risk of an order.',
    version: '0.1.0',
    input: z.object({ total: z.number() }),
    output: z.object({ level: z.enum(['low', 'high']) }),
    effects: [],
    mutating: false,
    handler: async ({ total }) => ({ level: total > 1000 ? ('high' as const) : ('low' as const) }),
  });

  if (defined.kind === 'err') throw new Error(defined.error.message);

  export default defined.value;
  ```

  `tools/quote-shipping/index.ts`

  ```ts
  import { defineTool } from '@kindgi/sdk/define';
  import type { ToolId } from '@kindgi/sdk/types';
  import { z } from 'zod';

  const defined = defineTool({
    id: 'acme.quote-shipping' as ToolId,
    description: 'Quotes how many days an order takes to ship.',
    version: '0.1.0',
    input: z.object({ items: z.array(z.object({ sku: z.string(), quantity: z.number() })) }),
    output: z.object({ days: z.number().int() }),
    effects: [],
    mutating: false,
    handler: async ({ items }) => ({ days: 2 + items.length }),
  });

  if (defined.kind === 'err') throw new Error(defined.error.message);

  export default defined.value;
  ```

  `flows/assess-order/index.ts`

  ```ts
  import { defineFlow } from '@kindgi/sdk/define';

  const defined = defineFlow({
    id: 'acme.assess-order',
    version: '0.1.0',
    name: 'Assess an order',
    description: 'Scores the risk and quotes shipping for an order, at the same time.',
    nodes: [
      {
        id: 'order',
        kind: 'tool',
        ref: 'acme.get-order',
        inputMapping: { orderId: { path: 'runInput.orderId' } },
      },
      {
        id: 'assess',
        kind: 'fanout',
        convergence: 'all-succeed',
        branches: [
          {
            branchId: 'risk',
            handler: 'acme.score-risk',
            outputSchema: { type: 'object', properties: { level: { type: 'string' } }, required: ['level'] },
          },
          {
            branchId: 'shipping',
            handler: 'acme.quote-shipping',
            outputSchema: { type: 'object', properties: { days: { type: 'integer' } }, required: ['days'] },
          },
        ],
      },
    ],
    edges: [
      { id: 'e1', from: '$start', to: 'order' },
      { id: 'e2', from: 'order', to: 'assess' },
      { id: 'e3', from: 'assess', to: '$end' },
    ],
    output: {
      mapping: {
        risk: { path: 'nodeOutputs.assess.risk.level' },
        shippingDays: { path: 'nodeOutputs.assess.shipping.days' },
      },
    },
  });

  if (defined.kind === 'err') throw new Error(defined.error.message);

  export default defined.value;
  ```

- **Python**

  `tools/score_risk.py`

  ```python
  from typing import Literal

  from pydantic import BaseModel

  from kindgi import tool


  class RiskInput(BaseModel):
      total: float


  class Risk(BaseModel):
      level: Literal["low", "high"]


  @tool(id="acme.score-risk", mutating=False)
  def score_risk(input: RiskInput) -> Risk:
      """Scores the fraud risk of an order."""
      return Risk(level="high" if input.total > 1000 else "low")
  ```

  `tools/quote_shipping.py`

  ```python
  from pydantic import BaseModel

  from kindgi import tool


  class Item(BaseModel):
      sku: str
      quantity: int


  class ShippingInput(BaseModel):
      items: list[Item]


  class Shipping(BaseModel):
      days: int


  @tool(id="acme.quote-shipping", mutating=False)
  def quote_shipping(input: ShippingInput) -> Shipping:
      """Quotes how many days an order takes to ship."""
      return Shipping(days=2 + len(input.items))
  ```

  `flows/assess_order.py`

  ```python
  from kindgi import Flow

  assess_order = Flow(
      id="acme.assess-order",
      version="0.1.0",
      name="Assess an order",
      description="Scores the risk and quotes shipping for an order, at the same time.",
      nodes=[
          {
              "id": "order",
              "kind": "tool",
              "ref": "acme.get-order",
              "inputMapping": {"orderId": {"path": "runInput.orderId"}},
          },
          {
              "id": "assess",
              "kind": "fanout",
              "convergence": "all-succeed",
              "branches": [
                  {
                      "branchId": "risk",
                      "handler": "acme.score-risk",
                      "outputSchema": {
                          "type": "object",
                          "properties": {"level": {"type": "string"}},
                          "required": ["level"],
                      },
                  },
                  {
                      "branchId": "shipping",
                      "handler": "acme.quote-shipping",
                      "outputSchema": {
                          "type": "object",
                          "properties": {"days": {"type": "integer"}},
                          "required": ["days"],
                      },
                  },
              ],
          },
      ],
      edges=[
          {"id": "e1", "from": "$start", "to": "order"},
          {"id": "e2", "from": "order", "to": "assess"},
          {"id": "e3", "from": "assess", "to": "$end"},
      ],
      output={
          "mapping": {
              "risk": {"path": "nodeOutputs.assess.risk.level"},
              "shippingDays": {"path": "nodeOutputs.assess.shipping.days"},
          }
      },
  )
  ```

- **Java**

  `src/main/java/acme/tools/ScoreRisk.java`

  ```java
  package acme.tools;

  import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.kindgi.pack.Tool;

  public final class ScoreRisk {
    // A branch gets the whole order: its other fields are left out.
    @JsonIgnoreProperties(ignoreUnknown = true)
    public record Input(double total) {}

    public enum Level {
      @JsonProperty("low") LOW,
      @JsonProperty("high") HIGH
    }

    public record Output(Level level) {}

    public static final Tool<Input, Output> TOOL = Tool.define("acme.score-risk")
        .description("Scores the fraud risk of an order.")
        .version("0.1.0")
        .input(Input.class)
        .output(Output.class)
        .mutating(false)
        .handler((input, ctx) -> new Output(input.total() > 1000 ? Level.HIGH : Level.LOW));

    private ScoreRisk() {}
  }
  ```

  `src/main/java/acme/tools/QuoteShipping.java`

  ```java
  package acme.tools;

  import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
  import com.kindgi.pack.Tool;
  import java.util.List;

  public final class QuoteShipping {
    public record Item(String sku, double quantity) {}

    // A branch gets the whole order: its other fields are left out.
    @JsonIgnoreProperties(ignoreUnknown = true)
    public record Input(List<Item> items) {}

    public record Output(int days) {}

    public static final Tool<Input, Output> TOOL = Tool.define("acme.quote-shipping")
        .description("Quotes how many days an order takes to ship.")
        .version("0.1.0")
        .input(Input.class)
        .output(Output.class)
        .mutating(false)
        .handler((input, ctx) -> new Output(2 + input.items().size()));

    private QuoteShipping() {}
  }
  ```

  `src/main/java/acme/flows/AssessOrder.java`

  ```java
  package acme.flows;

  import com.kindgi.pack.Flow;
  import java.util.List;
  import java.util.Map;

  public final class AssessOrder {
    public static final Flow FLOW = Flow.define("acme.assess-order")
        .version("0.1.0")
        .set("name", "Assess an order")
        .set("description", "Scores the risk and quotes shipping for an order, at the same time.")
        .node(Map.of("id", "order", "kind", "tool", "ref", "acme.get-order",
            "inputMapping", Map.of("orderId", Map.of("path", "runInput.orderId"))))
        .node(Map.of(
            "id", "assess",
            "kind", "fanout",
            "convergence", "all-succeed",
            "branches", List.of(
                Map.of("branchId", "risk", "handler", "acme.score-risk",
                    "outputSchema", Map.of("type", "object",
                        "properties", Map.of("level", Map.of("type", "string")), "required", List.of("level"))),
                Map.of("branchId", "shipping", "handler", "acme.quote-shipping",
                    "outputSchema", Map.of("type", "object",
                        "properties", Map.of("days", Map.of("type", "integer")), "required", List.of("days"))))))
        .edge("e1", "$start", "order")
        .edge("e2", "order", "assess")
        .edge("e3", "assess", "$end")
        .set("output", Map.of("mapping", Map.of(
            "risk", Map.of("path", "nodeOutputs.assess.risk.level"),
            "shippingDays", Map.of("path", "nodeOutputs.assess.shipping.days"))))
        .build();

    private AssessOrder() {}
  }
  ```

  A branch gets the whole output of the step before it (the order). A record refuses properties it doesn't name, so each branch's input says `@JsonIgnoreProperties(ignoreUnknown = true)`; a TypeScript zod object drops them by itself.

- **Scala**

  `src/main/scala/acme/tools/ScoreRisk.scala`

  ```scala
  package acme.tools

  import com.fasterxml.jackson.annotation.JsonIgnoreProperties
  import com.kindgi.pack.scaladsl._

  object ScoreRisk {
    // A branch gets the whole order: its other fields are left out.
    @JsonIgnoreProperties(ignoreUnknown = true)
    final case class Input(total: Double)

    enum Level {
      case low, high
    }

    final case class Output(level: Level)

    val tool: Tool[Input, Output] = Tool[Input, Output]("acme.score-risk")
      .description("Scores the fraud risk of an order.")
      .version("0.1.0")
      .readOnly
      .handler((in, _) => Output(if (in.total > 1000) Level.high else Level.low))
  }
  ```

  `src/main/scala/acme/tools/QuoteShipping.scala`

  ```scala
  package acme.tools

  import com.fasterxml.jackson.annotation.JsonIgnoreProperties
  import com.kindgi.pack.scaladsl._

  object QuoteShipping {
    final case class Item(sku: String, quantity: Double)

    // A branch gets the whole order: its other fields are left out.
    @JsonIgnoreProperties(ignoreUnknown = true)
    final case class Input(items: Seq[Item])

    final case class Output(days: Int)

    val tool: Tool[Input, Output] = Tool[Input, Output]("acme.quote-shipping")
      .description("Quotes how many days an order takes to ship.")
      .version("0.1.0")
      .readOnly
      .handler((in, _) => Output(2 + in.items.size))
  }
  ```

  `src/main/scala/acme/flows/AssessOrder.scala`

  ```scala
  package acme.flows

  import com.kindgi.pack.scaladsl._

  object AssessOrder {
    val flow: Flow = Flow("acme.assess-order")
      .version("0.1.0")
      .set("name", "Assess an order")
      .set("description", "Scores the risk and quotes shipping for an order, at the same time.")
      .node(Map("id" -> "order", "kind" -> "tool", "ref" -> "acme.get-order",
        "inputMapping" -> Map("orderId" -> Map("path" -> "runInput.orderId"))))
      .node(Map(
        "id" -> "assess",
        "kind" -> "fanout",
        "convergence" -> "all-succeed",
        "branches" -> List(
          Map("branchId" -> "risk", "handler" -> "acme.score-risk",
            "outputSchema" -> Map("type" -> "object",
              "properties" -> Map("level" -> Map("type" -> "string")), "required" -> List("level"))),
          Map("branchId" -> "shipping", "handler" -> "acme.quote-shipping",
            "outputSchema" -> Map("type" -> "object",
              "properties" -> Map("days" -> Map("type" -> "integer")), "required" -> List("days"))))))
      .edge("e1", "$start", "order")
      .edge("e2", "order", "assess")
      .edge("e3", "assess", "$end")
      .set("output", Map("mapping" -> Map(
        "risk" -> Map("path" -> "nodeOutputs.assess.risk.level"),
        "shippingDays" -> Map("path" -> "nodeOutputs.assess.shipping.days"))))
      .build()
  }
  ```

  A branch gets the whole order, and a case class refuses properties it doesn't name, so each branch's input says `@JsonIgnoreProperties(ignoreUnknown = true)`.

* Each **branch** has a `branchId`, the tool it runs (`handler`), and the `outputSchema` its output must match.
* Every branch gets the fanout step's input: the output of the step before it, here the whole order. Each tool takes the fields its input declares, such as `total` and `items`, and drops the rest.
* **`convergence`** decides when the step is done and what it returns, as [Convergence](https://docs.kindgi.com/v0.1.8/guides/flows/run-steps-in-parallel.md#convergence) shows.

Run it:

- **TypeScript**

  ```sh
  pnpm exec kindgi runs start --flow=acme.assess-order --input='{"orderId":"A-200"}'
  ```

- **Python**

  ```sh
  uv run kindgi runs start --flow=acme.assess-order --input='{"orderId":"A-200"}'
  ```

- **Java**

  ```sh
  ./kindgiw runs start --flow=acme.assess-order --input='{"orderId":"A-200"}'
  ```

- **Scala**

  ```sh
  ./kindgiw runs start --flow=acme.assess-order --input='{"orderId":"A-200"}'
  ```

`--flow` is the flow, and `--input` its input. It prints the run, with both tools' results:

```json excerpt
{
  …
  "status": "completed",
  …
  "output": {
    "risk": "high",
    "shippingDays": 4
  },
  …
}
```

In the journal, which `kindgi runs journal <run-id>` prints, both branches are dispatched before either completes:

```json excerpt
{"sequence": 6, "kind": "fanout.dispatched", "nodeId": "assess", "payload": {…, "handler": "acme.score-risk", "branchId": "risk", "fanoutNodeId": "assess"}, …}
{"sequence": 7, "kind": "fanout.dispatched", "nodeId": "assess", "payload": {…, "handler": "acme.quote-shipping", "branchId": "shipping", "fanoutNodeId": "assess"}, …}
{"sequence": 8, "kind": "fanout.branch-completed", "nodeId": "assess", "payload": {"output": {"days": 4}, "branchId": "shipping", "fanoutNodeId": "assess"}, …}
{"sequence": 9, "kind": "fanout.branch-completed", "nodeId": "assess", "payload": {"output": {"level": "high"}, "branchId": "risk", "fanoutNodeId": "assess"}, …}
{"sequence": 10, "kind": "fanout.converged", "nodeId": "assess", "payload": {"output": {"risk": {"level": "high"}, "shipping": {"days": 4}}, "convergence": "all-succeed", "fanoutNodeId": "assess"}, …}
```

## Convergence

| `convergence` | Done when                                                                                    | The step's output                                                                                     |
| ------------- | -------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------- |
| `all-succeed` | every branch succeeded. The first failure fails the step and cancels the others              | `{ "<branchId>": <output>, … }`                                                                       |
| `any-succeed` | the first branch succeeds, and the others are cancelled. It fails only if every branch fails | `{ "winnerBranchId": "<branchId>", "output": <output> }`                                              |
| `settle-all`  | every branch finished, succeeded or failed. The step itself succeeds                         | `{ "<branchId>": { "status": "succeeded", "output": … } \| { "status": "failed", "error": "…" }, … }` |

With `settle-all`, a failed branch is data for the next step, not a failed run. Here a `settle-all` fanout's `stock` branch runs `acme.check-stock` with an input that doesn't fit it, next to `risk`. The step's output:

```json
{
  "risk": { "output": { "level": "high" }, "status": "succeeded" },
  "stock": {
    "error": "input-validation-failed: Input for tool \"acme.check-stock\" failed validation",
    "status": "failed"
  }
}
```

With `all-succeed`, the same branch fails the run:

```text
[branch-failure] Fanout "assess" all-succeed mode: branch "stock" failed — input-validation-failed: Input for tool "acme.check-stock" failed validation
```

## Fanout or loop

A fanout runs **different** tools on the **same** input. To run the **same** steps on each element of a list, several at a time, use a `foreach` loop with `concurrency`, as [Repeat steps in a loop](https://docs.kindgi.com/v0.1.8/guides/flows/loop.md) shows.

## Next

- [Repeat steps in a loop](https://docs.kindgi.com/v0.1.8/guides/flows/loop.md)
- [Join branches](https://docs.kindgi.com/v0.1.8/guides/flows/join-branches.md)
- [Retry a failing step](https://docs.kindgi.com/v0.1.8/guides/flows/retry-a-step.md)