Open Agent Loops

Channels

A step-by-step tutorial — from a bare runAgent call to a live channel bot. Each step adds exactly one layer, and every step is a runnable file.

A channel is a long-lived, bursty transport — a Slack or Discord websocket — that emits an unbounded stream of inbound messages. runAgent is the opposite: a request/response unit, one prompt per session, that loops to a final answer. The job of this tutorial is to put the two together without coupling their two rate domains: the socket must drain continuously (stall it and the provider disconnects you), while the model is slow and rate-limited.

The punchline up front: every layer in this stack exists only to manufacture one call — runAgent({ ...base, sessionId, prompt }). The tutorial starts by making that call by hand, then adds one layer per step until it's a live bot, so nothing is ever hidden:

two rate domains:transport side — drains continuously (ChannelSource)bounded queue — the impedance matcher (Dispatcher)model side — slow, rate-limited (runAgent)
The socket must never block: it drains continuously while backpressure is applied at the queue. ChannelSource (the ringed node) is the one piece you implement per provider — swap it for Slack or Discord and nothing downstream changes. Drag the nodes to rearrange.
  • ChannelSource — the transport seam, analogous to ModelClient. It owns liveness only: connect, heartbeat, reconnect, and normalizing provider events into one InboundMessage { channelId, threadId, userId, text }.
  • ChannelBridge — the wiring in the middle. Inbound: maps each message to a sessionId and submits it. Outbound: coalesces the run's token stream into one reply per turn and posts it back to the originating thread.
  • Dispatcher — the throttling layer the bridge owns: a bounded queue, at most one in-flight run per session, a global concurrency cap.

Step 1: the agent loop, with no channel anything

Before any channel concept, run the agent by hand. base is just a variable holding the reusable half of runAgent's arguments — model and memory. The per-call half is sessionId and prompt:

examples/channels-tutorial/step1.ts
import {
  assistantMessage,
  contentToText,
  Role,
  runAgent,
  SessionMemoryStore,
  StreamEventType,
} from "@open-agent-loops/core";
import type { ModelClient, ModelRequest } from "@open-agent-loops/core";

// A stand-in LLM: replies with what it saw and how many user turns it was
// given. The turn count is how we'll SEE memory working.
const echoModel: ModelClient = {
  async *stream(request: ModelRequest) {
    const turns = request.messages.filter((m) => m.role === Role.User);
    const lastTurn = turns[turns.length - 1];
    const reply = `re: ${contentToText(lastTurn?.content ?? [])} (turn ${turns.length})`;
    yield { type: StreamEventType.Done, message: assistantMessage({ content: reply }) };
  },
};

// The reusable half of every runAgent call. Later steps hand this exact shape
// to the ChannelBridge as `base` — nothing more is ever in it.
const base = { model: echoModel, memory: new SessionMemoryStore() };

// One helper so each call prints its reply.
async function ask(sessionId: string, prompt: string): Promise<void> {
  const result = await runAgent({ ...base, sessionId, prompt });
  const last = result.messages[result.messages.length - 1];
  console.log(`${sessionId}  "${prompt}" →`, contentToText(last?.content ?? []));
}

// Same sessionId twice → the second run loads the first from memory: "turn 2".
await ask("demo:t1", "hello");
await ask("demo:t1", "still there?");

// A different sessionId → a fresh drawer: back to "turn 1".
await ask("demo:t2", "new thread");
bun run examples/channels-tutorial/step1.ts
demo:t1  "hello" → re: hello (turn 1)
demo:t1  "still there?" → re: still there? (turn 2)
demo:t2  "new thread" → re: new thread (turn 1)

The one idea to take from this step: multi-turn is nothing but the sessionId. The second call on demo:t1 says turn 2 because runAgent loaded the first exchange from memory under that key — no object was carried between the calls. demo:t2 is a different key, so it's a fresh conversation. Everything a "channel" will do later is compute this key from a thread and make this exact call.

Step 2: name the agent, and let the Dispatcher drive it

A socket callback can't await the agent inline: it must return immediately (or the socket stalls), and two concurrent runs on one session would corrupt memory ordering. The Dispatcher is the fix — but first, name the agent as a top-level value:

const agent: Agent = (call) => runAgent({ ...base, ...call });

An Agent closes over its own config (model, memory, tools, system) and receives only the per-call half — { sessionId, prompt, signal, onEvent } — the four things only its driver knows at submit time. It's just a function, so you can drive it by hand, hand it to a dispatcher, or test it directly.

Then new Dispatcher({ agent }), and submit(sessionId, message) per arrival — which returns instantly; the dispatcher calls the agent when it's safe: one run per session at a time, bounded queue in front. (The classic { base, run? } form still exists as shorthand when you don't want to name the agent.)

examples/channels-tutorial/step2.ts
import {
  AgentEventType,
  assistantMessage,
  contentToText,
  Dispatcher,
  Role,
  runAgent,
  SessionMemoryStore,
  StreamEventType,
  userMessage,
} from "@open-agent-loops/core";
import type { Agent, Message, ModelClient, ModelRequest } from "@open-agent-loops/core";

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

// Same stand-in LLM as step 1: the turn count shows memory working.
const echoModel: ModelClient = {
  async *stream(request: ModelRequest) {
    const turns = request.messages.filter((m) => m.role === Role.User);
    const lastTurn = turns[turns.length - 1];
    const reply = `re: ${contentToText(lastTurn?.content ?? [])} (turn ${turns.length})`;
    yield { type: StreamEventType.Done, message: assistantMessage({ content: reply }) };
  },
};

// THE AGENT, as a top-level value. It closes over its own config (the same
// `base` shape as step 1) and receives the per-call half from whoever drives
// it — you, a dispatcher, a test.
const base = { model: echoModel, memory: new SessionMemoryStore() };
const agent: Agent = (call) => {
  const prompt = call.prompt as Message[];
  const text = prompt.map((m) => contentToText(m.content)).join(" + ");
  console.log(`agent called with { sessionId: "${call.sessionId}", prompt: "${text}" }`);
  return runAgent({ ...base, ...call });
};

// Part A — step 1's manual call, seeding the conversation on demo:t1. An
// Agent is just a function, so you can still drive it by hand.
const seeded = await agent({ sessionId: "demo:t1", prompt: [userMessage({ content: "hello" })] });
const seededLast = seeded.messages[seeded.messages.length - 1];
console.log(`you drove the agent yourself →`, contentToText(seededLast?.content ?? []));

// Part B — the dispatcher drives the SAME agent, one run per session at a time.
const dispatcher = new Dispatcher({
  agent,
  // The run's events, tagged with their sessionId — how replies get back out.
  onSessionEvent: (sessionId, event) => {
    if (event.type !== AgentEventType.Message) return;
    if (event.message.role !== Role.Assistant) return;
    console.log(`reply for ${sessionId}       →`, contentToText(event.message.content));
  },
});

// "A message arrived" — what a channel does per inbound event. submit returns
// immediately; the run happens when a slot is free, one per session at a time.
dispatcher.submit("demo:t1", userMessage({ content: "still there?" }));
await sleep(100);
dispatcher.submit("demo:t2", userMessage({ content: "new thread" }));
await sleep(100);

// The proof it's the same machinery: the dispatcher's demo:t1 run said
// "turn 2" — it CONTINUED the conversation Part A started by hand, because
// it's the same agent, the same memory, the same sessionId.
bun run examples/channels-tutorial/step2.ts
agent called with { sessionId: "demo:t1", prompt: "hello" }
you drove the agent yourself → re: hello (turn 1)
agent called with { sessionId: "demo:t1", prompt: "still there?" }
reply for demo:t1       → re: still there? (turn 2)
agent called with { sessionId: "demo:t2", prompt: "new thread" }
reply for demo:t2       → re: new thread (turn 1)

The smoking gun is turn 2: the dispatcher's demo:t1 run continued the conversation the manual call started — same agent, same memory, same sessionId. The dispatcher added scheduling, not machinery. Replies now come back through onSessionEvent (the run's events tagged with their session) instead of a return value — that's what lets the next step route them to the right thread.

Step 3: a fake socket, end to end — composed from top-level components

Now the full picture from the diagram, with every layer constructed explicitly, in dependency order. A FakeSocket stands in for the provider websocket and pushes events; a ~30-line FakeSocketSource implements the ChannelSource seam over it (start / send / stop, normalizing frames into InboundMessage); the Dispatcher is the same component you built in step 2, driving the same kind of Agent value; and the ChannelBridge just connects source ↔ dispatcher — it computes the sessionId from the thread, submits, and routes the reply back out via source.send.

Because you own the dispatcher, you can attach your own observer with addSessionListener — the bridge's reply router registers through the same hook, so they are peers. Each hop logs its handoff, so the output is the diagram:

examples/channels-tutorial/step3.ts
import {
  AgentEventType,
  assistantMessage,
  ChannelBridge,
  contentToText,
  Dispatcher,
  Role,
  runAgent,
  SessionMemoryStore,
  StreamEventType,
} from "@open-agent-loops/core";
import type {
  Agent,
  ChannelSource,
  InboundMessage,
  ModelClient,
  ModelRequest,
  OutboundTarget,
} from "@open-agent-loops/core";

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

// ── The channel: a fake provider socket ─────────────────────────────────────
// Stands in for the Slack/Discord websocket. It knows nothing about this
// library — it just emits provider-shaped frames and accepts posted replies,
// which is all a real socket does.
type SocketFrame = { channel: string; thread: string; user: string; body: string };

class FakeSocket {
  private listener: ((frame: SocketFrame) => void) | undefined;

  connect(onFrame: (frame: SocketFrame) => void): void {
    this.listener = onFrame;
    console.log("[socket]     connected");
  }

  disconnect(): void {
    this.listener = undefined;
    console.log("[socket]     disconnected");
  }

  /** The provider pushing an event down the wire. */
  push(frame: SocketFrame): void {
    if (this.listener) this.listener(frame);
  }

  /** The provider's "post message" API. */
  post(channel: string, thread: string, text: string): void {
    console.log(`[socket]     ⇦ reply posted to ${channel}/${thread}: "${text}"`);
  }
}

// ── The seam: ChannelSource over the fake socket ────────────────────────────
// The one piece you implement per provider: start/send/stop, and normalizing
// provider frames into `InboundMessage`. No model, no queue — those live in
// the dispatcher below.
class FakeSocketSource implements ChannelSource {
  constructor(private readonly socket: FakeSocket) {}

  start(onMessage: (message: InboundMessage) => void): void {
    this.socket.connect((frame) => {
      const message: InboundMessage = {
        channelId: frame.channel,
        threadId: frame.thread,
        userId: frame.user,
        text: frame.body,
      };
      console.log(`[source]     frame → InboundMessage → bridge: "${message.text}"`);
      onMessage(message); // hand off and return — never block the socket
    });
  }

  send(target: OutboundTarget, text: string): void {
    this.socket.post(target.channelId, target.threadId ?? "", text);
  }

  stop(): void {
    this.socket.disconnect();
  }
}

// ── The agent loop's model: a stand-in LLM ──────────────────────────────────
const echoModel: ModelClient = {
  async *stream(request: ModelRequest) {
    await sleep(100);
    const turns = request.messages.filter((m) => m.role === Role.User);
    const lastTurn = turns[turns.length - 1];
    const reply = `re: ${contentToText(lastTurn?.content ?? [])} (turn ${turns.length})`;
    yield { type: StreamEventType.Done, message: assistantMessage({ content: reply }) };
  },
};

// ── The agent, as a top-level value (same shape as step 2) ──────────────────
const memory = new SessionMemoryStore();
const agent: Agent = (call) => runAgent({ model: echoModel, memory, ...call });

// ── The throttle: a Dispatcher YOU construct, driving that agent ────────────
// Same component as step 2. Because you own it, you can attach your own
// observer; the bridge's reply router will be a peer listener, not a
// privileged owner.
const dispatcher = new Dispatcher({
  agent,
  maxConcurrency: 1, // replies stay in order
});
dispatcher.addSessionListener((sessionId, event) => {
  if (event.type !== AgentEventType.Message) return;
  if (event.message.role !== Role.Assistant) return;
  console.log(`[you]        observed reply for session "${sessionId}"`);
});

// ── The wiring: ChannelBridge connects the source to YOUR dispatcher ────────
// Inbound: each InboundMessage maps to a sessionId (one per thread) and is
// submitted. Outbound: the run's events route back through the bridge to
// source.send, one coalesced reply per assistant turn.
const socket = new FakeSocket();
const bridge = new ChannelBridge({ source: new FakeSocketSource(socket), dispatcher });
await bridge.start();

// ── Events through the whole path ───────────────────────────────────────────
// Two messages on one thread (same sessionId → memory accumulates, "turn 2"),
// one on another thread (a fresh session, back to "turn 1").
socket.push({ channel: "#general", thread: "t1", user: "mike", body: "hello" });
await sleep(250);
socket.push({ channel: "#general", thread: "t1", user: "mike", body: "still there?" });
await sleep(250);
socket.push({ channel: "#general", thread: "t2", user: "ana", body: "new thread" });
await sleep(250);

await bridge.stop();
bun run examples/channels-tutorial/step3.ts
[socket]     connected
[source]     frame → InboundMessage → bridge: "hello"
[you]        observed reply for session "#general:t1"
[socket]     ⇦ reply posted to #general/t1: "re: hello (turn 1)"
[source]     frame → InboundMessage → bridge: "still there?"
[you]        observed reply for session "#general:t1"
[socket]     ⇦ reply posted to #general/t1: "re: still there? (turn 2)"
[source]     frame → InboundMessage → bridge: "new thread"
[you]        observed reply for session "#general:t2"
[socket]     ⇦ reply posted to #general/t2: "re: new thread (turn 1)"
[socket]     disconnected

Read the log against the diagram, hop by hop:

  • [socket] → [source] — the provider pushes a frame; the source normalizes it and hands it to the bridge's callback, returning immediately. The source never blocks the socket; absorbing bursts is the queue's job.
  • bridge → dispatcher → runAgent — the bridge maps the message to sessionId = "channelId:threadId" and submits — exactly what you did by hand in step 2, now computed from the thread.
  • [you] and [socket] ⇦ — the run's events fan out to every session listener: your observer logs the session, and the bridge coalesces the text into one reply per assistant turn and posts it to the originating thread.

The turn counts carry the same lesson as step 1: two messages on t1 are one session (turn 2); the message on t2 is a fresh drawer (turn 1). Nothing was passed between runs — the thread → sessionId mapping is the whole mechanism.

Three ways to hand the bridge its agent

new ChannelBridge({ source, … }) accepts exactly one of: dispatcher (a caller-owned Dispatcher, as above — full control, attach your own listeners), agent (a self-contained Agent; the bridge builds a dispatcher around it — steps 4 and 5 use this), or base (shared runAgent config, the original shorthand). The runtime path is identical in all three: construction-time wiring only.

Step 4: start and stop listening

The bridge is the on/off switch for the whole channel. bridge.start() connects the source and messages begin flowing; bridge.stop() disconnects it, and events on the wire go nowhere — exactly what a closed websocket does. Neither touches the agent: its memory lives in the agent's closure, not in any connection, so a stop/start cycle resumes every conversation where it left off:

examples/channels-tutorial/step4.ts
import {
  assistantMessage,
  ChannelBridge,
  contentToText,
  Role,
  runAgent,
  SessionMemoryStore,
  StreamEventType,
} from "@open-agent-loops/core";
import type {
  Agent,
  ChannelSource,
  InboundMessage,
  ModelClient,
  ModelRequest,
  OutboundTarget,
} from "@open-agent-loops/core";

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

// The same fake wire as step 2, trimmed to what this step teaches: when the
// source is stopped there is no listener, so pushed frames simply drop —
// exactly what a closed websocket does.
type SocketFrame = { channel: string; thread: string; user: string; body: string };

class FakeSocket {
  private listener: ((frame: SocketFrame) => void) | undefined;

  connect(onFrame: (frame: SocketFrame) => void): void {
    this.listener = onFrame;
  }

  disconnect(): void {
    this.listener = undefined;
  }

  push(frame: SocketFrame): void {
    if (!this.listener) {
      console.log(`[socket]   "${frame.body}" pushed — nobody listening, dropped`);
      return;
    }
    this.listener(frame);
  }

  post(channel: string, thread: string, text: string): void {
    console.log(`[socket]   ⇦ reply to ${channel}/${thread}: "${text}"`);
  }
}

class FakeSocketSource implements ChannelSource {
  constructor(private readonly socket: FakeSocket) {}

  start(onMessage: (message: InboundMessage) => void): void {
    this.socket.connect((frame) => {
      onMessage({ channelId: frame.channel, threadId: frame.thread, userId: frame.user, text: frame.body });
    });
    console.log("[source]   listening");
  }

  send(target: OutboundTarget, text: string): void {
    this.socket.post(target.channelId, target.threadId ?? "", text);
  }

  stop(): void {
    this.socket.disconnect();
    console.log("[source]   stopped");
  }
}

const echoModel: ModelClient = {
  async *stream(request: ModelRequest) {
    await sleep(50);
    const turns = request.messages.filter((m) => m.role === Role.User);
    const reply = `re: ${contentToText(turns[turns.length - 1]?.content ?? [])} (turn ${turns.length})`;
    yield { type: StreamEventType.Done, message: assistantMessage({ content: reply }) };
  },
};

// The agent, as a top-level value (same shape as steps 2 and 3). Its memory
// lives HERE, in the closure — not in any connection — which is what makes
// the stop/start cycle below resume conversations.
const memory = new SessionMemoryStore();
const agent: Agent = (call) => runAgent({ model: echoModel, memory, ...call });

const socket = new FakeSocket();
const bridge = new ChannelBridge({
  source: new FakeSocketSource(socket),
  agent,
  maxConcurrency: 1,
});

// ── 1. Start listening: messages flow socket → agent → back ────────────────
await bridge.start();
socket.push({ channel: "#general", thread: "t1", user: "mike", body: "hello" });
await sleep(150);

// ── 2. Stop listening: the wire is dead, pushes go nowhere ─────────────────
await bridge.stop();
socket.push({ channel: "#general", thread: "t1", user: "mike", body: "anyone home?" });
await sleep(150);

// ── 3. Start again: same bridge, same memory — the conversation resumes ────
// The reply says "turn 2": memory for "#general:t1" survived the stop, because
// it lives in the SessionMemoryStore, not in the connection.
await bridge.start();
socket.push({ channel: "#general", thread: "t1", user: "mike", body: "back again" });
await sleep(150);

await bridge.stop();
bun run examples/channels-tutorial/step4.ts
[source]   listening
[socket]   ⇦ reply to #general/t1: "re: hello (turn 1)"
[source]   stopped
[socket]   "anyone home?" pushed — nobody listening, dropped
[source]   listening
[socket]   ⇦ reply to #general/t1: "re: back again (turn 2)"
[source]   stopped

Two beats to notice. While stopped, the pushed message is dropped, not queued — the bounded queue only holds messages that arrived while listening; a dead wire is the provider's side (a real source would resume from a cursor on reconnect). And after the restart the reply says turn 2: memory never lived in the connection.

Step 5: backpressure — when the socket is faster than the model

Everything so far sent messages politely, one at a time. A real channel doesn't. This step gives the model latency and fires a burst of ten messages at one thread, faster than a run can even start — and watches the bounded queue do its job:

examples/channels-tutorial/step5.ts
import {
  assistantMessage,
  ChannelBridge,
  contentToText,
  InMemoryChannelSource,
  Role,
  runAgent,
  SessionMemoryStore,
  StreamEventType,
} from "@open-agent-loops/core";
import type { Agent, ModelClient, ModelRequest } from "@open-agent-loops/core";

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

// A stand-in model with latency: inbound messages pile up while a run is in
// flight, which is what makes backpressure observable. It echoes the (coalesced)
// user turns it actually received, so you can see which messages survived.
const echoModel: ModelClient = {
  async *stream(request: ModelRequest) {
    await sleep(150);
    const received = request.messages
      .filter((m) => m.role === Role.User)
      .map((m) => contentToText(m.content))
      .join(" + ");
    const reply = `handled: ${received}`;
    for (const piece of reply.match(/.{1,12}/g) ?? []) {
      yield { type: StreamEventType.TextDelta, text: piece };
    }
    yield { type: StreamEventType.Done, message: assistantMessage({ content: reply }) };
  },
};

// The agent, as a top-level value — same shape as every step before it.
const memory = new SessionMemoryStore();
const agent: Agent = (call) => runAgent({ model: echoModel, memory, ...call });

// The transport. A real bot swaps in a Slack/Discord ChannelSource here — the
// rest of the wiring is identical, because the transport is just a seam.
const source = new InMemoryChannelSource();

// The bridge wires the transport to the agent through a bounded, coalescing queue.
const bridge = new ChannelBridge({
  source,
  agent,
  capacity: 4, // per-thread spam ceiling
  overflow: "drop-oldest", // shed the stalest message under a flood
  maxConcurrency: 2, // at most 2 runs across all threads
});
await bridge.start();

// A burst: 10 messages to one thread, faster than a run can even start. The
// bounded buffer keeps the last 4 and sheds the rest; the survivors coalesce
// into ONE run, and its reply is posted back to the originating thread.
for (let i = 1; i <= 10; i++) {
  source.emit({ channelId: "#general", threadId: "t1", userId: "u", text: `msg ${i}` });
}
console.log("right after the burst:", bridge.dispatcher.stats());

await sleep(400); // let the slow run finish
console.log("reply posted back:    ", source.sent.map((s) => s.text));
console.log("final stats:          ", bridge.dispatcher.stats());

await bridge.stop();
bun run examples/channels-tutorial/step5.ts
right after the burst: { sessions: 1, inFlight: 1, queued: 4, dropped: 6, highWater: 4 }
reply posted back:     [ "handled: msg 7 + msg 8 + msg 9 + msg 10" ]
final stats:           { sessions: 1, inFlight: 0, queued: 0, dropped: 6, highWater: 4 }

The buffer's capacity: 4 + drop-oldest keeps the last four and sheds six; the survivors coalesce into one run, whose single reply is posted back to the thread. The socket was never blocked: backpressure was applied at the queue, and bridge.dispatcher.stats() reports it (dropped, highWater) as it happens.

This is the load-bearing idea: bounded ≠ adaptive. The bounded queue keeps the system from falling over and makes the load measurable; tuning to that load (an AIMD controller reading highWater/dropped) is a separate layer you add on top.

Going live

Swapping in a real provider means replacing only the two fakes from step 3:

  • The FakeSocket becomes the real websocket — plus heartbeat and reconnect-with-backoff, which live inside your ChannelSource.
  • The echo model becomes an OpenAICompatibleModel({ apiKey, model, baseURL }).

The bridge, the dispatcher, and runAgent don't change.

The knobs

All forwarded to the dispatcher the bridge owns:

  • capacity — the per-thread spam ceiling.
  • overflow — what happens when a thread's buffer is full: drop-oldest / drop-newest / block (propagate real backpressure to a blockable producer) / { coalesce } (fold the arrival in).
  • maxConcurrency — the global cap on runs in flight across all threads, the protection for the provider rate limit.
  • supersede — abort the in-flight run when a newer message lands, instead of queueing behind it. Safe because runAgent persists the prompt to memory before its first abort check, so a superseded turn survives in history.
  • sessionIdFor — the thread → sessionId mapping (default: one session per thread). Override for per-channel or per-user grain.

The fuller demo (two threads, isolation under a flood) is at examples/channels/channels.ts; for the design and open questions, see agent-loop-core/docs/channels.md.

Agent as a Tool

A short tutorial — wrap an agent as a tool another agent can call, then build a single-chat orchestrator that routes between specialist sub-agents.

Bring Your Own Front End

A short guide — render the agent loop's event stream however you like.

On this page

Step 1: the agent loop, with no channel anythingStep 2: name the agent, and let the Dispatcher drive itStep 3: a fake socket, end to end — composed from top-level componentsStep 4: start and stop listeningStep 5: backpressure — when the socket is faster than the modelGoing liveThe knobs