> Documentation index for m4trix: https://m4trix.dev/llms.txt. Every page is available as Markdown by appending `.md` to its URL.

# Patterns: Request/Response, Fan-Out, Join, Retries

> The common shapes of an m4trix network: one agent answering, fan-out, chains, waiting for a reply, joining results and retrying tool calls. Each with a recorded run.

Source: https://m4trix.dev/docs/guides/patterns

Every pattern below is a real network, typechecked and recorded with the runtime. Step through a run to see where each event goes.

| Pattern | Use it when |
| --- | --- |
| [Request/Response](#requestresponse-single-agent) | One agent turns a request into an answer |
| [Fan-out](#fan-out) | Several agents react to the same event |
| [Agent chain](#agent-chain-sequential) | Work passes through steps in order |
| [Emit-and-await](#loopback--emit-and-await) | An agent needs another agent's answer before it continues |
| [Join](#join-multiple-inputs) | One result needs several events first |
| [Retries](#retries) | A tool call can fail and is worth trying again |

## Request/Response (Single Agent)

The simplest pattern: one agent listens for a request on `main`, does its work and emits the response on a channel that streams to the client.

Topology of this network, generated from the recorded run:

- Channel `main` (the main channel): carries every event.
- Channel `client`: carries only `answer`; leaves the network through an `sse` proxy.
- Agent `assistant`: subscribes to `main` → publishes to `client`; listens to `user-message`, emits `answer`.
- The run ends on `answer` (`endsOn`).

```ts title="request-response.ts"
export const network = AgentNetwork.setup(
  ({ mainChannel, createChannel, proxy, registerAgent, endsOn }) => {
    const client = createChannel('client').events([Answer]).proxy(proxy.sse());

    registerAgent(assistant).subscribe(mainChannel).publishTo(client);

    endsOn(Answer);
  },
);
```

`endsOn(Answer)` ends the run when the answer is published, which closes the stream. See [Hello World](https://m4trix.dev/docs/getting-started/hello-world.md) for a full example.

## Fan-Out

One event, several agents: every agent subscribed to the channel that listens for the event gets its own copy. Agents handle it concurrently; each still handles its own events one at a time.

1. `user-message` is published on `main` from outside the network; `assistant`, `auditLog` and `usageMeter` each receive a copy.
2. `assistant` receives it and emits `answer` on `client`, a channel that only carries `answer`; no agent subscribes to `client`, so `answer` leaves the network through its `sse` proxy and, as a terminal event, ends the run.
3. The runtime then publishes `m4trix:run.completed` on every channel (`main` and `client`) to close the run.

```ts title="fan-out.ts"
export const network = AgentNetwork.setup(
  ({ mainChannel, createChannel, proxy, registerAgent, endsOn }) => {
    const client = createChannel('client').events([Answer]).proxy(proxy.sse());

    // Three subscribers on main: each gets its own copy of every user-message.
    registerAgent(assistant).subscribe(mainChannel).publishTo(client);
    registerAgent(auditLog).subscribe(mainChannel);
    registerAgent(usageMeter).subscribe(mainChannel);

    endsOn(Answer);
  },
);
```

- **The first terminal event ends the run.** Once `answer` is published, the run completes and agents still working on it are aborted through their `signal`; their later emits are dropped. Side work that emits nothing (logging, metering) is fine here. When the run needs every branch's result, [join](#join-multiple-inputs) them and end on the joined event.
- **No order between branches.** Nothing orders one agent's work against another's. See [Concurrency model](https://m4trix.dev/docs/api-reference/agent-network.md#concurrency-model).

Use it for parallel processing: several analyzers, or logging and analytics next to the main work.

## Agent Chain (Sequential)

Events flow through a series of agents. Each agent subscribes to the channel the previous one publishes to.

1. `user-message` is published on `main` from outside the network; `planner` picks it up.
2. `planner` receives it and emits `plan` on `plans`, a channel that only carries `plan`; `executor` takes it from there.
3. `executor` receives it and emits `answer` on `client`, a channel that only carries `answer`; no agent subscribes to `client`, so `answer` leaves the network through its `sse` proxy and, as a terminal event, ends the run.
4. The runtime then publishes `m4trix:run.completed` on every channel (`main`, `plans` and `client`) to close the run.

```ts title="chain.ts"
export const network = AgentNetwork.setup(
  ({ mainChannel, createChannel, proxy, registerAgent, endsOn }) => {
    const plans = createChannel('plans').events([Plan]);
    const client = createChannel('client').events([Answer]).proxy(proxy.sse());

    registerAgent(planner).subscribe(mainChannel).publishTo(plans);
    registerAgent(executor).subscribe(plans).publishTo(client);

    endsOn(Answer);
  },
);
```

`plans` carries only `plan`, so the [wiring check](https://m4trix.dev/docs/api-reference/agent-network.md#wiring-check) catches an agent wired to the wrong step at setup. Use it for multi-step workflows such as plan → execute → respond.

## Loopback / Emit-and-Await

Use `emitAndAwait` when an agent or tool needs to emit an event and pause until a later event on the network satisfies a matcher.

```ts title="emit-and-await.ts"
export const coordinator = AgentFactory.run()
  .listensTo([UserMessage])
  .emits([Task, Answer])
  .logic(async ({ triggerEvent, emitAndAwait, emit }) => {
    // Publishes `task`, then waits for the event that answers it.
    const reply = await emitAndAwait(
      { name: 'task', payload: { query: triggerEvent.payload.text } },
      (event) => TaskResult.is(event),
      { timeout: '30 seconds' },
    );
    // The reply comes back as a plain envelope; the guard types its payload.
    if (!TaskResult.is(reply)) throw new Error('Unexpected reply');
    emit({ name: 'answer', payload: { text: `Found ${reply.payload.rows} rows` } });
  })
  .produce({});

export const worker = AgentFactory.run()
  .listensTo([Task])
  // A reply: nothing listens to it, the waiting coordinator receives it.
  .emits([TaskResult], { reply: true })
  .logic(async ({ emit }) => {
    emit({ name: 'task-result', payload: { rows: 42 } });
  })
  .produce({});
```

1. `user-message` is published on `main` from outside the network; `coordinator` picks it up.
2. `coordinator` receives it and emits `task` on `work`, a channel declared for `task` and `task-result`; `worker` takes it from there.
3. `worker` receives it and emits `task-result` on `work`; `coordinator`, waiting in `emitAndAwait`, receives it as its reply (same `meta.correlationId`).
4. In response to `user-message`, `coordinator` emits `answer` on `client`, a channel that only carries `answer`; no agent subscribes to `client`, so `answer` leaves the network through its `sse` proxy and, as a terminal event, ends the run.
5. The runtime then publishes `m4trix:run.completed` on every channel (`main`, `work` and `client`) to close the run.

Each call is scoped with a generated `meta.correlationId`, and the matcher only runs for events that carry that same correlation id. Since emitted events copy the triggering event meta, a worker agent that handles the request and emits a reply will automatically echo the correlation id.

The reply must be produced by a different subscriber while the caller is waiting. If an agent emits a request and waits for a reply that only the same blocked invocation can produce, the wait will time out.

Declare the replying agent's event with `.emits([TaskResult], { reply: true })`: nothing listens to it (the waiting agent receives it as its reply), and the option tells the [wiring check](https://m4trix.dev/docs/api-reference/agent-network.md#wiring-check) so. The reply comes back as a plain envelope; narrow it with the event's `.is()` guard before reading its payload. A wait that times out rejects with `PublishAndAwaitTimeoutError`.

Waits are held in memory by the waiting invocation. For waits on a human that can take minutes to days (approvals, reviews, deploys), end the run instead and continue in a new run of the same context: see [Long human waits](https://m4trix.dev/docs/guides/deployment.md#long-human-waits).

## Join (Multiple Inputs)

To combine several events into one, register an `EventAggregator` that listens to all of them and emits once the run's history has every part.

```ts title="join.ts"
// The run's events up to this trigger. Of the two invocations, only the one for whichever
// draft arrives second sees both, so the report is emitted exactly once.
async function draftsUpTo(history: EventHistory, trigger: { meta: { eventId?: string } }) {
  const events = await history.run();
  return events.slice(
    0,
    events.findIndex((event) => event.meta.eventId === trigger.meta.eventId) + 1,
  );
}

export const editor = EventAggregator.listensTo([Finding, Critique])
  .emits([Report])
  .emitWhen(async ({ history, triggerEvent }) => {
    const drafts = await draftsUpTo(history, triggerEvent);
    return drafts.some(Finding.is) && drafts.some(Critique.is);
  })
  .mapToEmit(async ({ history, triggerEvent, emit }) => {
    const drafts = await draftsUpTo(history, triggerEvent);
    const finding = drafts.find(Finding.is);
    const critique = drafts.find(Critique.is);
    emit({
      name: 'report',
      payload: { text: `${finding?.payload.text} (${critique?.payload.text})` },
    });
  });

export const network = AgentNetwork.setup(
  ({ mainChannel, createChannel, proxy, registerAgent, registerAggregator, endsOn }) => {
    const drafts = createChannel('drafts').events([Finding, Critique]);
    const client = createChannel('client').events([Report]).proxy(proxy.sse());

    registerAgent(researcher).subscribe(mainChannel).publishTo(drafts);
    registerAgent(critic).subscribe(mainChannel).publishTo(drafts);
    registerAggregator(editor).subscribe(drafts).publishTo(client);

    endsOn(Report);
  },
);
```

1. `user-message` is published on `main` from outside the network; `researcher` and `critic` each receive a copy.
2. `researcher` receives it and emits `finding` on `drafts`, a channel declared for `finding` and `critique`; `editor` picks it up.
3. In response to `user-message`, `critic` emits `critique` on `drafts`; `editor` takes it from there.
4. `editor` receives it and emits `report` on `client`, a channel that only carries `report`; no agent subscribes to `client`, so `report` leaves the network through its `sse` proxy and, as a terminal event, ends the run.
5. The runtime then publishes `m4trix:run.completed` on every channel (`main`, `drafts` and `client`) to close the run.

- **Read history, don't keep state.** `history.run()` returns the run's durable events, and every event is stored before it is delivered. A module-level `Map` would leak between runs and break once runs are spread over several processes.
- **Look only up to the trigger.** The aggregator handles `finding` and `critique` one at a time, but by the time it handles the first, both may be stored. Cutting history at the trigger means only the invocation for whichever part arrives second sees both, so `report` is emitted exactly once.
- **Durable events only.** [Transient](https://m4trix.dev/docs/api-reference/agent-network-event.md#transient-events) events never reach history, so don't join on them.

`emitWhen` decides, `mapToEmit` builds the event. The same works with a plain agent: `listensTo([Finding, Critique])` and the same history check in its logic.

## Retries

Tools retry on their own: `.retry()` reruns a call when the tool throws, with an optional backoff.

```ts title="retries.ts"
export const getQuote = Tool.of({ name: 'getQuote', description: 'Latest price for a symbol.' })
  .input(S.Struct({ symbol: S.String }))
  .output(S.Struct({ price: S.Number }))
  .timeout('2 seconds') // per attempt; a timed-out attempt is not retried
  .retry({ times: 2, backoff: { base: '200 millis', factor: 2, max: '2 seconds' } })
  .define(async ({ input, signal }) => ({ price: await fetchPrice(input.symbol, signal) }));
```

Diagram: `quoteAgent` calls `getQuote`; attempt 1 throws, the tool waits 200 ms (retry 1); attempt 2 throws, the tool waits 400 ms (retry 2); attempt 3 succeeds and returns `{ price }` to `quoteAgent`.

- **What is retried.** Only the tool throwing (`ToolExecutionError`). Invalid input or output, a timeout (`.timeout()` applies per attempt), an invalid emit and cancellation are never retried.
- **Backoff.** Retry *n* waits `base * factor^(n - 1)`, capped at `max`; without `backoff`, retries start immediately. Aborting the run cuts a wait short, and no attempt starts after it.
- **When every attempt fails.** `execute()` rejects with the last `ToolExecutionError`; `executeForModel()` hands the failure to the model instead, so it can try something else.

Agent logic has no built-in retry: loop inside `.logic()` if a step outside a tool needs one. When it finally gives up, call `fail(error)` (or let it throw) so the run ends with `m4trix:run.failed`, rather than emitting a homemade error event. See [Run Lifecycle](https://m4trix.dev/docs/api-reference/io-adapters.md#run-lifecycle).

For more, see [Networks](https://m4trix.dev/docs/concepts/networks.md) and the [AgentNetwork API](https://m4trix.dev/docs/api-reference/agent-network.md).