Guides

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.

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	One agent turns a request into an answer
Fan-out	Several agents react to the same event
Agent chain	Work passes through steps in order
Emit-and-await	An agent needs another agent's answer before it continues
Join	One result needs several events first
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.

AgentNetwork

channelmain

channelclient

expose()

proxy.sse()

assistant

request-response.ts

export const const network: AgentNetwork<never>network = class AgentNetwork<TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never>AgentNetwork.AgentNetwork<TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never>.setup(callback: (ctx: AgentNetworkSetupContext) => void, options?: AgentNetworkSetupOptions): AgentNetwork<never>setup(
  ({ mainChannel: ConfiguredChannelmainChannel, createChannel: (name: string) => ConfiguredChannelcreateChannel, proxy: {
    sse(): ProxyDef;
    custom(kind: string, config?: unknown, direction?: ProxyDirection): ProxyDef;
}proxy, registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent, endsOn: <E extends EventDef>(event: E, when?: (envelope: EventEnvelope<E>) => boolean) => voidDeclares a terminal event: when it is published in a run (and `when` accepts it, if given),
the run completes with `m4trix:run.completed`, its in-flight agents are aborted and its
stream closes. Call it once per terminal event.endsOn }) => {
    const const client: ConfiguredChannelclient = createChannel: (name: string) => ConfiguredChannelcreateChannel('client').ConfiguredChannel.events(events: ReadonlyArray<EventDef>): ConfiguredChannelDeclares the events this channel carries. Publishing any other event to it is refused: an
agent's emit throws `EmitValidationError` (reason `channel`), an external publish resolves
`false`. Runtime `m4trix:*` events always pass. Without a declaration a channel carries
every event.events([const Answer: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>Answer]).ConfiguredChannel.proxy(...proxies: ProxyDef[]): ConfiguredChannelproxy(proxy: {
    sse(): ProxyDef;
    custom(kind: string, config?: unknown, direction?: ProxyDirection): ProxyDef;
}proxy.function sse(): ProxyDefsse());

    registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent(const assistant: Agent<unknown, {
    name: "user-message";
    meta: EventMeta;
    payload: {
        readonly text: string;
    };
}, {
    name: "answer";
    payload: {
        readonly text: string;
    };
}, never, never, never>assistant).function subscribe(channel: ConfiguredChannel): AgentBindingsubscribe(mainChannel: ConfiguredChannelmainChannel).function publishTo(channel: ConfiguredChannel): AgentBindingpublishTo(const client: ConfiguredChannelclient);

    endsOn: <AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>>(event: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>, when?: ((envelope: {
    ...;
}) => boolean) | undefined) => voidDeclares a terminal event: when it is published in a run (and `when` accepts it, if given),
the run completes with `m4trix:run.completed`, its in-flight agents are aborted and its
stream closes. Call it once per terminal event.endsOn(const Answer: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>Answer);
  },
);

endsOn(Answer) ends the run when the answer is published, which closes the stream. See Hello World 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.

AgentNetwork

channelmain

channelclient

user-message

expose()

proxy.sse()

assistant

auditLog

usageMeter

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.

Payload user-message

{
  "text": "Summarize the incident"
}

fan-out.ts

export const const network: AgentNetwork<never>network = class AgentNetwork<TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never>AgentNetwork.AgentNetwork<TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never>.setup(callback: (ctx: AgentNetworkSetupContext) => void, options?: AgentNetworkSetupOptions): AgentNetwork<never>setup(
  ({ mainChannel: ConfiguredChannelmainChannel, createChannel: (name: string) => ConfiguredChannelcreateChannel, proxy: {
    sse(): ProxyDef;
    custom(kind: string, config?: unknown, direction?: ProxyDirection): ProxyDef;
}proxy, registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent, endsOn: <E extends EventDef>(event: E, when?: (envelope: EventEnvelope<E>) => boolean) => voidDeclares a terminal event: when it is published in a run (and `when` accepts it, if given),
the run completes with `m4trix:run.completed`, its in-flight agents are aborted and its
stream closes. Call it once per terminal event.endsOn }) => {
    const const client: ConfiguredChannelclient = createChannel: (name: string) => ConfiguredChannelcreateChannel('client').ConfiguredChannel.events(events: ReadonlyArray<EventDef>): ConfiguredChannelDeclares the events this channel carries. Publishing any other event to it is refused: an
agent's emit throws `EmitValidationError` (reason `channel`), an external publish resolves
`false`. Runtime `m4trix:*` events always pass. Without a declaration a channel carries
every event.events([const Answer: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>Answer]).ConfiguredChannel.proxy(...proxies: ProxyDef[]): ConfiguredChannelproxy(proxy: {
    sse(): ProxyDef;
    custom(kind: string, config?: unknown, direction?: ProxyDirection): ProxyDef;
}proxy.function sse(): ProxyDefsse());

    // Three subscribers on main: each gets its own copy of every user-message.
    registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent(const assistant: Agent<unknown, {
    name: "user-message";
    meta: EventMeta;
    payload: {
        readonly text: string;
    };
}, {
    name: "answer";
    payload: {
        readonly text: string;
    };
}, never, never, never>assistant).function subscribe(channel: ConfiguredChannel): AgentBindingsubscribe(mainChannel: ConfiguredChannelmainChannel).function publishTo(channel: ConfiguredChannel): AgentBindingpublishTo(const client: ConfiguredChannelclient);
    registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent(const auditLog: Agent<unknown, {
    name: "user-message";
    meta: EventMeta;
    payload: {
        readonly text: string;
    };
}, never, never, never, never>auditLog).function subscribe(channel: ConfiguredChannel): AgentBindingsubscribe(mainChannel: ConfiguredChannelmainChannel);
    registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent(const usageMeter: Agent<unknown, {
    name: "user-message";
    meta: EventMeta;
    payload: {
        readonly text: string;
    };
}, never, never, never, never>usageMeter).function subscribe(channel: ConfiguredChannel): AgentBindingsubscribe(mainChannel: ConfiguredChannelmainChannel);

    endsOn: <AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>>(event: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>, when?: ((envelope: {
    ...;
}) => boolean) | undefined) => voidDeclares a terminal event: when it is published in a run (and `when` accepts it, if given),
the run completes with `m4trix:run.completed`, its in-flight agents are aborted and its
stream closes. Call it once per terminal event.endsOn(const Answer: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>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 them and end on the joined event.
No order between branches. Nothing orders one agent's work against another's. See 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.

AgentNetwork

channelmain

channelplans

channelclient

user-message

expose()

proxy.sse()

planner

executor

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.

Payload user-message

{
  "text": "What changed in the last release?"
}

chain.ts

export const const network: AgentNetwork<never>network = class AgentNetwork<TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never>AgentNetwork.AgentNetwork<TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never>.setup(callback: (ctx: AgentNetworkSetupContext) => void, options?: AgentNetworkSetupOptions): AgentNetwork<never>setup(
  ({ mainChannel: ConfiguredChannelmainChannel, createChannel: (name: string) => ConfiguredChannelcreateChannel, proxy: {
    sse(): ProxyDef;
    custom(kind: string, config?: unknown, direction?: ProxyDirection): ProxyDef;
}proxy, registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent, endsOn: <E extends EventDef>(event: E, when?: (envelope: EventEnvelope<E>) => boolean) => voidDeclares a terminal event: when it is published in a run (and `when` accepts it, if given),
the run completes with `m4trix:run.completed`, its in-flight agents are aborted and its
stream closes. Call it once per terminal event.endsOn }) => {
    const const plans: ConfiguredChannelplans = createChannel: (name: string) => ConfiguredChannelcreateChannel('plans').ConfiguredChannel.events(events: ReadonlyArray<EventDef>): ConfiguredChannelDeclares the events this channel carries. Publishing any other event to it is refused: an
agent's emit throws `EmitValidationError` (reason `channel`), an external publish resolves
`false`. Runtime `m4trix:*` events always pass. Without a declaration a channel carries
every event.events([const Plan: AgentNetworkEventDef<"plan", S.Struct<{
    steps: S.Array$<typeof S.String>;
}>>Plan]);
    const const client: ConfiguredChannelclient = createChannel: (name: string) => ConfiguredChannelcreateChannel('client').ConfiguredChannel.events(events: ReadonlyArray<EventDef>): ConfiguredChannelDeclares the events this channel carries. Publishing any other event to it is refused: an
agent's emit throws `EmitValidationError` (reason `channel`), an external publish resolves
`false`. Runtime `m4trix:*` events always pass. Without a declaration a channel carries
every event.events([const Answer: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>Answer]).ConfiguredChannel.proxy(...proxies: ProxyDef[]): ConfiguredChannelproxy(proxy: {
    sse(): ProxyDef;
    custom(kind: string, config?: unknown, direction?: ProxyDirection): ProxyDef;
}proxy.function sse(): ProxyDefsse());

    registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent(const planner: Agent<unknown, {
    name: "user-message";
    meta: EventMeta;
    payload: {
        readonly text: string;
    };
}, {
    name: "plan";
    payload: {
        readonly steps: readonly string[];
    };
}, never, never, never>planner).function subscribe(channel: ConfiguredChannel): AgentBindingsubscribe(mainChannel: ConfiguredChannelmainChannel).function publishTo(channel: ConfiguredChannel): AgentBindingpublishTo(const plans: ConfiguredChannelplans);
    registerAgent: (agent: AnyAgent) => AgentBindingregisterAgent(const executor: Agent<unknown, {
    name: "plan";
    meta: EventMeta;
    payload: {
        readonly steps: readonly string[];
    };
}, {
    name: "answer";
    payload: {
        readonly text: string;
    };
}, never, never, never>executor).function subscribe(channel: ConfiguredChannel): AgentBindingsubscribe(const plans: ConfiguredChannelplans).function publishTo(channel: ConfiguredChannel): AgentBindingpublishTo(const client: ConfiguredChannelclient);

    endsOn: <AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>>(event: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>, when?: ((envelope: {
    ...;
}) => boolean) | undefined) => voidDeclares a terminal event: when it is published in a run (and `when` accepts it, if given),
the run completes with `m4trix:run.completed`, its in-flight agents are aborted and its
stream closes. Call it once per terminal event.endsOn(const Answer: AgentNetworkEventDef<"answer", S.Struct<{
    text: typeof S.String;
}>>Answer);
  },
);

plans carries only plan, so the 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.

AgentNetwork

channelmain

channelwork

channelclient

user-message

expose()

proxy.sse()

coordinator

worker

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.

Payload user-message

{
  "text": "orders from last week"
}

emit-and-await.tsline 53, not shown

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({});

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

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.

AgentNetwork

channelmain

channeldrafts

channelclient

user-message

expose()

proxy.sse()

researcher

critic

editor

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.

Payload user-message

{
  "text": "the new caching layer"
}

join.tsline 80, not shown

// 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);
  },
);

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

retries.ts

export const const getQuote: ToolDefinition<{
    readonly symbol: string;
}, {
    readonly price: number;
}, never, never>getQuote = class Tool<TInput = unknown, TOutput = unknown, TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never, TEmits extends EventDef = never>Tool.Tool<TInput = unknown, TOutput = unknown, TDeps extends DependencyLayerDef<string, unknown, S.Schema.Any> = never, TEmits extends EventDef = never>.of(options: {
    name: string;
    description: string;
}): Tool<unknown, unknown, never, never>of({ name: stringname: 'getQuote', description: stringdescription: 'Latest price for a symbol.' })
  .Tool<unknown, unknown, never, never>.input<S.Struct<{
    symbol: typeof S.String;
}>>(schema: S.Struct<{
    symbol: typeof S.String;
}>): Tool<{
    readonly symbol: string;
}, unknown, never, never>input(import SS.function Struct<{
    symbol: typeof S.String;
}>(fields: {
    symbol: typeof S.String;
}): S.Struct<{
    symbol: typeof S.String;
}> (+1 overload)@categoryconstructors@since3.10.0Struct({ symbol: typeof S.Stringsymbol: import SS.class String
export String@categoryprimitives@since3.10.0String }))
  .Tool<{ readonly symbol: string; }, unknown, never, never>.output<S.Struct<{
    price: typeof S.Number;
}>>(schema: S.Struct<{
    price: typeof S.Number;
}>): Tool<{
    readonly symbol: string;
}, {
    readonly price: number;
}, never, never>output(import SS.function Struct<{
    price: typeof S.Number;
}>(fields: {
    price: typeof S.Number;
}): S.Struct<{
    price: typeof S.Number;
}> (+1 overload)@categoryconstructors@since3.10.0Struct({ price: typeof S.Numberprice: import SS.class Number
export Number@categoryprimitives@since3.10.0Number }))
  .Tool<{ readonly symbol: string; }, { readonly price: number; }, never, never>.timeout(duration: DurationInput): Tool<{
    readonly symbol: string;
}, {
    readonly price: number;
}, never, never>Fails a call with `ToolTimeoutError` when it runs longer than `duration` (per attempt with
`.retry()`). The tool's `signal` is aborted with that error so well-behaved tools stop;
the calling run's signal is left alone.timeout('2 seconds') // per attempt; a timed-out attempt is not retried
  .Tool<{ readonly symbol: string; }, { readonly price: number; }, never, never>.retry(options: ToolRetryOptions): Tool<{
    readonly symbol: string;
}, {
    readonly price: number;
}, never, never>Retries a call up to `times` more times when the tool throws (`ToolExecutionError`), with an
optional backoff. Invalid input, invalid output and timeouts are never retried, and no
attempt starts once the signal is aborted (an abort also cuts a backoff wait short).retry({ times: numberExtra attempts after the first one fails. `0` disables retries.times: 2, backoff?: {
    readonly base: DurationInput;
    readonly factor?: number;
    readonly max?: DurationInput;
} | undefinedWait before retry `n` (1-based): `base * factor^(n - 1)`, capped at `max`. Without it,
retries start immediately.backoff: { base: DurationInputbase: '200 millis', factor?: number | undefinedGrowth per retry; `1` waits `base` every time. Defaults to `2`.factor: 2, max?: DurationInput | undefinedmax: '2 seconds' } })
  .Tool<{ readonly symbol: string; }, { readonly price: number; }, never, never>.define(fn: ToolExecuteFn<{
    readonly symbol: string;
}, {
    readonly price: number;
}, Record<string, never>, never>): ToolDefinition<{
    readonly symbol: string;
}, {
    readonly price: number;
}, never, never>define(async ({ input: {
    readonly symbol: string;
}input, signal: AbortSignalThe calling agent's signal: aborted when its run ends. Never aborts for unbound use.signal }) => ({ price: numberprice: await function fetchPrice(symbol: string, signal: AbortSignal): Promise<number>fetchPrice(input: {
    readonly symbol: string;
}input.symbol: stringsymbol, signal: AbortSignalThe calling agent's signal: aborted when its run ends. Never aborts for unbound use.signal) }));

toolgetQuote

quoteAgent

attempt 1throws → retry

attempt 2throws → retry

attempt 3returns { price }

200 msretry 1

400 msretry 2

{ price }

retry({ times: 2, backoff: { base: 200 ms, factor: 2 } })

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.

For more, see Networks and the AgentNetwork API.