AgentNetwork

The AgentNetwork orchestrates agents, channels, and the event plane. Use AgentNetwork.setup() to wire everything together.

Setup

const network = AgentNetwork.setup(
  ({ mainChannel, createChannel, proxy, registerAgent, registerAggregator, endsOn }) => {
    // ...
  },
);

Setup Context

mainChannel

The network's main channel, named main. The network creates it, so it is a channel rather than a function: pass it straight to subscribe or publishTo. Start events are published here when the network is exposed as an API.

registerAgent(router).subscribe(mainChannel);

createChannel(name)

Creates an additional named channel. Names must be kebab-case.

const client = createChannel('client');
const analytics = createChannel('analytics');

proxy

Provides proxy factories: proxy.sse() (exposed with registerSSEStream) and proxy.custom(kind, config?, direction?) for your own kinds (see defineProxyKind):

const client = createChannel('client').proxy(proxy.sse());
const events = createChannel('events').proxy(proxy.custom('queue', { topic: 'events' }));

Kafka and Socket.IO bridges are not built in; they are planned on top of EventTransport.

registerAgent(agent)

Registers an agent and returns a binding builder:

registerAgent(myAgent)
  .subscribe(main)
  .publishTo(client);

An agent can subscribe to and publish to multiple channels:

registerAgent(routerAgent)
  .subscribe(main)
  .subscribe(feedback)
  .publishTo(client)
  .publishTo(analytics);

endsOn(event, when?)

Declares 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. when receives the typed envelope. Call endsOn once per terminal event.

endsOn(AnswerEvent);

// The user's message and the reply share an event: end on the reply only.
endsOn(MessageEvent, (event) => event.payload.role === 'assistant');

Terminal events live on the network because agents emit them, and because the event plane enforces them, they apply to expose(), custom proxies and network.run() alike. An agent can also end its run from logic with complete(). See Run Lifecycle.

Setup Options

The second argument of AgentNetwork.setup():

const network = AgentNetwork.setup(
  ({ mainChannel, registerAgent }) => {
    // ...
  },
  { onAgentError: 'continue' },
);

Agents can always fail their run explicitly with fail(error), whatever the policy.

The four run limits apply to every run of the network; Infinity disables one. Override them per plane with network.run({ maxDepth }) or per exposed run with registerSSEStream({ maxEvents }). A run that exceeds one fails with m4trix:run.failed, error.kind: 'limit-exceeded' and the limit ('depth' | 'events' | 'tokens' | 'cost') and its max; the event that would have exceeded maxDepth / maxEvents is not delivered. See Run limits, usage and budgets.

Wiring check

When the setup callback returns, AgentNetwork.setup() checks the network's wiring: which agents (and their tools) emit and listen to which events on which channels, against what the channels accept (.events()). Contradictions throw a NetworkWiringError that lists all of them at once (error.errors); smells are logged through the logger option and do not stop the network.

Two ways to silence a warning you expect:

• Reply events. Declare events that only answer an emitAndAwait with { reply: true }, on agents and tools alike. The waiting agent receives them without listening, so they are not reported as unconsumed (they must still be accepted by a channel the agent publishes to):

  AgentFactory.run()
    .listensTo([TaskRequested])
    .emits([Progress]) // listened to or streamed: checked as usual
    .emits([TaskCompleted], { reply: true }); // only answers the requester's emitAndAwait
  Tool.of({ name: 'approve', description: '…' }).emits([Approved], { reply: true });

• wiring.ignore leaves whole warning codes out of the check, at setup and in network.validate(), e.g. when you read channels yourself with plane.subscribe (errors cannot be ignored):

  AgentNetwork.setup(setup, { wiring: { ignore: ['unconsumed-event'] } });

Run the same check yourself, e.g. to assert a clean network in a test:

const { errors, warnings } = network.validate(); // ReadonlyArray<{ code, message, agentId?, event?, channels? }>
expect(warnings).toEqual([]);

Each issue has a code, a readable message and, where they apply, the agentId, event and channels concerned. network.validate() never throws; it also reports problems introduced after setup (e.g. .events() called on a channel later).

Running the Network

HTTP API (recommended)

const api = network.expose(
  registerSSEStream({
    channel: 'client',
    triggerEvents: [requestEvent],
  }),
);

See IO + Adapters.

Programmatic Run

await Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* () {
      // Returns once every agent is subscribed: events published next are not missed.
      const plane = yield* network.run();
      // publish events, wait for results...
    }),
  ),
);

Concurrency model

• Serial per run, parallel across runs. Each agent has one mailbox per run, merged across all channels it subscribes to. It handles a run's events one at a time, in publish order; a slow invocation for one run never delays another run. Cap an agent's invocations across runs with .concurrency(n).
• Back-pressure per run. Mailboxes hold capacity events (network.run({ capacity }), default 16). When one is full, outside publishers of that run (plane.publish, inbound proxy events) wait; other runs are unaffected. Events the run's own agents emit are queued past the bound instead, so one busy agent cannot hold back the rest of the run, e.g. the reply it is waiting for with emitAndAwait; the run's maxEvents limit bounds that growth. Transient events replace the oldest buffered transient event instead of waiting.
• No self-replies. Since an agent handles a run's events one at a time, it cannot emitAndAwait a reply that it would produce itself (in a later invocation of the same run); that wait only ends at its timeout. Replies from other agents are fine.
• Ordered emits. Everything an invocation emits (including emitAndAwait requests and complete()) is published in call order, inside the network's scope.
• Released on end. When a run ends, its queued events are dropped and its mailboxes and fibers are released; the plane only remembers the outcome of recently ended runs.

Event history and stores

Every durable event a run publishes is written to the network's store before it is delivered, so an agent always finds its trigger (and everything earlier) in history. Agents and aggregators read it on demand:

.logic(async ({ triggerEvent, history, emit }) => {
  const runEvents = await history.run(); // this run so far, in publish order, trigger included
  const page = await history.context({ limit: 50 }); // newest 50 of the conversation, oldest first
  const older = page.cursor ? await history.context({ limit: 50, before: page.cursor }) : undefined;
  const prompts = await history.context({ names: ['user-prompt'] }); // only events with these names
  const newer = await history.context({ after: page.head }); // what was appended after `page`
  const latest = await history.latest(['user-prompt']); // { event, cursor } | undefined
})

• history.run() returns the run's durable events. history.context({ limit?, before? }) returns { events, cursor?, head? }: the newest limit events of the context across its runs (all of them without limit), oldest first. Pass cursor as before for the page of older events; it is absent once the first event is reached.
• head is a cursor at the page's newest event (present whenever the page has events). Pass it as after to read forward: history.context({ after, limit? }) returns the events appended after that cursor, oldest first, the oldest limit of them, without a cursor. Keep each page's head and read after it next time to see only new events; cursors stay valid while events are appended. after and before together are an error.
• names keeps only the events with one of these names, in every kind of read. Cursors are still positions in the whole context, so filtered reads page and chain without gaps or repeats, also when events with other names are appended in between.
• history.latest(names) returns the context's newest event with one of names and a cursor at it (pass it as after to read what followed), or undefined when there is none, e.g. to start from the latest checkpoint instead of the first event (folds read their own checkpoints this way).
• History includes the runtime's fold checkpoints (m4trix:fold.checkpoint:<name>@<version>, see Compaction): isFoldCheckpointEvent(event) tells them apart. They are stored with their run but published to no channel, so streams never carry them.
• A page read after a cursor whose history is gone (the context was deleted or expired since, and may have started over) has reset: true: it holds the current history from its start, so anything derived from earlier pages is stale. Context folds are built on these reads and rebuild their state on a reset.
• History is scoped to the run's principal (see Auth + Multi-Tenant): (namespace, contextId), never contextId alone.
• Transient events are streamed and traced but never stored, so they are never in history.
• History holds the stored form. Each durable event is stored as a deep-frozen JSON copy, its payload encoded with its event's schema (S.encode): a Date (S.Date) is an ISO string, S.NumberFromString a string, an Option { _tag, value }. History events are therefore untyped (StoredEvent); treat them as read-only (the in-memory store hands out the frozen objects it got, so writing to them throws; a database store returns fresh, unfrozen ones). For plain schemas (strings, numbers, structs, arrays) the stored and the decoded form are the same; for transforming ones (a field renamed with S.fromKey is also stored under its encoded key), decode what you read: S.decodeUnknownSync(UserPrompt.payload)(event.payload), or UserPrompt.decode(event) (an Effect). UserPrompt.is checks the decoded form, so events.filter(Event.is) silently matches nothing for an event whose schema transforms: no error, just an empty list. Context folds decode for you.
• EventAggregator callbacks (emitWhen, mapToEmit) get the same history. Outside a network (agent.invoke()), history is empty. Code holding a plane can read it with plane.history({ runId, contextId, principalId? }).

Write failures and ordering

A store write that fails is retried (3 attempts in total, 10 ms then 20 ms apart). If it still fails, the event is not delivered (history never misses an event an agent saw) and the run fails with m4trix:run.failed, error.kind: 'store-error' (a client-safe NetworkStoreError; the store's own error is its cause, for tracers only). The run-end event itself is still streamed when the store is down. Writes of one run are applied one after another in publish order, whatever the store's latency; runs and contexts write concurrently.

What is stored. The runtime encodes each durable event's payload with its event's schema and stores a deep-frozen JSON copy, so the store never shares an object with what agents received:

• An agent's emit is encoded with the agent's definition of the event when it calls emit (changing the payload object afterwards changes nothing stored). The JSON copy is what JSON.stringify writes: a value with a toJSON method is stored as what it returns (a Date in an S.Unknown field as its ISO string, a Decimal as its own form), and undefined properties are left out. When a toJSON changed a value, the copy is decoded with the same schema again, so nothing is stored that reads back as something else. A payload its schema cannot encode, or that JSON cannot hold (a Map, Set, Error or class instance without a toJSON, a bigint, NaN, undefined in an array, a getter or a toJSON that throws, or a Date under S.DateFromSelf, whose ISO string is no Date), throws EmitValidationError (reason: 'invalid-payload'; for JSON, the message reads … is not JSON-compatible: a Map at output[0].rows and the issue's path names the value), like a payload its schema rejects: the run fails with invalid-emit.
• An event published from outside (plane.publish, a start event, a proxy's publish) is encoded with the schema the network knows for its name (from a channel's .events(), an agent's listensTo or emits, or a fold's handlers; the first definition of a name wins, and the wiring check warns when a name has different schemas). plane.publish takes the payload as agents receive it (decoded, e.g. from Event.makeBound); request bodies and proxy publishes are wire data and are decoded first. A payload that is not a value of the schema's type is stored as given, and folds skip it if it does not decode. The JSON copy follows the same rules as for an emit. A payload JSON cannot hold is refused by the publish itself: plane.publish, publishToChannels and publishAndAwait fail with NonJsonPayloadError (_tag, eventName, path, found, e.g. a Map) in their error channel, a proxy's publish rejects with it, and expose() answers such a start event with a 400. Nothing is published (no agent sees it, nothing is stored, nothing is retried) and the run goes on.
• Effect values in S.Unknown fields are stored as their toJSON, Effect's debug format, not a contract: an Option as { "_id": "Option", "_tag": "Some", "value": … } (which a later S.Option decodes, S.OptionFromSelf does not), an Either, Chunk or Duration likewise, a DateTime as its ISO string, a Data.TaggedError as its fields. Data.struct and Data.Class values have no toJSON and are refused, like an Error. Give such values a schema (S.Option(…)) where readers need them typed.
• S.Redacted encodes to the underlying value, the secret in clear, so a durable event's schema must not hold one: setup fails with durable-event-redacted. A transient event may carry one: it is never stored, and streams send it as "<redacted>". A Redacted value inside an S.Unknown field is stored as "<redacted>" (its toJSON).
• Runtime events (run lifecycle, fold checkpoints) are copied as they are: a checkpoint's state is encoded by its fold.
• Transient events are never stored, so they are not encoded at publish time; streams encode them as they send them (see wire format).

The in-memory store

The default store keeps history in the process: lost on restart and not shared between instances. It is bounded:

import { AgentNetwork, createInMemoryNetworkStore } from '@m4trix/core/matrix';

const network = AgentNetwork.setup(setup, {
  store: createInMemoryNetworkStore({
    ttl: '30 minutes', // drop a context this long after its last event (reads don't extend it)
    maxContexts: 1000, // beyond this, evict the least recently used context
  }),
});

Both are the defaults; Infinity disables either. Expiry is lazy (on read, or when another context is written), so the store runs no timers. Pass clock (anything with unsafeCurrentTimeMillis(), such as an Effect Clock) to control time in tests. Use a persistent store when history must survive restarts or be shared across instances.

Writing a store adapter

A store implements AgentNetworkStore:

interface AgentNetworkStore {
  append(key: AgentNetworkStoreKey, event: StoredEvent): Promise<void>;
  readRun(key: AgentNetworkStoreKey, runId: string): Promise<readonly StoredEvent[]>;
  readContext(key: AgentNetworkStoreKey, options?: ReadContextOptions): Promise<EventPage>;
  readLatest?(
    key: AgentNetworkStoreKey,
    names: readonly string[],
  ): Promise<{ readonly event: StoredEvent; readonly cursor: string } | undefined>;
  deleteContext?(key: AgentNetworkStoreKey): Promise<void>;
}
type AgentNetworkStoreKey = { readonly namespace: string; readonly contextId: string };
type ReadContextOptions = {
  readonly limit?: number;
  readonly before?: string;
  readonly after?: string;
  readonly names?: readonly string[];
};
type EventPage = {
  readonly events: readonly StoredEvent[];
  readonly cursor?: string;
  readonly head?: string;
  readonly reset?: true;
};

• append resolves once the event is durable and rejects on failure (the runtime retries). The runtime never has two appends of one run in flight, so appending in call order keeps each run in order.
• readRun returns one run's events in append order.
• readContext returns the newest limit events (before the before cursor) in append order, with a cursor for the older page while older events remain. With after it reads forward instead: the oldest limit events after that cursor, without a cursor. Reject after together with before. Every non-empty page has a head, a cursor at its newest event that works as after.
• Cursors are opaque strings of your choosing that stand for a position in the context's log (e.g. a sequence number) and must stay valid while newer events are appended. They are persisted: a fold checkpoint keeps the cursor of the read it was computed from, so keep their format stable across versions and restarts of your store.
• Positions only grow. An event appended while a cursor is out must become visible after it, never before it: a fold reads only the events after its cursor and would never fold one that lands before. A sequence handed out before a transaction commits can break this when transactions commit out of order (concurrent runs of one context append concurrently); assign positions at commit, or serialize appends per context.
• reset. An after cursor that does not belong to the context's current history (the context was deleted, expired or evicted since, also when it has started over) must not be read as a position in the new one. Return the current history from its start, as if read without after, with reset: true. Make cursors identify the history they belong to, e.g. a generation id plus a sequence number. The same for an after cursor your store does not recognise at all (another store's, e.g. in a migrated history): a reset page, not an error. Folds rely on this to never add new events onto state derived from a deleted history, and to stay readable when a checkpoint carries a foreign cursor (it is then ignored). A store whose pages do not follow its cursors makes a fold read fail after 100 rebases instead of looping.
• names keeps only events with one of these names, in backward and forward reads alike. cursor and head stay positions in the whole log: cursor points before the page's oldest matching event and is absent once no older matching event remains; head points at the page's newest event. The runtime never passes an empty list.
• readLatest (optional) returns the newest event with one of names and a cursor at it that works as after; undefined when there is none. Implement it when your backend can look it up cheaply (e.g. an index on the event name); without it, history.latest reads readContext({ names, limit: 1 }) and uses its head.
• deleteContext (optional) removes one context of one namespace, e.g. when a user clears a chat. Call it yourself through network.getStore().
• Keys. Scope every read and write by both namespace and contextId. Encode them unambiguously when you build one backend key (JSON.stringify([namespace, contextId]), or two columns): principal:a + b:c must not collide with principal:a:b + c.
• JSON in, the same JSON out. Events are JSON-compatible (name, meta, payload, and channels) and arrive deep-frozen: store them as JSON and return them unchanged. Do not revive values (e.g. ISO strings into Dates): folds decode payloads with the event's schema, which expects the encoded form. Key order may change (e.g. Postgres jsonb). The in-memory store keeps the frozen objects as they are.
• Do not add fields in place. append gets a frozen object: adding a field to it (a sequence number, a timestamp) throws in strict mode. Copy it first ({ ...event, seq }), or keep such data next to the event.
• Keep channels. The runtime stores each event with the channels it was published to (StoredEvent.channels). Resumed SSE streams replay only their own channel's events with it; without it they fall back to a guess (lifecycle events, and agents' events the channel accepts). Fold checkpoints are stored with channels: [] and never replayed.
• The store's owner manages its connection: core never opens or closes it.

Check an adapter against the contract suite from @m4trix/core/testing. It does not depend on a test framework: pass your runner's describe, it and expect (only toEqual is used):

// redis-network-store.spec.ts
import { describe, expect, it } from 'vitest';
import { runStoreContract } from '@m4trix/core/testing';
import { createRedisNetworkStore } from './redis-network-store';

runStoreContract({ describe, it, expect }, () => createRedisNetworkStore(redis));

It covers append and read order, JSON payloads returned exactly as appended, keeping channels, run vs context reads, paging cursors, head and forward reads with after (including chained heads while events are appended), names filters, reset for stale and unknown after cursors, namespace and context isolation, readLatest and deleteContext (when implemented), concurrent appends to different contexts, and positions that only grow while runs of one context append concurrently. Every case uses fresh keys, so one database may serve all cases. names, after, head and reset are required behavior of readContext, not optional extras: context folds read through them.

For a small, complete adapter, see examples/m4trix-code/src/file-store.ts: one JSON lines file, cursors that stay valid across restarts (jsonl:<n>), fresh objects on every read, and a torn last line dropped on open. It passes the contract and keeps a terminal session across restarts, compactions included.

Event transport

The event plane hands channel events to their subscribers (SSE streams, plane.subscribe) and emitAndAwait replies to their waiters through an EventTransport. Everything per run stays in the plane: mailboxes, ordering, back-pressure, limits, ownership and run ends.

type EventTransport = {
  // To every subscriber of each channel, and to every reply waiter the event matches.
  publish(channels: ReadonlyArray<string>, envelope: Envelope): Effect.Effect<void>;
  // Calls `handler` for each event published to `channel` from now on, until the scope closes.
  // The handler's completion acknowledges the event.
  subscribe(channel: string, handler: (envelope: Envelope) => Effect.Effect<void>): Effect.Effect<void, never, Scope.Scope>;
  // Waits for the first later event (not the request itself, by `eventId`) that `match` accepts.
  expectReply(request: Envelope, match: (reply: Envelope) => boolean): Effect.Effect<Effect.Effect<Envelope>, never, Scope.Scope>;
};

The default is a fresh createInMemoryTransport() per plane, which keeps everything in the process. Pass one instance as the transport setup option to connect all planes of the network in the process (e.g. a stream on one expose() request sees events published on another plane). A transport backed by Redis Streams, NATS or Kafka would connect several instances; see Deployment. Check an implementation with runTransportContract({ describe, it, expect }, createTransport) from @m4trix/core/testing, like a store.

plane.publishAndAwait and emitAndAwait fail with PublishAndAwaitTimeoutError (_tag, timeout) when no reply arrives in time (default 30 seconds). plane.publish, publishToChannels and publishAndAwait are the way in from outside the network: they fail with ReservedEventNameError (_tag, eventName) for a reserved m4trix: name (the runtime publishes its own lifecycle events and fold checkpoints itself) and with NonJsonPayloadError (_tag, eventName, path, found) for a durable event whose payload the store cannot keep as JSON (see What is stored). Either way nothing is published and the run goes on.

Accessors

network.getChannels();             // Map<ChannelName, ConfiguredChannel>
network.getMainChannel();          // ConfiguredChannel | undefined
network.getAgentRegistrations();   // Map<string, AgentRegistration>
network.getTerminalEvents();       // ReadonlyArray<RunTerminalEvent> (from endsOn)
network.getAgentErrorPolicy();     // 'fail' | 'continue'
network.getRunLimits();            // { maxDepth, maxEvents, maxTokens, maxCostUsd }, defaults filled in
network.getStore();                // AgentNetworkStore (the `store` option or the in-memory default)
network.getFoldCacheOptions();     // the `foldCache` option
network.getTransport();            // EventTransport | undefined (the `transport` option)
network.getLogger();               // the `logger` option (or console), with every level
network.getWiringOptions();        // the `wiring` option
network.validate();                // { errors, warnings } of the wiring check

See Also

• Networks (Concepts)
• Context (Folds and Views)
• Patterns
• IO + Adapters