feat: add channel adapter abstraction with Telegram and WebChat adapters
Implement Phase 3 channel adapters that decouple message sources from the agent via a uniform ChannelAdapter interface and ChannelRegistry. - Add ChannelAdapter/InboundMessage/OutboundMessage types - Add ChannelRegistry for adapter lifecycle and message routing - Add TelegramAdapter (grammy bot, auth middleware, confirmations, chunking) - Add WebChatAdapter (thin shim over GatewayServer) - Refactor daemon to use ChannelRegistry with per-channel-per-user agents - Add config.get/config.patch gateway handlers (Phase 2 loose end) - Add system.restart gateway handler (Phase 2 loose end) - Add implementation plans and design docs Tests: 225 passing (33 new channel adapter + gateway handler tests)
This commit is contained in:
William Valentin
@@ -0,0 +1,256 @@
|
||||
import { describe, it, expect, vi, beforeEach } from 'vitest';
|
||||
|
||||
// ── Mock grammy before importing adapter ──────────────────────────
|
||||
|
||||
const mockUse = vi.fn();
|
||||
const mockOn = vi.fn();
|
||||
const mockCommand = vi.fn();
|
||||
const mockStart = vi.fn();
|
||||
const mockStop = vi.fn();
|
||||
const mockSendMessage = vi.fn();
|
||||
|
||||
vi.mock('grammy', () => ({
|
||||
Bot: vi.fn().mockImplementation(() => ({
|
||||
use: mockUse,
|
||||
on: mockOn,
|
||||
command: mockCommand,
|
||||
start: mockStart,
|
||||
stop: mockStop,
|
||||
api: { sendMessage: mockSendMessage },
|
||||
})),
|
||||
}));
|
||||
|
||||
import { TelegramAdapter, type TelegramAdapterConfig } from './adapter.js';
|
||||
import type { InboundMessage } from '../types.js';
|
||||
|
||||
const baseConfig: TelegramAdapterConfig = {
|
||||
botToken: 'test-token-123',
|
||||
allowedChatIds: [100, 200],
|
||||
};
|
||||
|
||||
describe('TelegramAdapter', () => {
|
||||
let adapter: TelegramAdapter;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
adapter = new TelegramAdapter(baseConfig);
|
||||
});
|
||||
|
||||
// ── Basic properties ──────────────────────────────────────────
|
||||
|
||||
it('has name "telegram"', () => {
|
||||
expect(adapter.name).toBe('telegram');
|
||||
});
|
||||
|
||||
it('starts as disconnected', () => {
|
||||
expect(adapter.status).toBe('disconnected');
|
||||
});
|
||||
|
||||
// ── connect / disconnect ──────────────────────────────────────
|
||||
|
||||
it('connect creates a bot and sets status to connected', async () => {
|
||||
await adapter.connect();
|
||||
|
||||
expect(adapter.status).toBe('connected');
|
||||
// Bot constructor called with the token
|
||||
const { Bot } = await import('grammy');
|
||||
expect(Bot).toHaveBeenCalledWith('test-token-123');
|
||||
});
|
||||
|
||||
it('connect registers auth middleware, commands, and message handler', async () => {
|
||||
await adapter.connect();
|
||||
|
||||
// .use() for auth middleware
|
||||
expect(mockUse).toHaveBeenCalledTimes(1);
|
||||
// .command() for /start and /reset
|
||||
expect(mockCommand).toHaveBeenCalledTimes(2);
|
||||
expect(mockCommand.mock.calls[0][0]).toBe('start');
|
||||
expect(mockCommand.mock.calls[1][0]).toBe('reset');
|
||||
// .on('message:text', ...) for text handler
|
||||
expect(mockOn).toHaveBeenCalledWith('message:text', expect.any(Function));
|
||||
// .start() to begin long polling
|
||||
expect(mockStart).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('connect registers callback_query handler when hookEngine is provided', async () => {
|
||||
const hookEngine = { resolveConfirmation: vi.fn() };
|
||||
const adapterWithHooks = new TelegramAdapter({
|
||||
...baseConfig,
|
||||
hookEngine: hookEngine as never,
|
||||
});
|
||||
|
||||
await adapterWithHooks.connect();
|
||||
|
||||
// Should have .on('callback_query:data', ...) plus .on('message:text', ...)
|
||||
expect(mockOn).toHaveBeenCalledWith('callback_query:data', expect.any(Function));
|
||||
expect(mockOn).toHaveBeenCalledWith('message:text', expect.any(Function));
|
||||
});
|
||||
|
||||
it('disconnect stops the bot and sets status to disconnected', async () => {
|
||||
await adapter.connect();
|
||||
expect(adapter.status).toBe('connected');
|
||||
|
||||
await adapter.disconnect();
|
||||
expect(mockStop).toHaveBeenCalledTimes(1);
|
||||
expect(adapter.status).toBe('disconnected');
|
||||
});
|
||||
|
||||
it('disconnect is safe to call when not connected', async () => {
|
||||
await adapter.disconnect();
|
||||
expect(adapter.status).toBe('disconnected');
|
||||
expect(mockStop).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── send ──────────────────────────────────────────────────────
|
||||
|
||||
it('send throws when adapter is not connected', async () => {
|
||||
await expect(adapter.send('100', { text: 'hello' })).rejects.toThrow(
|
||||
'Telegram adapter not connected',
|
||||
);
|
||||
});
|
||||
|
||||
it('send delivers a short message in a single API call', async () => {
|
||||
await adapter.connect();
|
||||
|
||||
await adapter.send('100', { text: 'Hello there' });
|
||||
|
||||
expect(mockSendMessage).toHaveBeenCalledTimes(1);
|
||||
expect(mockSendMessage).toHaveBeenCalledWith(100, 'Hello there', { parse_mode: 'Markdown' });
|
||||
});
|
||||
|
||||
it('send chunks a long message that exceeds 4096 chars', async () => {
|
||||
await adapter.connect();
|
||||
|
||||
// Create a message that is longer than 4096 chars — two halves joined by a newline
|
||||
const half = 'A'.repeat(3000);
|
||||
const longMessage = `${half}\n${'B'.repeat(3000)}`;
|
||||
|
||||
await adapter.send('200', { text: longMessage });
|
||||
|
||||
// Should have been split into 2 chunks
|
||||
expect(mockSendMessage.mock.calls.length).toBeGreaterThanOrEqual(2);
|
||||
// Each call uses numeric chatId and parse_mode
|
||||
for (const call of mockSendMessage.mock.calls) {
|
||||
expect(call[0]).toBe(200);
|
||||
expect(call[2]).toEqual({ parse_mode: 'Markdown' });
|
||||
}
|
||||
});
|
||||
|
||||
// ── onMessage / inbound handling ──────────────────────────────
|
||||
|
||||
it('onMessage registers a handler that receives text messages', async () => {
|
||||
const handler = vi.fn();
|
||||
adapter.onMessage(handler);
|
||||
|
||||
await adapter.connect();
|
||||
|
||||
// Get the registered message:text handler from mockOn
|
||||
const textHandlerCall = mockOn.mock.calls.find(
|
||||
(call) => call[0] === 'message:text',
|
||||
);
|
||||
expect(textHandlerCall).toBeDefined();
|
||||
|
||||
const textHandler = textHandlerCall![1];
|
||||
|
||||
// Simulate a grammy context object
|
||||
const ctx = {
|
||||
message: { message_id: 42, text: 'Hello Flynn' },
|
||||
chat: { id: 100 },
|
||||
from: { first_name: 'Will' },
|
||||
replyWithChatAction: vi.fn(),
|
||||
};
|
||||
|
||||
await textHandler(ctx);
|
||||
|
||||
expect(ctx.replyWithChatAction).toHaveBeenCalledWith('typing');
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
|
||||
const msg: InboundMessage = handler.mock.calls[0][0];
|
||||
expect(msg.channel).toBe('telegram');
|
||||
expect(msg.senderId).toBe('100');
|
||||
expect(msg.senderName).toBe('Will');
|
||||
expect(msg.text).toBe('Hello Flynn');
|
||||
expect(msg.id).toBe('42');
|
||||
});
|
||||
|
||||
it('text handler does nothing when no message handler is registered', async () => {
|
||||
// Don't call onMessage — no handler
|
||||
await adapter.connect();
|
||||
|
||||
const textHandlerCall = mockOn.mock.calls.find(
|
||||
(call) => call[0] === 'message:text',
|
||||
);
|
||||
const textHandler = textHandlerCall![1];
|
||||
|
||||
const ctx = {
|
||||
message: { message_id: 1, text: 'test' },
|
||||
chat: { id: 100 },
|
||||
from: { first_name: 'Will' },
|
||||
replyWithChatAction: vi.fn(),
|
||||
};
|
||||
|
||||
// Should not throw
|
||||
await textHandler(ctx);
|
||||
expect(ctx.replyWithChatAction).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── /reset command ────────────────────────────────────────────
|
||||
|
||||
it('/reset command delivers a reset inbound message', async () => {
|
||||
const handler = vi.fn();
|
||||
adapter.onMessage(handler);
|
||||
|
||||
await adapter.connect();
|
||||
|
||||
// Find the /reset command handler
|
||||
const resetCall = mockCommand.mock.calls.find((call) => call[0] === 'reset');
|
||||
expect(resetCall).toBeDefined();
|
||||
|
||||
const resetHandler = resetCall![1];
|
||||
|
||||
const ctx = {
|
||||
message: { message_id: 99 },
|
||||
chat: { id: 100 },
|
||||
from: { first_name: 'Will' },
|
||||
reply: vi.fn(),
|
||||
};
|
||||
|
||||
await resetHandler(ctx);
|
||||
|
||||
expect(ctx.reply).toHaveBeenCalledWith('Conversation reset.');
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
|
||||
const msg: InboundMessage = handler.mock.calls[0][0];
|
||||
expect(msg.text).toBe('/reset');
|
||||
expect(msg.metadata).toEqual({ isCommand: true, command: 'reset' });
|
||||
});
|
||||
|
||||
// ── Auth middleware ───────────────────────────────────────────
|
||||
|
||||
it('auth middleware blocks unauthorized chat IDs', async () => {
|
||||
await adapter.connect();
|
||||
|
||||
// The first .use() call is the auth middleware
|
||||
const authMiddleware = mockUse.mock.calls[0][0];
|
||||
|
||||
const next = vi.fn();
|
||||
const ctx = { chat: { id: 999 } }; // Not in allowedChatIds
|
||||
|
||||
await authMiddleware(ctx, next);
|
||||
|
||||
expect(next).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('auth middleware allows authorized chat IDs', async () => {
|
||||
await adapter.connect();
|
||||
|
||||
const authMiddleware = mockUse.mock.calls[0][0];
|
||||
|
||||
const next = vi.fn();
|
||||
const ctx = { chat: { id: 100 } }; // In allowedChatIds
|
||||
|
||||
await authMiddleware(ctx, next);
|
||||
|
||||
expect(next).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,208 @@
|
||||
import { Bot } from 'grammy';
|
||||
|
||||
import type { HookEngine } from '../../hooks/index.js';
|
||||
import type {
|
||||
InboundMessage,
|
||||
OutboundMessage,
|
||||
ChannelAdapter,
|
||||
ChannelStatus,
|
||||
} from '../types.js';
|
||||
import { isAllowedChat } from '../../frontends/telegram/handlers.js';
|
||||
import { parseConfirmationCallback } from '../../frontends/telegram/confirmations.js';
|
||||
|
||||
/** Configuration for the Telegram channel adapter. */
|
||||
export interface TelegramAdapterConfig {
|
||||
botToken: string;
|
||||
allowedChatIds: number[];
|
||||
hookEngine?: HookEngine;
|
||||
}
|
||||
|
||||
/**
|
||||
* Split a long message into chunks that respect Telegram's 4096 char limit.
|
||||
* Prefers splitting at newlines, then spaces, then hard-cuts.
|
||||
*/
|
||||
function splitMessage(text: string, maxLength: number): string[] {
|
||||
const chunks: string[] = [];
|
||||
let remaining = text;
|
||||
|
||||
while (remaining.length > 0) {
|
||||
if (remaining.length <= maxLength) {
|
||||
chunks.push(remaining);
|
||||
break;
|
||||
}
|
||||
|
||||
// Try to split at a newline within the allowed window
|
||||
let splitIndex = remaining.lastIndexOf('\n', maxLength);
|
||||
if (splitIndex === -1 || splitIndex < maxLength / 2) {
|
||||
splitIndex = remaining.lastIndexOf(' ', maxLength);
|
||||
}
|
||||
if (splitIndex === -1 || splitIndex < maxLength / 2) {
|
||||
splitIndex = maxLength;
|
||||
}
|
||||
|
||||
chunks.push(remaining.slice(0, splitIndex));
|
||||
remaining = remaining.slice(splitIndex).trimStart();
|
||||
}
|
||||
|
||||
return chunks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Telegram channel adapter backed by grammy.
|
||||
*
|
||||
* Handles authentication via allowed-chat-id filtering,
|
||||
* confirmation callbacks (when a HookEngine is provided),
|
||||
* and message chunking for Telegram's 4096-char limit.
|
||||
*/
|
||||
export class TelegramAdapter implements ChannelAdapter {
|
||||
readonly name = 'telegram';
|
||||
|
||||
private _status: ChannelStatus = 'disconnected';
|
||||
private bot: Bot | null = null;
|
||||
private messageHandler?: (msg: InboundMessage) => void;
|
||||
private config: TelegramAdapterConfig;
|
||||
|
||||
get status(): ChannelStatus {
|
||||
return this._status;
|
||||
}
|
||||
|
||||
constructor(config: TelegramAdapterConfig) {
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
/** Register the inbound message handler. Called by the registry before connect(). */
|
||||
onMessage(handler: (msg: InboundMessage) => void): void {
|
||||
this.messageHandler = handler;
|
||||
}
|
||||
|
||||
/** Create the grammy bot, wire up middleware & handlers, and start long-polling. */
|
||||
async connect(): Promise<void> {
|
||||
this.bot = new Bot(this.config.botToken);
|
||||
this._status = 'connecting';
|
||||
|
||||
// ── Auth middleware — reject messages from unknown chats ──
|
||||
this.bot.use(async (ctx, next) => {
|
||||
const chatId = ctx.chat?.id;
|
||||
if (chatId === undefined || !isAllowedChat(chatId, this.config.allowedChatIds)) {
|
||||
console.log(`Rejected message from unauthorized chat: ${chatId}`);
|
||||
return;
|
||||
}
|
||||
await next();
|
||||
});
|
||||
|
||||
// ── Confirmation callback handler (requires hookEngine) ──
|
||||
if (this.config.hookEngine) {
|
||||
const hookEngine = this.config.hookEngine;
|
||||
|
||||
this.bot.on('callback_query:data', async (ctx) => {
|
||||
const data = ctx.callbackQuery.data;
|
||||
const parsed = parseConfirmationCallback(data);
|
||||
|
||||
if (!parsed) {
|
||||
await ctx.answerCallbackQuery({ text: 'Invalid action' });
|
||||
return;
|
||||
}
|
||||
|
||||
const resolved = hookEngine.resolveConfirmation(parsed.id, {
|
||||
approved: parsed.approved,
|
||||
reason: parsed.approved ? undefined : 'Denied by user',
|
||||
});
|
||||
|
||||
if (resolved) {
|
||||
await ctx.answerCallbackQuery({
|
||||
text: parsed.approved ? '✅ Approved' : '❌ Denied',
|
||||
});
|
||||
await ctx.editMessageText(
|
||||
ctx.callbackQuery.message?.text + `\n\n${parsed.approved ? '✅ Approved' : '❌ Denied'}`,
|
||||
{ parse_mode: 'Markdown' },
|
||||
);
|
||||
} else {
|
||||
await ctx.answerCallbackQuery({ text: 'Confirmation expired or not found' });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// ── Command handlers ──
|
||||
|
||||
this.bot.command('start', async (ctx) => {
|
||||
await ctx.reply('Flynn is ready. Send me a message!');
|
||||
});
|
||||
|
||||
this.bot.command('reset', async (ctx) => {
|
||||
// Deliver a special reset message through the channel
|
||||
if (this.messageHandler) {
|
||||
this.messageHandler({
|
||||
id: String(ctx.message?.message_id ?? Date.now()),
|
||||
channel: 'telegram',
|
||||
senderId: String(ctx.chat.id),
|
||||
senderName: ctx.from?.first_name,
|
||||
text: '/reset',
|
||||
timestamp: Date.now(),
|
||||
metadata: { isCommand: true, command: 'reset' },
|
||||
});
|
||||
}
|
||||
await ctx.reply('Conversation reset.');
|
||||
});
|
||||
|
||||
// ── Text message handler ──
|
||||
|
||||
this.bot.on('message:text', async (ctx) => {
|
||||
if (!this.messageHandler) return;
|
||||
|
||||
const text = ctx.message.text;
|
||||
|
||||
// Show typing indicator while processing
|
||||
await ctx.replyWithChatAction('typing');
|
||||
|
||||
this.messageHandler({
|
||||
id: String(ctx.message.message_id),
|
||||
channel: 'telegram',
|
||||
senderId: String(ctx.chat.id),
|
||||
senderName: ctx.from?.first_name,
|
||||
text,
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
});
|
||||
|
||||
// ── Start long polling ──
|
||||
|
||||
this.bot.start({
|
||||
onStart: (botInfo) => {
|
||||
console.log(`Telegram bot started: @${botInfo.username}`);
|
||||
this._status = 'connected';
|
||||
},
|
||||
});
|
||||
|
||||
// bot.start() returns immediately for long polling.
|
||||
// The onStart callback sets connected above; also set here for safety
|
||||
// in case the callback fires before this line is reached.
|
||||
this._status = 'connected';
|
||||
}
|
||||
|
||||
/** Stop the bot and clean up. */
|
||||
async disconnect(): Promise<void> {
|
||||
if (this.bot) {
|
||||
await this.bot.stop();
|
||||
this.bot = null;
|
||||
}
|
||||
this._status = 'disconnected';
|
||||
}
|
||||
|
||||
/** Send an outbound message, automatically chunking if it exceeds Telegram's limit. */
|
||||
async send(peerId: string, message: OutboundMessage): Promise<void> {
|
||||
if (!this.bot) throw new Error('Telegram adapter not connected');
|
||||
|
||||
const chatId = Number(peerId);
|
||||
const text = message.text;
|
||||
|
||||
// Telegram enforces a 4096-character limit per message
|
||||
if (text.length <= 4096) {
|
||||
await this.bot.api.sendMessage(chatId, text, { parse_mode: 'Markdown' });
|
||||
} else {
|
||||
const chunks = splitMessage(text, 4096);
|
||||
for (const chunk of chunks) {
|
||||
await this.bot.api.sendMessage(chatId, chunk, { parse_mode: 'Markdown' });
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
export { TelegramAdapter, type TelegramAdapterConfig } from './adapter.js';
|
||||
Reference in New Issue
Block a user