Gateway

OverviewCopy Anchor Link

The Gateway is the central class of @cogitator-ai/channels. It connects messaging channels (Telegram, Discord, Slack, WhatsApp, WebChat) to your AI agents, handling sessions, middleware, streaming, and message formatting.

pnpm add @cogitator-ai/channels

Quick StartCopy Anchor Link

import { Cogitator, Agent } from '@cogitator-ai/core';
import { Gateway, telegramChannel } from '@cogitator-ai/channels';

const agent = new Agent({
  name: 'assistant',
  model: 'anthropic/claude-sonnet-5-5',
  instructions: 'You are a helpful personal assistant.',
});

const gateway = new Gateway({
  agent,
  cogitator: new Cogitator({
    llm: {
      providers: { anthropic: { apiKey: process.env.ANTHROPIC_API_KEY } },
    },
  }),
  channels: [telegramChannel({ token: process.env.TG_TOKEN! })],
});

await gateway.start();

GatewayConfigCopy Anchor Link

interface GatewayConfig {
  agent: Agent | ((user: ChannelUser) => Agent | Promise<Agent>);
  channels: Channel[];
  cogitator: Cogitator;
  memory?: MemoryAdapter;
  sessionManager?: SessionManager; // defaults to SessionManager(memory)
  middleware?: GatewayMiddleware[];
  hooks?: HookRegistry;
  mediaProcessor?: MediaProcessor;
  runTimeout?: number; // ms, defaults to the agent's timeout

  session?: {
    threadKey?: (msg: ChannelMessage) => string;
    compaction?: CompactionConfig;
  };

  stream?: StreamConfig;
  reactions?: StatusReactionConfig;
  debounce?: DebounceConfig;
  envelope?: EnvelopeConfig;
  queueMode?: 'parallel' | 'sequential' | 'interrupt' | 'collect';
  approvals?: {
    format?: (approvals: readonly ToolApprovalRequest[], words: ApprovalReplyWords) => string;
    approveWords?: readonly string[]; // default: approve, yes, да, одобряю
    denyWords?: readonly string[]; // default: deny, no, нет, отклоняю
    notAllowedMessage?: string;
  };

  onError?: (error: Error, msg: ChannelMessage) => void;
}

GatewayConfig.owner is deprecated and never read: configure owners on the middleware that uses them, ownerCommands({ ownerIds }) and dmPolicy({ ownerIds }) (see Middleware).

Without memory (and without sessionManager) the Gateway keeps no history: every run starts from an empty thread. onError receives failures of runs, middleware and compaction; the message is dropped silently when it is not set.

Per-user AgentsCopy Anchor Link

Pass a factory function to create different agents per user:

const gateway = new Gateway({
  agent: (user) => {
    if (user.id === 'admin') {
      return adminAgent;
    }
    return defaultAgent;
  },
  // ...
});

Message FlowCopy Anchor Link

User sends message
  → Channel converts to ChannelMessage
  → Debouncer (if enabled): buffer rapid messages, merge into one
  → MessageQueue (if queueMode is not 'parallel'): sequential/interrupt/collect per thread
  → Middleware pipeline (dm-policy → owner-commands → rate-limit → ...)
  → message:received hook
  → Agent resolved (factory called per message), session lookup/create (emits session:created)
  → Compaction (if the thread holds ≥ threshold messages)
  → Status reaction 👀 → 🤔, typing indicator every 4s
  → Approve/deny reply on a paused thread? cogitator.resume(agent, threadId, { defaultDecision })
    otherwise: envelope (if enabled), media processing, cogitator.run(agent, { input, threadId, ... })
  → 🔥 while tools run, back to 🤔 after each result
  → StreamBuffer flushes progressively via sendText/editText (or plain chunked reply)
  → Platform-specific markdown adaptation
  → Response delivered, status reaction 👍 (or 😱 on failure)
  → Run paused for approval? Prompt listing the waiting tool calls

Runs use threadId from session.threadKey (default `${channelType}:${userId}`), threadAccess: 'shared', and pass userId, channelType and channelId to the run, so tools see them in their context. Scheduled messages (raw.scheduled === true, see Scheduler) are answered without quoting the trigger message.

Status ReactionsCopy Anchor Link

Show processing progress with emoji reactions on the user's message. Users see immediate feedback instead of waiting 10-30 seconds in silence.

const gateway = new Gateway({
  // ...
  reactions: {
    enabled: true,
    emojis: {
      queued: '👀', // message received
      thinking: '🤔', // LLM is generating
      tool: '🔥', // tool execution
      done: '👍', // response sent
      error: '😱', // something failed
    },
    debounceMs: 700, // skip rapid phase changes
    stallSoftMs: 10000, // show 🥱 after 10s on same phase
    stallHardMs: 30000, // show 😨 after 30s
  },
});

Supported channels: Telegram (setMessageReaction), Discord (react), Slack (reactions.add, default emojis mapped to Slack names). Channels without setReaction silently skip reactions.

Inbound DebouncingCopy Anchor Link

When a user sends 3 messages in 1 second, without debouncing you get 3 parallel LLM calls. With debouncing, they merge into one.

const gateway = new Gateway({
  // ...
  debounce: {
    enabled: true,
    delayMs: 1500, // wait 1.5s of silence before processing
    byChannel: { discord: 2000 }, // per-channel override
  },
});

Messages are grouped by channelType:channelId:userId. Texts are joined with newlines, attachments are merged. Merged messages then go through queueMode like any other message.

Envelope FormattingCopy Anchor Link

Give the LLM context about who is talking, from where, and when — without polluting agent instructions.

const gateway = new Gateway({
  // ...
  envelope: {
    enabled: true,
    includeTimestamp: true,
    includeElapsed: true,
    timezone: 'utc', // 'utc' | 'local' | IANA timezone string
  },
});

The user's message Hello world becomes:

[Feb 28, 20:15 | telegram | Alice | +2m30s] Hello world

Use includeChatType: true to add DM/group, and includeSender/includeChannel: false to drop those parts. Invalid time zones fall back to UTC.

The elapsed time shows how long since the last message in this thread. The LLM naturally picks up on timing and context without explicit instructions.

Queue ModesCopy Anchor Link

Control how concurrent messages from the same user/thread are handled.

const gateway = new Gateway({
  // ...
  queueMode: 'sequential',
});

For personal assistants, sequential or collect usually give the best UX. Queues are keyed by the same thread id as sessions.

Multi-channel SetupCopy Anchor Link

import {
  Gateway,
  telegramChannel,
  discordChannel,
  slackChannel,
  webchatChannel,
} from '@cogitator-ai/channels';

const gateway = new Gateway({
  agent,
  cogitator,
  channels: [
    telegramChannel({ token: process.env.TG_TOKEN! }),
    discordChannel({ token: process.env.DISCORD_TOKEN! }),
    slackChannel({
      token: process.env.SLACK_BOT_TOKEN!,
      signingSecret: process.env.SLACK_SIGNING_SECRET!,
      appToken: process.env.SLACK_APP_TOKEN!,
    }),
    webchatChannel({ port: 18789 }),
  ],
});

Platform ComparisonCopy Anchor Link

ApprovalsCopy Anchor Link

Tools marked requiresApproval (for example create_tool from capabilities.selfTools) pause the run before they execute — unless the Cogitator has guardrails.onToolApproval, which then decides instead. The Gateway turns that pause into a chat conversation:

1. The paused run's text (if any) is sent, followed by a prompt listing each waiting call — tool name, description and compact JSON arguments (cut at 300 characters).
2. The user replies approve / yes to run the calls, or deny / no to refuse, optionally followed by a reason (no, too risky) that the agent sees. Replies are case-insensitive and may end with punctuation; да / одобряю and нет / отклоняю work out of the box.
3. The Gateway calls cogitator.resume(agent, threadId, { userId, defaultDecision }) and delivers the result like any reply, streaming included. If the run pauses again, a new prompt is sent.

Any other message on a paused thread runs as usual, and the runtime answers the waiting calls as declined. An approve-only word with more text after it (yes, but rename it) counts as a new message, so nothing runs by accident.

Only the user who started the run can answer it: the runtime checks the run's userId on resume. In a shared thread (a threadKey per group chat), someone else replying approve gets notAllowedMessage and the pause stays put.

Pauses survive restarts when the runtime persists them (in the thread's memory once the Cogitator has a memory adapter, or in runCheckpoints): the first approve/deny reply on a thread the Gateway has not seen since it started tries resume, and runs as a normal message if nothing is paused. On threads it has already seen, the Gateway resumes only pauses it sent a prompt for, so one Gateway process should serve a given thread.

const gateway = new Gateway({
  // ...
  hooks,
  approvals: {
    approveWords: ['approve', 'yes', 'да'],
    denyWords: ['deny', 'no', 'нет'],
    format: (approvals, { approveWords, denyWords }) =>
      `Разрешить ${approvals.map((a) => a.toolName).join(', ')}? ` +
      `Ответьте «${approveWords[0]}» или «${denyWords[0]}».`,
    notAllowedMessage: 'Подтвердить может только автор запроса.',
  },
});

hooks.on('approval:requested', ({ threadId, approvals }) => {
  audit.log(
    threadId,
    approvals.map((a) => a.toolName)
  );
});
hooks.on('approval:resolved', ({ threadId, decision, superseded }) => {
  audit.log(threadId, decision.approved, superseded);
});

approval:resolved carries superseded: true when the user sent a new message instead of answering (the calls were declined), and has no approvals when the pause predates this process.

On channels with buttons (Telegram), the prompt also carries Approve and Deny buttons. A press answers like the reply words, through the same middleware and checks, and the buttons then show the decision. approvals: { buttons: false } turns them off, and buttonLabels renames them. A press counts only from the user the prompt asked, while their run is still paused; anyone else gets notAllowedMessage, and a press after the request was answered, dropped or the Gateway restarted gets expiredMessage (a reply word still works then).

parseApprovalReply(text, words) and formatApprovalPrompt(approvals, words) are exported for custom channels and UIs, together with DEFAULT_APPROVE_WORDS, DEFAULT_DENY_WORDS and DEFAULT_NOT_ALLOWED_MESSAGE.

With cogitator up, set approveWords, denyWords and notAllowedMessage in the approvals block of cogitator.yml; a custom format goes in RuntimeBuilder's approvals option, which is merged over the YAML block (see Wizard Setup).

Lifecycle HooksCopy Anchor Link

Hooks observe the pipeline without changing it. Create a registry with createHookRegistry() and pass it as hooks:

import { createHookRegistry, Gateway } from '@cogitator-ai/channels';

const hooks = createHookRegistry();

hooks.on('agent:after_run', ({ threadId, output }) => {
  console.log(`[${threadId}] ${output.length} chars`);
});
hooks.on('agent:error', (e) => console.error(`[${e.threadId}]`, e.error.message));

const gateway = new Gateway({ agent, cogitator, channels, hooks });

Handlers are typed by hook name: hooks.on(name, handler) gives the handler the payload type HookPayloads[name]. HookPayloads is exported from @cogitator-ai/channels and @cogitator-ai/types; the per-hook event types below come from @cogitator-ai/types (ApprovalRequestedEvent / ApprovalResolvedEvent also from @cogitator-ai/channels). A handler typed with an unknown payload (HookHandler) is still accepted for any hook.

Handlers run in registration order and are awaited, so a slow handler delays the reply; a throwing handler is logged and does not stop the others. message:sending / message:sent fire for replies when stream is off and for approval prompts — with stream on, the answer reports its messages through stream:finished. A thrown non-Error value is wrapped in an Error before it reaches agent:error. hooks.off(name, handler) removes a handler.

Media and Voice MessagesCopy Anchor Link

Channels deliver photos, voice notes, videos and files as attachments. Pass a MediaProcessor to turn them into model input:

import { Gateway, MediaProcessor, LocalWhisper, DeepgramSttProvider } from '@cogitator-ai/channels';
import { getModel } from '@cogitator-ai/models';

const mediaProcessor = new MediaProcessor(
  new LocalWhisper(), // offline fallback, used once its model is downloaded
  (modelId) => getModel(modelId)?.capabilities?.supportsVision ?? true,
  new DeepgramSttProvider({ apiKey: process.env.DEEPGRAM_API_KEY! }),
  { maxInlineTextChars: 20_000, maxDownloadBytes: 25 * 1024 * 1024 } // defaults
);

const gateway = new Gateway({ agent, cogitator, channels, mediaProcessor });

STT providers: DeepgramSttProvider (nova-3), GroqSttProvider (whisper-large-v3), OpenAISttProvider (gpt-transcribe) — each takes { apiKey, model? } (Deepgram also language) — or any { transcribe(buffer, mimeType) }. The provider is tried first; without one, LocalWhisper (Xenova whisper-tiny, cached in ~/.cogitator/models) transcribes once downloaded. createWhisperDownloadTool(whisper) gives the agent a download_stt_model tool that installs it on request. When nothing can transcribe, the agent is told so instead of failing. Attachments fetched by URL (Discord) are capped at maxDownloadBytes.

RuntimeBuilder wires this up automatically, picking DEEPGRAM_API_KEY, then GROQ_API_KEY, then OPENAI_API_KEY.

Session ManagementCopy Anchor Link

Gateway integrates with @cogitator-ai/memory for persistent conversations:

import { SQLiteAdapter } from '@cogitator-ai/memory';

const memory = new SQLiteAdapter({ provider: 'sqlite', path: './sessions.db' });
await memory.connect();

const gateway = new Gateway({
  // ...
  memory,
  session: {
    threadKey: (msg) => `${msg.channelType}:${msg.userId}`,
    compaction: {
      strategy: 'summary',
      threshold: 100, // messages in the thread before compaction runs
      keepRecent: 20, // newest messages kept verbatim
      summaryModel: 'google/gemini-3.8-flash', // optional, defaults to the agent model
    },
  },
});

The threadKey function determines how sessions are grouped. By default, each user gets their own session per channel.

Compaction summarizes everything except the newest keepRecent messages with an LLM call (summaryPrompt overrides the default prompt) and stores the summary as a system message. Strategies: summary, sliding-window (drop old messages), hybrid. A failed compaction is reported to onError and the message is still answered. Force it with await gateway.compactThread(threadId).

StatsCopy Anchor Link

const stats = gateway.stats;
// {
//   uptime: 7200000,
//   activeSessions: 3,
//   totalSessions: 12,
//   messagesToday: 847,
//   connectedChannels: ['telegram', 'discord'],
// }

gateway.getSessions();
// [{ threadId: 'telegram:42', channelType: 'telegram', userId: '42', userName: 'Ann',
//    messageCount: 18, lastActiveAt: 1767225600000, active: false }, ...]

activeSessions counts threads with a run in progress; messagesToday resets at local midnight.

Environment VariablesCopy Anchor Link

A typical .env for a multi-channel bot:

# LLM
GOOGLE_API_KEY=...

# Telegram
TG_TOKEN=7204891735:AAHr...

# Discord
DISCORD_TOKEN=MTI...

# Slack
SLACK_BOT_TOKEN=xoxb-...
SLACK_SIGNING_SECRET=...
SLACK_APP_TOKEN=xapp-...

# WebChat
WEBCHAT_TOKEN=your-secret-token

Running with the CLICopy Anchor Link

cogitator assistant runs a Gateway you build in code, with a live status view:

export const gateway = new Gateway({ agent, cogitator, channels });

cogitator assistant                 # loads .env into process.env, imports src/gateway.ts
cogitator assistant -c bot/gateway.ts --quiet

The module must export gateway; the command starts it, shows stats and a live log, and stops it on q or Ctrl+C. For a config-driven assistant use cogitator.yml and cogitator up instead.

Graceful ShutdownCopy Anchor Link

process.on('SIGTERM', async () => {
  await gateway.stop(); // flushes debouncer, disposes queue, stops channels
  process.exit(0);
});