Channels & the Envelope

A channel is anything a run can come from or talk to: a cron schedule, the REST API and MCP, generic webhooks, GitHub, an OpenClaw chat, the console. One contract covers all of them, so starting a run, answering where it came from and asking the owner for an approval work the same way on every transport.

interface IChannel<TInput = ChannelRequest, TOutput = Response> {
    readonly name: string;
    readonly defaultAddress?: string;            // where emit()/approve() send when no address is given
    start?(ctx: ChannelContext): Promise<void>;  // background intake: cron ticks, sockets, polling
    stop?(): Promise<void>;
    in?(input: TInput, ctx: ChannelContext): Promise<TOutput>;
    out?(message: ChannelMessage, address: string, opts: ChannelSendOptions): Promise<void>; // idempotent on message.key
}

type ChannelRequest = { request: Request; subpath: string };  // the raw request, the raw path below /channels/<name>
type ChannelContext = { engine: IEngine; dispatch(envelope: ChannelDispatchEnvelope): Promise<DispatchResult> };

type ChannelMessage = {
    kind: 'message' | 'approval';
    key: string;          // stable per logical message: the transport's dedupe key
    slot?: string;        // explicit named persistent message, where supported (GitHub)
    runId: string;
    agent: string;
    title?: string;
    body: string;
    approvalId?: string;  // kind 'approval'
    expiresAt?: Date;     // kind 'approval'
    options?: ApprovalOption[]; // kind 'approval' with options: { id, label, detail? }
    multiple?: boolean;   // kind 'approval', true when several options may be selected
};

type ChannelSendOptions = { signal: AbortSignal; trigger: TriggerEnvelope };

The presence of start, in and out is a channel's only capability: start/stop run background intake (cron ticks, sockets, polling), in is the single inbound entry a transport feeds (the HTTP mount, tests), and out sends a message to an address the channel defines (owner/repo#12 on GitHub, a chat recipient on OpenClaw). A channel's name follows the agent-name rule (^[a-z0-9]+(-[a-z0-9]+)*$) and is unique per engine. An out-only channel that always talks to one recipient sets defaultAddress (a non-empty string): consoleChannel() uses its name, the OpenClaw channel its deliver.to.

Registering channels

The engine holds the channel instances:

const triage = defineAgent({
    name: 'triage',
    model,
    instructions: 'Triage the issue in the payload.',
    channels: [
        http.on(), // channel 'http', id 'http'
        webhook.on('/issues'), // channel 'webhook', id 'webhook:/issues'
        cron.on('*/30 * * * *', { overlap: 'skip' }), // channel 'cron', id 'cron:*/30 * * * *'
    ],
});

const engine = new Engine({
    store: createMemoryStore(),
    agents: [triage],
    // every subscribed channel must be registered (InvalidAgentDefinitionError otherwise)
    channels: [httpChannel(), webhookChannel({ auth }), cronChannel()],
});

new Engine({ channels }) refuses two channels with one name (InvalidAgentDefinitionError).
engine.start() recovers runs, then calls every channel's start(ctx) in registration order; a throwing start stops the channels already started and rejects engine.start(). engine.start({ channels: false }) starts none (a one-shot cron job uses it).
engine.stop() stops the poll and calls every stop(), resolving once all of them settled (await it before closing the store); engine.drain() does the same first (no new intake), then drains. A throwing stop goes to onError and the drain goes on.
ctx.dispatch(envelope) is engine.dispatch() with channel set to the channel's own name: a channel cannot dispatch as another.
The name agent belongs to agentChannel() from @chamvi/core (no in, no out), the channel hand-offs come in on. @chamvi/app contributes it, so a user channel named agent clashes at setup (The channel "agent" is registered more than once.).

Subscriptions

Every subscription builder accepts concurrencyMaxRuns?: number | false: a positive safe integer adds a shared-store execution ceiling for that subscription; false or omission adds none. It never bypasses the agent's own concurrencyMaxRuns. REST and MCP runs share the cap of http.on(). This is separate from concurrency keys, cron overlap and HTTP request rate limiting. See Concurrency.

An agent subscribes to channels with channels: ChannelSubscription[]. Each channel package exports a builder named after its default channel; its on() makes subscriptions, and they are pure data, so an agent folder imports them without an instance:

Builder	Package	Channel	Id	Options
`http.on(opts?)`	`@chamvi/http`	`http` (REST) and `mcp`	`http`	`schema?`: a Standard Schema for the payload. At most one per agent; REST, the UI and MCP start runs through it
`webhook.on(path, opts?)`	`@chamvi/http`	`webhook`	`webhook:<path>`	`path` (starts with `/`), `schema?`, `idempotencyKey?`, `concurrencyKey?`. Many per agent
`cron.on(expression, opts?)`	`@chamvi/channel-cron`	`cron`	`cron:<expression>`	`timezone?`, `overlap?` (`skip` by default, `queue-one`, `allow`), `payload?`, `id?`
`github.on(events, opts?)`	`@chamvi/github`	`github`	`github:<events>`	`events` (`'issues'` or `'issues.opened'`, one or an array), `id?`, `filter?`, `idempotencyKey?` (default the delivery id), `concurrencyKey?` (default `<owner>/<repo>#<number>`). See GitHub App
`handoff.on(opts?)`	`@chamvi/core`	`agent`	`handoff`	`schema?`, `maxPerRun?` (default 100), `idempotencyKey?`, `concurrencyKey?`. At most one per agent; the opt-in for `startHandoff()`

A second instance of a channel (cronChannel({ name: 'cron-eu' })) has its own on() bound to its name, so its subscriptions reach only it. Subscription ids must be unique per agent. The engine (constructor and putAgent) refuses a subscription naming a channel it does not hold, so register httpChannel() for http.on(), cronChannel() for cron.on(), and so on. With @chamvi/app the built-in modules register agent, http, mcp, webhook and cron for you.

A subscription may carry concurrencyKey(payload) and idempotencyKey(payload) functions; the channel that dispatches computes the envelope keys from them.

The envelope

/*
 * Copyright (c) 2026.
 * Author Peter Placzek (tada5hi)
 * For the full copyright and license information,
 * view the LICENSE file that was distributed with this source code.
 */

import type { Artifact } from '../artifact/types.ts';

/**
 * Who started a run (or answered an approval), as an authenticator resolved
 * it: a snapshot of identity and grants taken at dispatch, never a token.
 * Revoking a grant later does not reach a run that already holds it.
 */
export type Invoker = {
    id: string;
    /**
     * The authenticator's label: `@chamvi/http`'s staticTokens() sets
     * 'token' and authup() sets 'authup'; oidc() keeps the kind its
     * `principal` mapper returns (the docs use 'oidc'); a custom authenticator
     * or a direct dispatch may use its own (e.g. 'user').
     */
    kind: string;
    name?: string;
    permissions: readonly string[];
};

/**
 * The run that started this one. Only startHandoff() sets it (kind
 * `handoff`); a dispatch never carries one, so no caller forges lineage.
 */
export type TriggerParent = {
    kind: 'handoff';
    runId: string;
    agent: string;
    depth: number;
};

export type TriggerEnvelope<P = unknown> = {
    id: string;
    agent: string;
    /** The id of the agent's subscription that started the run. */
    trigger: string;
    /** The name of the channel the run came from. */
    channel: string;
    payload: P;
    idempotencyKey?: string;
    concurrencyKey?: string;
    /** Where the run came from on that channel (an issue, a thread, a caller); the channel defines it. */
    address?: string;
    /** Files the run was started with, stored as artifacts of the run; the model gets them as file parts. */
    attachments?: Artifact[];
    /**
     * Who started the run; only ever set from an authenticator (never from a
     * payload). Absent for runs a channel started without one (cron, webhooks). With an invoker the
     * idempotency key is scoped to it, so only the same invoker's dispatch
     * is deduplicated (and gets its own first run).
     */
    invoker?: Invoker;
    /** The run that handed this one off; set only by startHandoff(), never from a dispatch. */
    parent?: TriggerParent;
    receivedAt: Date;
};

dispatch() takes the envelope without id and receivedAt; it fills in a random UUID and the current time. trigger is the subscription id ('http', 'webhook:/issues', ...), channel the name of the channel the run came from ('mcp' for a run started through http.on() over MCP). address is where the run came from on that channel, as the channel defines it (the issue for GitHub, a chat recipient); it is stored on the run as run.address and is where ctx.channels.findOne() answers. An idempotencyKey, concurrencyKey or address above 255 characters or with a NUL is refused (ValidationError) before anything is written.

The engine renders the channel, subscription id and payload as a JSON block and makes it the run's first user message. The model sees exactly that; put what it needs into the payload.

invoker is who started the run. Only a channel sets it, from its authenticated caller (@chamvi/http fills it from the principal), never from a payload. It is not part of the rendered block, so the model never sees it. See Restricting tools per invoker.

Runs stored before channels existed (source and thread in the envelope) read back as channel and address from both repositories.

Outbound from tools

The engine never sends a message of its own. A tool sends through a channel:

export const resolve = defineTool({
    description: 'Report the fix and tell the on-call channel.',
    inputSchema: z.object({ summary: z.string() }),
    async execute({ summary }, { context }) {
        // the run's source channel at its source address; undefined when that channel has no out
        await context.channels.findOne()?.emit({ body: summary });
        // any registered channel with out; the address is required off the source channel
        await context.channels.findOneOrFail('console').emit({ title: 'Resolved', body: summary }, 'on-call');

        return 'reported';
    },
});

export function createEngine(model: LanguageModel) {
    return new Engine({
        store: createMemoryStore(),
        agents: [defineAgent({
            name: 'fixer',
            model,
            instructions: 'Fix the issue, then report it.',
            tools: { resolve },
        })],
        channels: [consoleChannel()],
    });
}

ctx.channels.findOne(name) returns a handle on a registered channel with out, or undefined for an unknown name or a channel without out. Without a name it is the run's source channel, bound to its source address, so a tool answers where the run came from without knowing the channel (and gets undefined for a run from http, mcp or cron).
ctx.channels.findOneOrFail(name?) is the same, but throws ChannelNotFoundError (channel_not_found) instead of returning undefined: use it for a channel the agent cannot work without.
ctx.channels.find() returns a handle on every channel with out, in registration order.
handle.emit({ title?, body, key?, slot? }, address?) sends through out. The address defaults to the source address on the source channel, else to the channel's defaultAddress; without either a missing address is a validation_error tool error.
An emit is an effect: its key is <runId>:emit:<channel>:<sha256 of address and message> (or your key), recorded in the effect store, so identical emits of one step and a step redone after a crash send once. out gets the key as message.key, to dedupe on in the transport. A throwing out records nothing and is a tool error, so the next identical emit sends again.
emit outside a tool call (in the tools factory) throws InvalidRunStateError: the tool call is the unit of retry.
handle.approve({ title, body?, options?, multiple?, timeoutMs?, address?, slot? }) asks the owner through the channel; see Approvals.

Named message slots

A transport can support a persistent message addressed by slot. GitHub uses it to create a comment once, then update it across runs; omit slot to append an ordinary comment. A slot is a non-empty, well-formed Unicode string without NUL characters, chosen explicitly on each emit() or channel approve() call. A single channel and agent can mix ordinary replies and several independent slots. Slot selection is separate from the per-delivery key, which still deduplicates retries. The effect key includes the slot when present, so identical text or custom keys aimed at different slots do not suppress each other. Approval slots are stored with the waiting run and retained on recovery. Other shipped transports currently ignore slot hints. See GitHub persistent status comments.

Writing a channel

A chat bridge with both directions:

type ChatPost = {
    channel: string;
    text: string;
    user: string
};

/** A chat bridge: messages in at POST /channels/chat, answers out to the chat's API. */
export function chatChannel(options: { secret: string; apiUrl: string }): IChannel {
    return {
        name: 'chat',

        // served by createHttpApp({ channels: [chat] }) at POST /channels/chat[/...]
        async in({ request }: ChannelRequest, ctx: ChannelContext): Promise<Response> {
            // the http app never authenticates channel routes: check your own credentials
            if (request.headers.get('authorization') !== `Bearer ${options.secret}`) {
                return new Response(null, { status: 401 });
            }

            // 413 above the limit
            const bytes = await readRequestBody(request, { limit: 64 * 1024 });
            const post = JSON.parse(new TextDecoder().decode(bytes)) as ChatPost;

            if (typeof post.text !== 'string') {
                throw new ValidationError('text is required.');
            }

            // dispatches as this channel (envelope channel 'chat'), never as another
            const { runId } = await ctx.dispatch({
                agent: 'helpdesk',
                trigger: 'chat',
                payload: post,
                address: post.channel,
            });

            return Response.json({ runId }, { status: 202 });
        },

        // called for emit() and for approval requests; must be idempotent on message.key
        async out(message: ChannelMessage, address: string, opts: ChannelSendOptions): Promise<void> {
            const response = await fetch(`${options.apiUrl}/messages`, {
                method: 'POST',
                headers: { 'content-type': 'application/json', 'idempotency-key': message.key },
                body: JSON.stringify({
                    channel: address,
                    text: [message.title, message.body].filter(Boolean).join('\n'),
                    // an approval carries its id and expiry: link to the UI
                    link: message.approvalId ?
                        `https://chamvi.example/ui/approvals?run=${message.runId}&approval=${message.approvalId}` :
                        undefined,
                }),
                signal: opts.signal,
            });

            // throw on failure: an emit records nothing, an approval is re-sent on the next poll
            if (!response.ok) {
                throw new Error(`Chat answered ${response.status}.`);
            }
        },
    };
}

Rules for a channel:

in checks its own credentials. createHttpApp({ channels }) mounts POST /channels/<name> and everything below it for the listed channels, rate-limited but never authenticated by the app's authenticators. Read the body with readRequestBody(request, { limit }) from @chamvi/http (413 above the limit); any thrown error renders through the app's JSON error handler.
out throws on failure. A failed emit records nothing; a failed approval message leaves waiting.notifiedAt unset and tick() re-sends it on every poll.
out is idempotent on message.key. A crash after a successful send but before the record re-sends with the same key; pass it to the transport (an Idempotency-Key header, a hidden marker) so the owner gets one message.
Link to the approval. A message carries the approval id; link to the approval in the dashboard, where the owner answers signed in.

ChannelSendOptions carries the call's signal and trigger, the run's stored envelope (e.g. to find the run's issue); read it, never write it.

Shipped channels

Channel	Package	In	Out
`httpChannel()`, `mcpChannel()`	`@chamvi/http`	REST `POST /agents/:name/runs`, MCP `run_<agent>` (served by the app's routes)	none
`webhookChannel({ auth })`	`@chamvi/http`	`POST /channels/webhook/<agent><path>`	none
`cronChannel()`	`@chamvi/channel-cron`	ticks from `start()`	none
`githubChannel({ app })`	`@chamvi/github`	signed deliveries at `POST /channels/github`	a comment on an issue or pull request (`owner/repo#n`), deduplicated by a hidden marker
`openclawChannel({ ... })`	`@chamvi/channel-openclaw`	none	an OpenClaw chat through its hooks endpoint, deduplicated by `Idempotency-Key`
`chatChannel({ ... })`	`@chamvi/chat`	Chat SDK adapters: webhooks at `POST /channels/chat/<adapter>`, socket messages from `start()`	a text message in the chat (approvals as a UI link), at least once
`createWhatsAppChat({ ... })`	`@chamvi/chat-whatsapp`	WhatsApp Web socket messages from `start()` (a `chatChannel()` with the WhatsApp platform hooks)	a text message in the WhatsApp chat, at least once
`consoleChannel({ name?, write? })`	`@chamvi/core`	none	one header line per message, the body lines and, for an approval, until when it waits (default `console.log`); never throws

Attachments

dispatch() also takes files for the model, attachments: [{ mediaType, filename?, data }] with data a Uint8Array:

const withFile = await engine.dispatch({
    agent: 'triage',
    trigger: 'http',
    channel: 'http',
    payload: { repo: 'acme/app', issue: 43 },
    // stored as artifacts of the run; the model gets them as file parts
    attachments: [{
        mediaType: 'image/png',
        filename: 'crash.png',
        data: new Uint8Array(await readFile('crash.png')),
    }],
});

const run = await engine.getRun(withFile.runId);

run?.trigger.attachments; // [{ id, runId, agent, mediaType: 'image/png', filename: 'crash.png', size, createdAt }]

Each attachment is checked before anything is written (mediaType of the form type/subtype without parameters, data a Uint8Array; ValidationError otherwise), then stored as an artifact of the run, after the idempotency check (a deduplicated dispatch stores nothing) and before the run is created. The run's trigger.attachments holds their metadata. The first user message then carries the rendered envelope as text followed by one file part per attachment, so a model that reads images or PDFs gets them directly. Attachment ids are derived from the run id and the position (<runId>.0000, <runId>.0001, ...), so the run's artifacts list in the order they were given. A crash between storing the files and creating the run leaves orphan artifacts; nothing purges them yet. Two dispatches that race to create the run of the same reserved idempotency key (its first run was never created) write to the same ids, so the second write of each file is a no-op and the run names exactly the files that were stored; only a racing dispatch with more files than the winner leaves its extra files listed among the run's artifacts.

Over HTTP, POST /agents/:name/runs takes the same list with data as base64 (see HTTP API).

Idempotency keys

const first = await engine.dispatch({
    agent: 'triage',
    trigger: 'webhook:/issues',
    channel: 'webhook',
    payload: { repo: 'acme/app', issue: 42 },
    // where the run came from on its channel; ctx.channels.findOne() answers there
    address: 'acme/app#42',
    // a redelivery with the same key returns the same run
    idempotencyKey: 'delivery-7f3a',
    // one run at a time per issue
    concurrencyKey: 'acme/app#42',
});

const again = await engine.dispatch({
    agent: 'triage',
    trigger: 'webhook:/issues',
    channel: 'webhook',
    payload: { repo: 'acme/app', issue: 42 },
    idempotencyKey: 'delivery-7f3a',
});

again.runId === first.runId; // true
again.deduplicated; // true

With an idempotencyKey, dispatch() claims the key in the store under the scope <agent>:<channel> (for example triage:webhook) for 7 days. The first dispatch creates the run. Every later dispatch with the same agent, channel and key returns the existing run id with deduplicated: true, and no new run is created.

A dispatch with an invoker claims the key under <agent>:<channel>:<kind>:<id> instead (kind and id URI-encoded). Keys are then per invoker: the same invoker's retry is deduplicated, while another invoker sending the same key gets a run of its own and never learns about the first.

Use a key that is stable across redeliveries: a GitHub delivery id, a cron subscription plus its scheduled time, or a key your client sends.

If the store claimed the key but the run was never created (for example runs.insert failed right after the claim), the next dispatch with that key creates the run under the reserved id. runs.insert is insert-if-absent, so concurrent dispatches create it once.

Concurrency keys

A run with a concurrencyKey starts only when no other run of the same agent with the same key is running, waiting or interrupted. Runs with different keys, and runs without a key, may run in parallel within the engine's local slots and any shared agent/subscription concurrencyMaxRuns ceilings. See Concurrency, Leases & Multi-Replica.

Typical keys: one per repository, issue or pull request (acme/app#42), or one per mailbox.

Channels & the Envelope ​

Registering channels ​

Subscriptions ​

The envelope ​

Outbound from tools ​

Named message slots ​

Writing a channel ​

Shipped channels ​

Attachments ​

Idempotency keys ​

Concurrency keys ​