feat: add Codex CLI support, dual Discord bots, and Discord typing fix

- Add Codex as alternative agent type (agentType: 'codex' per group)
- Codex container runner with session resume and 100-turn limit
- Dual Discord bot support (DISCORD_BOT_TOKEN + DISCORD_CODEX_BOT_TOKEN)
- Each bot handles only its agentType's registered groups
- Discord typing indicator refreshes every 8s (was single-shot, expired after ~10s)
- Agent runner 100-turn limit to prevent infinite bot-to-bot loops
- Apple Container runtime support (convert-to-apple-container skill)
- Session commands (/compact) support
- DB migration for agent_type column
This commit is contained in:
Eyejoker
2026-03-10 20:45:48 +09:00
parent d572bab5c6
commit 4a34a9c802
24 changed files with 2606 additions and 176 deletions

View File

@@ -0,0 +1,781 @@
import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest';
// --- Mocks ---
// Mock registry (registerChannel runs at import time)
vi.mock('./registry.js', () => ({ registerChannel: vi.fn() }));
// Mock env reader (used by the factory, not needed in unit tests)
vi.mock('../env.js', () => ({ readEnvFile: vi.fn(() => ({})) }));
// Mock config
vi.mock('../config.js', () => ({
ASSISTANT_NAME: 'Andy',
TRIGGER_PATTERN: /^@Andy\b/i,
}));
// Mock logger
vi.mock('../logger.js', () => ({
logger: {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
},
}));
// --- discord.js mock ---
type Handler = (...args: any[]) => any;
const clientRef = vi.hoisted(() => ({ current: null as any }));
vi.mock('discord.js', () => {
const Events = {
MessageCreate: 'messageCreate',
ClientReady: 'ready',
Error: 'error',
};
const GatewayIntentBits = {
Guilds: 1,
GuildMessages: 2,
MessageContent: 4,
DirectMessages: 8,
};
class MockClient {
eventHandlers = new Map<string, Handler[]>();
user: any = { id: '999888777', tag: 'Andy#1234' };
private _ready = false;
constructor(_opts: any) {
clientRef.current = this;
}
on(event: string, handler: Handler) {
const existing = this.eventHandlers.get(event) || [];
existing.push(handler);
this.eventHandlers.set(event, existing);
return this;
}
once(event: string, handler: Handler) {
return this.on(event, handler);
}
async login(_token: string) {
this._ready = true;
// Fire the ready event
const readyHandlers = this.eventHandlers.get('ready') || [];
for (const h of readyHandlers) {
h({ user: this.user });
}
}
isReady() {
return this._ready;
}
channels = {
fetch: vi.fn().mockResolvedValue({
send: vi.fn().mockResolvedValue(undefined),
sendTyping: vi.fn().mockResolvedValue(undefined),
}),
};
destroy() {
this._ready = false;
}
}
// Mock TextChannel type
class TextChannel {}
return {
Client: MockClient,
Events,
GatewayIntentBits,
TextChannel,
};
});
import { DiscordChannel, DiscordChannelOpts } from './discord.js';
// --- Test helpers ---
function createTestOpts(
overrides?: Partial<DiscordChannelOpts>,
): DiscordChannelOpts {
return {
onMessage: vi.fn(),
onChatMetadata: vi.fn(),
registeredGroups: vi.fn(() => ({
'dc:1234567890123456': {
name: 'Test Server #general',
folder: 'test-server',
trigger: '@Andy',
added_at: '2024-01-01T00:00:00.000Z',
},
})),
...overrides,
};
}
function createMessage(overrides: {
channelId?: string;
content?: string;
authorId?: string;
authorUsername?: string;
authorDisplayName?: string;
memberDisplayName?: string;
isBot?: boolean;
guildName?: string;
channelName?: string;
messageId?: string;
createdAt?: Date;
attachments?: Map<string, any>;
reference?: { messageId?: string };
mentionsBotId?: boolean;
}) {
const channelId = overrides.channelId ?? '1234567890123456';
const authorId = overrides.authorId ?? '55512345';
const botId = '999888777'; // matches mock client user id
const mentionsMap = new Map();
if (overrides.mentionsBotId) {
mentionsMap.set(botId, { id: botId });
}
return {
channelId,
id: overrides.messageId ?? 'msg_001',
content: overrides.content ?? 'Hello everyone',
createdAt: overrides.createdAt ?? new Date('2024-01-01T00:00:00.000Z'),
author: {
id: authorId,
username: overrides.authorUsername ?? 'alice',
displayName: overrides.authorDisplayName ?? 'Alice',
bot: overrides.isBot ?? false,
},
member: overrides.memberDisplayName
? { displayName: overrides.memberDisplayName }
: null,
guild: overrides.guildName
? { name: overrides.guildName }
: null,
channel: {
name: overrides.channelName ?? 'general',
messages: {
fetch: vi.fn().mockResolvedValue({
author: { username: 'Bob', displayName: 'Bob' },
member: { displayName: 'Bob' },
}),
},
},
mentions: {
users: mentionsMap,
},
attachments: overrides.attachments ?? new Map(),
reference: overrides.reference ?? null,
};
}
function currentClient() {
return clientRef.current;
}
async function triggerMessage(message: any) {
const handlers = currentClient().eventHandlers.get('messageCreate') || [];
for (const h of handlers) await h(message);
}
// --- Tests ---
describe('DiscordChannel', () => {
beforeEach(() => {
vi.clearAllMocks();
});
afterEach(() => {
vi.restoreAllMocks();
});
// --- Connection lifecycle ---
describe('connection lifecycle', () => {
it('resolves connect() when client is ready', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
expect(channel.isConnected()).toBe(true);
});
it('registers message handlers on connect', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
expect(currentClient().eventHandlers.has('messageCreate')).toBe(true);
expect(currentClient().eventHandlers.has('error')).toBe(true);
expect(currentClient().eventHandlers.has('ready')).toBe(true);
});
it('disconnects cleanly', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
expect(channel.isConnected()).toBe(true);
await channel.disconnect();
expect(channel.isConnected()).toBe(false);
});
it('isConnected() returns false before connect', () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
expect(channel.isConnected()).toBe(false);
});
});
// --- Text message handling ---
describe('text message handling', () => {
it('delivers message for registered channel', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'Hello everyone',
guildName: 'Test Server',
channelName: 'general',
});
await triggerMessage(msg);
expect(opts.onChatMetadata).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.any(String),
'Test Server #general',
'discord',
true,
);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
id: 'msg_001',
chat_jid: 'dc:1234567890123456',
sender: '55512345',
sender_name: 'Alice',
content: 'Hello everyone',
is_from_me: false,
}),
);
});
it('only emits metadata for unregistered channels', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
channelId: '9999999999999999',
content: 'Unknown channel',
guildName: 'Other Server',
});
await triggerMessage(msg);
expect(opts.onChatMetadata).toHaveBeenCalledWith(
'dc:9999999999999999',
expect.any(String),
expect.any(String),
'discord',
true,
);
expect(opts.onMessage).not.toHaveBeenCalled();
});
it('delivers bot messages with is_bot_message flag', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({ isBot: true, content: 'I am a bot' });
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
expect.any(String),
expect.objectContaining({
content: 'I am a bot',
is_bot_message: true,
}),
);
});
it('uses member displayName when available (server nickname)', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'Hi',
memberDisplayName: 'Alice Nickname',
authorDisplayName: 'Alice Global',
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({ sender_name: 'Alice Nickname' }),
);
});
it('falls back to author displayName when no member', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'Hi',
memberDisplayName: undefined,
authorDisplayName: 'Alice Global',
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({ sender_name: 'Alice Global' }),
);
});
it('uses sender name for DM chats (no guild)', async () => {
const opts = createTestOpts({
registeredGroups: vi.fn(() => ({
'dc:1234567890123456': {
name: 'DM',
folder: 'dm',
trigger: '@Andy',
added_at: '2024-01-01T00:00:00.000Z',
},
})),
});
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'Hello',
guildName: undefined,
authorDisplayName: 'Alice',
});
await triggerMessage(msg);
expect(opts.onChatMetadata).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.any(String),
'Alice',
'discord',
false,
);
});
it('uses guild name + channel name for server messages', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'Hello',
guildName: 'My Server',
channelName: 'bot-chat',
});
await triggerMessage(msg);
expect(opts.onChatMetadata).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.any(String),
'My Server #bot-chat',
'discord',
true,
);
});
});
// --- @mention translation ---
describe('@mention translation', () => {
it('translates <@botId> mention to trigger format', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: '<@999888777> what time is it?',
mentionsBotId: true,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '@Andy what time is it?',
}),
);
});
it('does not translate if message already matches trigger', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: '@Andy hello <@999888777>',
mentionsBotId: true,
guildName: 'Server',
});
await triggerMessage(msg);
// Should NOT prepend @Andy — already starts with trigger
// But the <@botId> should still be stripped
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '@Andy hello',
}),
);
});
it('does not translate when bot is not mentioned', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'hello everyone',
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: 'hello everyone',
}),
);
});
it('handles <@!botId> (nickname mention format)', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: '<@!999888777> check this',
mentionsBotId: true,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '@Andy check this',
}),
);
});
});
// --- Attachments ---
describe('attachments', () => {
it('stores image attachment with placeholder', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const attachments = new Map([
['att1', { name: 'photo.png', contentType: 'image/png' }],
]);
const msg = createMessage({
content: '',
attachments,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '[Image: photo.png]',
}),
);
});
it('stores video attachment with placeholder', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const attachments = new Map([
['att1', { name: 'clip.mp4', contentType: 'video/mp4' }],
]);
const msg = createMessage({
content: '',
attachments,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '[Video: clip.mp4]',
}),
);
});
it('stores file attachment with placeholder', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const attachments = new Map([
['att1', { name: 'report.pdf', contentType: 'application/pdf' }],
]);
const msg = createMessage({
content: '',
attachments,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '[File: report.pdf]',
}),
);
});
it('includes text content with attachments', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const attachments = new Map([
['att1', { name: 'photo.jpg', contentType: 'image/jpeg' }],
]);
const msg = createMessage({
content: 'Check this out',
attachments,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: 'Check this out\n[Image: photo.jpg]',
}),
);
});
it('handles multiple attachments', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const attachments = new Map([
['att1', { name: 'a.png', contentType: 'image/png' }],
['att2', { name: 'b.txt', contentType: 'text/plain' }],
]);
const msg = createMessage({
content: '',
attachments,
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '[Image: a.png]\n[File: b.txt]',
}),
);
});
});
// --- Reply context ---
describe('reply context', () => {
it('includes reply author in content', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const msg = createMessage({
content: 'I agree with that',
reference: { messageId: 'original_msg_id' },
guildName: 'Server',
});
await triggerMessage(msg);
expect(opts.onMessage).toHaveBeenCalledWith(
'dc:1234567890123456',
expect.objectContaining({
content: '[Reply to Bob] I agree with that',
}),
);
});
});
// --- sendMessage ---
describe('sendMessage', () => {
it('sends message via channel', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
await channel.sendMessage('dc:1234567890123456', 'Hello');
const fetchedChannel = await currentClient().channels.fetch('1234567890123456');
expect(currentClient().channels.fetch).toHaveBeenCalledWith('1234567890123456');
});
it('strips dc: prefix from JID', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
await channel.sendMessage('dc:9876543210', 'Test');
expect(currentClient().channels.fetch).toHaveBeenCalledWith('9876543210');
});
it('handles send failure gracefully', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
currentClient().channels.fetch.mockRejectedValueOnce(
new Error('Channel not found'),
);
// Should not throw
await expect(
channel.sendMessage('dc:1234567890123456', 'Will fail'),
).resolves.toBeUndefined();
});
it('does nothing when client is not initialized', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
// Don't connect — client is null
await channel.sendMessage('dc:1234567890123456', 'No client');
// No error, no API call
});
it('splits messages exceeding 2000 characters', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const mockChannel = {
send: vi.fn().mockResolvedValue(undefined),
sendTyping: vi.fn(),
};
currentClient().channels.fetch.mockResolvedValue(mockChannel);
const longText = 'x'.repeat(3000);
await channel.sendMessage('dc:1234567890123456', longText);
expect(mockChannel.send).toHaveBeenCalledTimes(2);
expect(mockChannel.send).toHaveBeenNthCalledWith(1, 'x'.repeat(2000));
expect(mockChannel.send).toHaveBeenNthCalledWith(2, 'x'.repeat(1000));
});
});
// --- ownsJid ---
describe('ownsJid', () => {
it('owns dc: JIDs', () => {
const channel = new DiscordChannel('test-token', createTestOpts());
expect(channel.ownsJid('dc:1234567890123456')).toBe(true);
});
it('does not own WhatsApp group JIDs', () => {
const channel = new DiscordChannel('test-token', createTestOpts());
expect(channel.ownsJid('12345@g.us')).toBe(false);
});
it('does not own Telegram JIDs', () => {
const channel = new DiscordChannel('test-token', createTestOpts());
expect(channel.ownsJid('tg:123456789')).toBe(false);
});
it('does not own unknown JID formats', () => {
const channel = new DiscordChannel('test-token', createTestOpts());
expect(channel.ownsJid('random-string')).toBe(false);
});
});
// --- setTyping ---
describe('setTyping', () => {
it('sends typing indicator when isTyping is true', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const mockChannel = {
send: vi.fn(),
sendTyping: vi.fn().mockResolvedValue(undefined),
};
currentClient().channels.fetch.mockResolvedValue(mockChannel);
await channel.setTyping('dc:1234567890123456', true);
expect(mockChannel.sendTyping).toHaveBeenCalled();
});
it('does nothing when isTyping is false', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
await channel.setTyping('dc:1234567890123456', false);
// channels.fetch should NOT be called
expect(currentClient().channels.fetch).not.toHaveBeenCalled();
});
it('does nothing when client is not initialized', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
// Don't connect
await channel.setTyping('dc:1234567890123456', true);
// No error
});
});
// --- Channel properties ---
describe('channel properties', () => {
it('has name "discord"', () => {
const channel = new DiscordChannel('test-token', createTestOpts());
expect(channel.name).toBe('discord');
});
});
});

291
src/channels/discord.ts Normal file
View File

@@ -0,0 +1,291 @@
import { Client, Events, GatewayIntentBits, Message, TextChannel } from 'discord.js';
import { ASSISTANT_NAME, TRIGGER_PATTERN } from '../config.js';
import { readEnvFile } from '../env.js';
import { logger } from '../logger.js';
import { registerChannel, ChannelOpts } from './registry.js';
import {
AgentType,
Channel,
OnChatMetadata,
OnInboundMessage,
RegisteredGroup,
} from '../types.js';
export interface DiscordChannelOpts {
onMessage: OnInboundMessage;
onChatMetadata: OnChatMetadata;
registeredGroups: () => Record<string, RegisteredGroup>;
}
export class DiscordChannel implements Channel {
name = 'discord';
private client: Client | null = null;
private opts: DiscordChannelOpts;
private botToken: string;
private typingIntervals = new Map<string, NodeJS.Timeout>();
private agentTypeFilter?: AgentType;
constructor(botToken: string, opts: DiscordChannelOpts, agentTypeFilter?: AgentType) {
this.botToken = botToken;
this.opts = opts;
this.agentTypeFilter = agentTypeFilter;
if (agentTypeFilter) {
this.name = `discord-${agentTypeFilter}`;
}
}
async connect(): Promise<void> {
this.client = new Client({
intents: [
GatewayIntentBits.Guilds,
GatewayIntentBits.GuildMessages,
GatewayIntentBits.MessageContent,
GatewayIntentBits.DirectMessages,
],
});
this.client.on(Events.MessageCreate, async (message: Message) => {
// Ignore own messages only
if (message.author.id === this.client?.user?.id) return;
const channelId = message.channelId;
const chatJid = `dc:${channelId}`;
let content = message.content;
const timestamp = message.createdAt.toISOString();
const senderName =
message.member?.displayName ||
message.author.displayName ||
message.author.username;
const sender = message.author.id;
const msgId = message.id;
// Determine chat name
let chatName: string;
if (message.guild) {
const textChannel = message.channel as TextChannel;
chatName = `${message.guild.name} #${textChannel.name}`;
} else {
chatName = senderName;
}
// Translate Discord @bot mentions into TRIGGER_PATTERN format.
// Discord mentions look like <@botUserId> — these won't match
// TRIGGER_PATTERN (e.g., ^@Andy\b), so we prepend the trigger
// when the bot is @mentioned.
if (this.client?.user) {
const botId = this.client.user.id;
const isBotMentioned =
message.mentions.users.has(botId) ||
content.includes(`<@${botId}>`) ||
content.includes(`<@!${botId}>`);
if (isBotMentioned) {
// Strip the <@botId> mention to avoid visual clutter
content = content
.replace(new RegExp(`<@!?${botId}>`, 'g'), '')
.trim();
// Prepend trigger if not already present
if (!TRIGGER_PATTERN.test(content)) {
content = `@${ASSISTANT_NAME} ${content}`;
}
}
}
// Handle attachments — store placeholders so the agent knows something was sent
if (message.attachments.size > 0) {
const attachmentDescriptions = [...message.attachments.values()].map((att) => {
const contentType = att.contentType || '';
if (contentType.startsWith('image/')) {
return `[Image: ${att.name || 'image'}]`;
} else if (contentType.startsWith('video/')) {
return `[Video: ${att.name || 'video'}]`;
} else if (contentType.startsWith('audio/')) {
return `[Audio: ${att.name || 'audio'}]`;
} else {
return `[File: ${att.name || 'file'}]`;
}
});
if (content) {
content = `${content}\n${attachmentDescriptions.join('\n')}`;
} else {
content = attachmentDescriptions.join('\n');
}
}
// Handle reply context — include who the user is replying to
if (message.reference?.messageId) {
try {
const repliedTo = await message.channel.messages.fetch(
message.reference.messageId,
);
const replyAuthor =
repliedTo.member?.displayName ||
repliedTo.author.displayName ||
repliedTo.author.username;
content = `[Reply to ${replyAuthor}] ${content}`;
} catch {
// Referenced message may have been deleted
}
}
// Store chat metadata for discovery
const isGroup = message.guild !== null;
this.opts.onChatMetadata(chatJid, timestamp, chatName, 'discord', isGroup);
// Only deliver full message for registered groups matching our agent type
const group = this.opts.registeredGroups()[chatJid];
if (!group) {
logger.debug(
{ chatJid, chatName },
'Message from unregistered Discord channel',
);
return;
}
if (this.agentTypeFilter && (group.agentType || 'claude-code') !== this.agentTypeFilter) {
return; // This JID belongs to a different agent type's bot
}
// Deliver message — startMessageLoop() will pick it up
this.opts.onMessage(chatJid, {
id: msgId,
chat_jid: chatJid,
sender,
sender_name: senderName,
content,
timestamp,
is_from_me: false,
is_bot_message: message.author.bot ?? false,
});
logger.info(
{ chatJid, chatName, sender: senderName },
'Discord message stored',
);
});
// Handle errors gracefully
this.client.on(Events.Error, (err) => {
logger.error({ err: err.message }, 'Discord client error');
});
return new Promise<void>((resolve) => {
this.client!.once(Events.ClientReady, (readyClient) => {
logger.info(
{ username: readyClient.user.tag, id: readyClient.user.id },
'Discord bot connected',
);
console.log(`\n Discord bot: ${readyClient.user.tag}`);
console.log(
` Use /chatid command or check channel IDs in Discord settings\n`,
);
resolve();
});
this.client!.login(this.botToken);
});
}
async sendMessage(jid: string, text: string): Promise<void> {
if (!this.client) {
logger.warn('Discord client not initialized');
return;
}
try {
const channelId = jid.replace(/^dc:/, '');
const channel = await this.client.channels.fetch(channelId);
if (!channel || !('send' in channel)) {
logger.warn({ jid }, 'Discord channel not found or not text-based');
return;
}
const textChannel = channel as TextChannel;
// Discord has a 2000 character limit per message — split if needed
const MAX_LENGTH = 2000;
if (text.length <= MAX_LENGTH) {
await textChannel.send(text);
} else {
for (let i = 0; i < text.length; i += MAX_LENGTH) {
await textChannel.send(text.slice(i, i + MAX_LENGTH));
}
}
logger.info({ jid, length: text.length }, 'Discord message sent');
} catch (err) {
logger.error({ jid, err }, 'Failed to send Discord message');
}
}
isConnected(): boolean {
return this.client !== null && this.client.isReady();
}
ownsJid(jid: string): boolean {
if (!jid.startsWith('dc:')) return false;
if (!this.agentTypeFilter) return true;
const group = this.opts.registeredGroups()[jid];
if (!group) return false;
return (group.agentType || 'claude-code') === this.agentTypeFilter;
}
async disconnect(): Promise<void> {
if (this.client) {
this.client.destroy();
this.client = null;
logger.info('Discord bot stopped');
}
}
async setTyping(jid: string, isTyping: boolean): Promise<void> {
if (!this.client) return;
// Clear any existing interval for this channel
const existing = this.typingIntervals.get(jid);
if (existing) {
clearInterval(existing);
this.typingIntervals.delete(jid);
}
if (!isTyping) return;
const sendOnce = async () => {
try {
const channelId = jid.replace(/^dc:/, '');
const channel = await this.client!.channels.fetch(channelId);
if (channel && 'sendTyping' in channel) {
await (channel as TextChannel).sendTyping();
}
} catch (err) {
logger.debug({ jid, err }, 'Failed to send Discord typing indicator');
}
};
// Send immediately, then refresh every 8 seconds (Discord expires at ~10s)
await sendOnce();
this.typingIntervals.set(jid, setInterval(sendOnce, 8000));
}
}
registerChannel('discord', (opts: ChannelOpts) => {
const envVars = readEnvFile(['DISCORD_BOT_TOKEN', 'DISCORD_CODEX_BOT_TOKEN']);
const token =
process.env.DISCORD_BOT_TOKEN || envVars.DISCORD_BOT_TOKEN || '';
if (!token) {
logger.warn('Discord: DISCORD_BOT_TOKEN not set');
return null;
}
// If a second Codex bot token exists, this instance only handles claude-code groups
const hasCodexBot = !!(process.env.DISCORD_CODEX_BOT_TOKEN || envVars.DISCORD_CODEX_BOT_TOKEN);
return new DiscordChannel(token, opts, hasCodexBot ? 'claude-code' : undefined);
});
registerChannel('discord-codex', (opts: ChannelOpts) => {
const envVars = readEnvFile(['DISCORD_CODEX_BOT_TOKEN']);
const token =
process.env.DISCORD_CODEX_BOT_TOKEN || envVars.DISCORD_CODEX_BOT_TOKEN || '';
if (!token) return null; // Codex Discord bot is optional
return new DiscordChannel(token, opts, 'codex');
});

View File

@@ -2,6 +2,7 @@
// Each import triggers the channel module's registerChannel() call.
// discord
import './discord.js';
// gmail

View File

@@ -39,6 +39,8 @@ export const DATA_DIR = path.resolve(PROJECT_ROOT, 'data');
export const CONTAINER_IMAGE =
process.env.CONTAINER_IMAGE || 'nanoclaw-agent:latest';
export const CODEX_CONTAINER_IMAGE =
process.env.CODEX_CONTAINER_IMAGE || 'nanoclaw-codex-agent:latest';
export const CONTAINER_TIMEOUT = parseInt(
process.env.CONTAINER_TIMEOUT || '1800000',
10,

View File

@@ -4,9 +4,11 @@
*/
import { ChildProcess, exec, spawn } from 'child_process';
import fs from 'fs';
import os from 'os';
import path from 'path';
import {
CODEX_CONTAINER_IMAGE,
CONTAINER_IMAGE,
CONTAINER_MAX_OUTPUT_SIZE,
CONTAINER_TIMEOUT,
@@ -26,6 +28,7 @@ import {
stopContainer,
} from './container-runtime.js';
import { detectAuthMode } from './credential-proxy.js';
import { readEnvFile } from './env.js';
import { validateAdditionalMounts } from './mount-security.js';
import { RegisteredGroup } from './types.js';
@@ -41,6 +44,7 @@ export interface ContainerInput {
isMain: boolean;
isScheduledTask?: boolean;
assistantName?: string;
agentType?: 'claude-code' | 'codex';
}
export interface ContainerOutput {
@@ -76,17 +80,6 @@ function buildVolumeMounts(
readonly: true,
});
// Shadow .env so the agent cannot read secrets from the mounted project root.
// Credentials are injected by the credential proxy, never exposed to containers.
const envFile = path.join(projectRoot, '.env');
if (fs.existsSync(envFile)) {
mounts.push({
hostPath: '/dev/null',
containerPath: '/workspace/project/.env',
readonly: true,
});
}
// Main also gets its group folder as the working directory
mounts.push({
hostPath: groupDir,
@@ -163,6 +156,40 @@ function buildVolumeMounts(
readonly: false,
});
// For Codex groups: create a per-session writable .codex/ directory
if (group.agentType === 'codex') {
const hostCodexDir = path.join(process.env.HOME || os.homedir(), '.codex');
const sessionCodexDir = path.join(
DATA_DIR,
'sessions',
group.folder,
'.codex',
);
fs.mkdirSync(sessionCodexDir, { recursive: true });
// Always refresh auth credentials from host
const authSrc = path.join(hostCodexDir, 'auth.json');
const authDst = path.join(sessionCodexDir, 'auth.json');
if (fs.existsSync(authSrc)) {
fs.copyFileSync(authSrc, authDst);
}
// Only copy config files if they don't exist yet (preserves per-session customization)
for (const file of ['config.toml', 'config.json']) {
const src = path.join(hostCodexDir, file);
const dst = path.join(sessionCodexDir, file);
if (fs.existsSync(src) && !fs.existsSync(dst)) {
fs.copyFileSync(src, dst);
}
}
mounts.push({
hostPath: sessionCodexDir,
containerPath: '/home/node/.codex',
readonly: false,
});
}
// Per-group IPC namespace: each group gets its own IPC directory
// This prevents cross-group privilege escalation via IPC
const groupIpcDir = resolveGroupIpcPath(group.folder);
@@ -175,20 +202,24 @@ function buildVolumeMounts(
readonly: false,
});
// Copy agent-runner source into a per-group writable location so agents
// Copy runner source into a per-group writable location so agents
// can customize it (add tools, change behavior) without affecting other
// groups. Recompiled on container startup via entrypoint.sh.
const agentType = group.agentType || 'claude-code';
const runnerDirName = agentType === 'codex' ? 'codex-runner' : 'agent-runner';
const runnerSrcSuffix =
agentType === 'codex' ? 'codex-runner-src' : 'agent-runner-src';
const agentRunnerSrc = path.join(
projectRoot,
'container',
'agent-runner',
runnerDirName,
'src',
);
const groupAgentRunnerDir = path.join(
DATA_DIR,
'sessions',
group.folder,
'agent-runner-src',
runnerSrcSuffix,
);
if (!fs.existsSync(groupAgentRunnerDir) && fs.existsSync(agentRunnerSrc)) {
fs.cpSync(agentRunnerSrc, groupAgentRunnerDir, { recursive: true });
@@ -215,39 +246,69 @@ function buildVolumeMounts(
function buildContainerArgs(
mounts: VolumeMount[],
containerName: string,
isMain: boolean,
agentType: 'claude-code' | 'codex' = 'claude-code',
): string[] {
const args: string[] = ['run', '-i', '--rm', '--name', containerName];
// Pass host timezone so container's local time matches the user's
args.push('-e', `TZ=${TIMEZONE}`);
// Route API traffic through the credential proxy (containers never see real secrets)
args.push(
'-e',
`ANTHROPIC_BASE_URL=http://${CONTAINER_HOST_GATEWAY}:${CREDENTIAL_PROXY_PORT}`,
);
// Mirror the host's auth method with a placeholder value.
// API key mode: SDK sends x-api-key, proxy replaces with real key.
// OAuth mode: SDK exchanges placeholder token for temp API key,
// proxy injects real OAuth token on that exchange request.
const authMode = detectAuthMode();
if (authMode === 'api-key') {
args.push('-e', 'ANTHROPIC_API_KEY=placeholder');
if (agentType === 'codex') {
// Codex uses OpenAI API — pass the key directly (no credential proxy)
const openaiKey =
process.env.CODEX_OPENAI_API_KEY || process.env.OPENAI_API_KEY;
if (openaiKey) {
args.push('-e', `OPENAI_API_KEY=${openaiKey}`);
}
} else {
args.push('-e', 'CLAUDE_CODE_OAUTH_TOKEN=placeholder');
// Route API traffic through the credential proxy (containers never see real secrets)
args.push(
'-e',
`ANTHROPIC_BASE_URL=http://${CONTAINER_HOST_GATEWAY}:${CREDENTIAL_PROXY_PORT}`,
);
// Mirror the host's auth method with a placeholder value.
const authMode = detectAuthMode();
if (authMode === 'api-key') {
args.push('-e', 'ANTHROPIC_API_KEY=placeholder');
} else {
args.push('-e', 'CLAUDE_CODE_OAUTH_TOKEN=placeholder');
}
}
// Runtime-specific args for host gateway resolution
args.push(...hostGatewayArgs());
// Pass model and thinking configuration to container agent (read from .env file)
const modelEnv = readEnvFile(['CLAUDE_MODEL', 'CLAUDE_THINKING', 'CLAUDE_THINKING_BUDGET', 'CLAUDE_EFFORT']);
if (modelEnv.CLAUDE_MODEL) {
args.push('-e', `CLAUDE_MODEL=${modelEnv.CLAUDE_MODEL}`);
}
if (modelEnv.CLAUDE_THINKING) {
args.push('-e', `CLAUDE_THINKING=${modelEnv.CLAUDE_THINKING}`);
}
if (modelEnv.CLAUDE_THINKING_BUDGET) {
args.push('-e', `CLAUDE_THINKING_BUDGET=${modelEnv.CLAUDE_THINKING_BUDGET}`);
}
if (modelEnv.CLAUDE_EFFORT) {
args.push('-e', `CLAUDE_EFFORT=${modelEnv.CLAUDE_EFFORT}`);
}
// Run as host user so bind-mounted files are accessible.
// Skip when running as root (uid 0), as the container's node user (uid 1000),
// or when getuid is unavailable (native Windows without WSL).
const hostUid = process.getuid?.();
const hostGid = process.getgid?.();
if (hostUid != null && hostUid !== 0 && hostUid !== 1000) {
args.push('--user', `${hostUid}:${hostGid}`);
if (isMain) {
// Main containers start as root so the entrypoint can mount --bind
// to shadow .env. Privileges are dropped via setpriv in entrypoint.sh.
args.push('-e', `RUN_UID=${hostUid}`);
args.push('-e', `RUN_GID=${hostGid}`);
} else {
args.push('--user', `${hostUid}:${hostGid}`);
}
args.push('-e', 'HOME=/home/node');
}
@@ -259,7 +320,8 @@ function buildContainerArgs(
}
}
args.push(CONTAINER_IMAGE);
// Select container image based on agent type
args.push(agentType === 'codex' ? CODEX_CONTAINER_IMAGE : CONTAINER_IMAGE);
return args;
}
@@ -278,7 +340,13 @@ export async function runContainerAgent(
const mounts = buildVolumeMounts(group, input.isMain);
const safeName = group.folder.replace(/[^a-zA-Z0-9-]/g, '-');
const containerName = `nanoclaw-${safeName}-${Date.now()}`;
const containerArgs = buildContainerArgs(mounts, containerName);
const agentType = group.agentType || 'claude-code';
const containerArgs = buildContainerArgs(
mounts,
containerName,
input.isMain,
agentType,
);
logger.debug(
{

View File

@@ -32,9 +32,12 @@ beforeEach(() => {
// --- Pure functions ---
describe('readonlyMountArgs', () => {
it('returns -v flag with :ro suffix', () => {
it('returns --mount flag with type=bind and readonly', () => {
const args = readonlyMountArgs('/host/path', '/container/path');
expect(args).toEqual(['-v', '/host/path:/container/path:ro']);
expect(args).toEqual([
'--mount',
'type=bind,source=/host/path,target=/container/path,readonly',
]);
});
});
@@ -55,18 +58,35 @@ describe('ensureContainerRuntimeRunning', () => {
ensureContainerRuntimeRunning();
expect(mockExecSync).toHaveBeenCalledTimes(1);
expect(mockExecSync).toHaveBeenCalledWith(`${CONTAINER_RUNTIME_BIN} info`, {
stdio: 'pipe',
timeout: 10000,
});
expect(logger.debug).toHaveBeenCalledWith(
'Container runtime already running',
expect(mockExecSync).toHaveBeenCalledWith(
`${CONTAINER_RUNTIME_BIN} system status`,
{ stdio: 'pipe' },
);
expect(logger.debug).toHaveBeenCalledWith('Container runtime already running');
});
it('throws when docker info fails', () => {
it('auto-starts when system status fails', () => {
// First call (system status) fails
mockExecSync.mockImplementationOnce(() => {
throw new Error('Cannot connect to the Docker daemon');
throw new Error('not running');
});
// Second call (system start) succeeds
mockExecSync.mockReturnValueOnce('');
ensureContainerRuntimeRunning();
expect(mockExecSync).toHaveBeenCalledTimes(2);
expect(mockExecSync).toHaveBeenNthCalledWith(
2,
`${CONTAINER_RUNTIME_BIN} system start`,
{ stdio: 'pipe', timeout: 30000 },
);
expect(logger.info).toHaveBeenCalledWith('Container runtime started');
});
it('throws when both status and start fail', () => {
mockExecSync.mockImplementation(() => {
throw new Error('failed');
});
expect(() => ensureContainerRuntimeRunning()).toThrow(
@@ -79,17 +99,21 @@ describe('ensureContainerRuntimeRunning', () => {
// --- cleanupOrphans ---
describe('cleanupOrphans', () => {
it('stops orphaned nanoclaw containers', () => {
// docker ps returns container names, one per line
mockExecSync.mockReturnValueOnce(
'nanoclaw-group1-111\nnanoclaw-group2-222\n',
);
it('stops orphaned nanoclaw containers from JSON output', () => {
// Apple Container ls returns JSON
const lsOutput = JSON.stringify([
{ status: 'running', configuration: { id: 'nanoclaw-group1-111' } },
{ status: 'stopped', configuration: { id: 'nanoclaw-group2-222' } },
{ status: 'running', configuration: { id: 'nanoclaw-group3-333' } },
{ status: 'running', configuration: { id: 'other-container' } },
]);
mockExecSync.mockReturnValueOnce(lsOutput);
// stop calls succeed
mockExecSync.mockReturnValue('');
cleanupOrphans();
// ps + 2 stop calls
// ls + 2 stop calls (only running nanoclaw- containers)
expect(mockExecSync).toHaveBeenCalledTimes(3);
expect(mockExecSync).toHaveBeenNthCalledWith(
2,
@@ -98,17 +122,17 @@ describe('cleanupOrphans', () => {
);
expect(mockExecSync).toHaveBeenNthCalledWith(
3,
`${CONTAINER_RUNTIME_BIN} stop nanoclaw-group2-222`,
`${CONTAINER_RUNTIME_BIN} stop nanoclaw-group3-333`,
{ stdio: 'pipe' },
);
expect(logger.info).toHaveBeenCalledWith(
{ count: 2, names: ['nanoclaw-group1-111', 'nanoclaw-group2-222'] },
{ count: 2, names: ['nanoclaw-group1-111', 'nanoclaw-group3-333'] },
'Stopped orphaned containers',
);
});
it('does nothing when no orphans exist', () => {
mockExecSync.mockReturnValueOnce('');
mockExecSync.mockReturnValueOnce('[]');
cleanupOrphans();
@@ -116,9 +140,9 @@ describe('cleanupOrphans', () => {
expect(logger.info).not.toHaveBeenCalled();
});
it('warns and continues when ps fails', () => {
it('warns and continues when ls fails', () => {
mockExecSync.mockImplementationOnce(() => {
throw new Error('docker not available');
throw new Error('container not available');
});
cleanupOrphans(); // should not throw
@@ -130,7 +154,11 @@ describe('cleanupOrphans', () => {
});
it('continues stopping remaining containers when one stop fails', () => {
mockExecSync.mockReturnValueOnce('nanoclaw-a-1\nnanoclaw-b-2\n');
const lsOutput = JSON.stringify([
{ status: 'running', configuration: { id: 'nanoclaw-a-1' } },
{ status: 'running', configuration: { id: 'nanoclaw-b-2' } },
]);
mockExecSync.mockReturnValueOnce(lsOutput);
// First stop fails
mockExecSync.mockImplementationOnce(() => {
throw new Error('already stopped');

View File

@@ -3,58 +3,35 @@
* All runtime-specific logic lives here so swapping runtimes means changing one file.
*/
import { execSync } from 'child_process';
import fs from 'fs';
import os from 'os';
import { logger } from './logger.js';
/** The container runtime binary name. */
export const CONTAINER_RUNTIME_BIN = 'docker';
/** Hostname containers use to reach the host machine. */
export const CONTAINER_HOST_GATEWAY = 'host.docker.internal';
export const CONTAINER_RUNTIME_BIN = 'container';
/**
* Address the credential proxy binds to.
* Docker Desktop (macOS): 127.0.0.1 — the VM routes host.docker.internal to loopback.
* Docker (Linux): bind to the docker0 bridge IP so only containers can reach it,
* falling back to 0.0.0.0 if the interface isn't found.
* Hostname containers use to reach the host machine.
* Apple Container VMs access the host via the default gateway (192.168.64.1).
*/
export const PROXY_BIND_HOST =
process.env.CREDENTIAL_PROXY_HOST || detectProxyBindHost();
export const CONTAINER_HOST_GATEWAY = '192.168.64.1';
function detectProxyBindHost(): string {
if (os.platform() === 'darwin') return '127.0.0.1';
/**
* Address the credential proxy binds to on the host.
* Apple Container VMs reach the host via 192.168.64.1, so we must bind to all interfaces.
*/
export const PROXY_BIND_HOST = '0.0.0.0';
// WSL uses Docker Desktop (same VM routing as macOS) — loopback is correct.
// Check /proc filesystem, not env vars — WSL_DISTRO_NAME isn't set under systemd.
if (fs.existsSync('/proc/sys/fs/binfmt_misc/WSLInterop')) return '127.0.0.1';
// Bare-metal Linux: bind to the docker0 bridge IP instead of 0.0.0.0
const ifaces = os.networkInterfaces();
const docker0 = ifaces['docker0'];
if (docker0) {
const ipv4 = docker0.find((a) => a.family === 'IPv4');
if (ipv4) return ipv4.address;
}
return '0.0.0.0';
}
/** CLI args needed for the container to resolve the host gateway. */
/**
* CLI args needed for the container to resolve the host gateway.
* Apple Container provides host networking natively on macOS — no extra args needed.
*/
export function hostGatewayArgs(): string[] {
// On Linux, host.docker.internal isn't built-in — add it explicitly
if (os.platform() === 'linux') {
return ['--add-host=host.docker.internal:host-gateway'];
}
return [];
}
/** Returns CLI args for a readonly bind mount. */
export function readonlyMountArgs(
hostPath: string,
containerPath: string,
): string[] {
return ['-v', `${hostPath}:${containerPath}:ro`];
export function readonlyMountArgs(hostPath: string, containerPath: string): string[] {
return ['--mount', `type=bind,source=${hostPath},target=${containerPath},readonly`];
}
/** Returns the shell command to stop a container by name. */
@@ -65,61 +42,62 @@ export function stopContainer(name: string): string {
/** Ensure the container runtime is running, starting it if needed. */
export function ensureContainerRuntimeRunning(): void {
try {
execSync(`${CONTAINER_RUNTIME_BIN} info`, {
stdio: 'pipe',
timeout: 10000,
});
execSync(`${CONTAINER_RUNTIME_BIN} system status`, { stdio: 'pipe' });
logger.debug('Container runtime already running');
} catch (err) {
logger.error({ err }, 'Failed to reach container runtime');
console.error(
'\n╔════════════════════════════════════════════════════════════════╗',
);
console.error(
'║ FATAL: Container runtime failed to start ║',
);
console.error(
'║ ║',
);
console.error(
'║ Agents cannot run without a container runtime. To fix: ║',
);
console.error(
'║ 1. Ensure Docker is installed and running ║',
);
console.error(
'║ 2. Run: docker info ║',
);
console.error(
'║ 3. Restart NanoClaw ║',
);
console.error(
'╚════════════════════════════════════════════════════════════════╝\n',
);
throw new Error('Container runtime is required but failed to start');
} catch {
logger.info('Starting container runtime...');
try {
execSync(`${CONTAINER_RUNTIME_BIN} system start`, { stdio: 'pipe', timeout: 30000 });
logger.info('Container runtime started');
} catch (err) {
logger.error({ err }, 'Failed to start container runtime');
console.error(
'\n╔════════════════════════════════════════════════════════════════╗',
);
console.error(
'║ FATAL: Container runtime failed to start ║',
);
console.error(
'║ ║',
);
console.error(
'║ Agents cannot run without a container runtime. To fix: ║',
);
console.error(
'║ 1. Ensure Apple Container is installed ║',
);
console.error(
'║ 2. Run: container system start ║',
);
console.error(
'║ 3. Restart NanoClaw ║',
);
console.error(
'╚════════════════════════════════════════════════════════════════╝\n',
);
throw new Error('Container runtime is required but failed to start');
}
}
}
/** Kill orphaned NanoClaw containers from previous runs. */
export function cleanupOrphans(): void {
try {
const output = execSync(
`${CONTAINER_RUNTIME_BIN} ps --filter name=nanoclaw- --format '{{.Names}}'`,
{ stdio: ['pipe', 'pipe', 'pipe'], encoding: 'utf-8' },
);
const orphans = output.trim().split('\n').filter(Boolean);
const output = execSync(`${CONTAINER_RUNTIME_BIN} ls --format json`, {
stdio: ['pipe', 'pipe', 'pipe'],
encoding: 'utf-8',
});
const containers: { status: string; configuration: { id: string } }[] = JSON.parse(output || '[]');
const orphans = containers
.filter((c) => c.status === 'running' && c.configuration.id.startsWith('nanoclaw-'))
.map((c) => c.configuration.id);
for (const name of orphans) {
try {
execSync(stopContainer(name), { stdio: 'pipe' });
} catch {
/* already stopped */
}
} catch { /* already stopped */ }
}
if (orphans.length > 0) {
logger.info(
{ count: orphans.length, names: orphans },
'Stopped orphaned containers',
);
logger.info({ count: orphans.length, names: orphans }, 'Stopped orphaned containers');
}
} catch (err) {
logger.warn({ err }, 'Failed to clean up orphaned containers');

View File

@@ -187,25 +187,26 @@ describe('getMessagesSince', () => {
'2024-01-01T00:00:02.000Z',
'Andy',
);
// Should exclude m1, m2 (before/at timestamp), m3 (bot message)
expect(msgs).toHaveLength(1);
expect(msgs[0].content).toBe('third');
// Should exclude m1, m2 (before/at timestamp); m3 (bot) is now included
expect(msgs).toHaveLength(2);
expect(msgs[0].content).toBe('bot reply');
expect(msgs[1].content).toBe('third');
});
it('excludes bot messages via is_bot_message flag', () => {
it('includes bot messages with is_bot_message flag', () => {
const msgs = getMessagesSince(
'group@g.us',
'2024-01-01T00:00:00.000Z',
'Andy',
);
const botMsgs = msgs.filter((m) => m.content === 'bot reply');
expect(botMsgs).toHaveLength(0);
expect(botMsgs).toHaveLength(1);
});
it('returns all non-bot messages when sinceTimestamp is empty', () => {
it('returns all messages including bot when sinceTimestamp is empty', () => {
const msgs = getMessagesSince('group@g.us', '', 'Andy');
// 3 user messages (bot message excluded)
expect(msgs).toHaveLength(3);
// 3 user messages + 1 bot message
expect(msgs).toHaveLength(4);
});
it('filters pre-migration bot messages via content prefix backstop', () => {
@@ -275,8 +276,8 @@ describe('getNewMessages', () => {
'2024-01-01T00:00:00.000Z',
'Andy',
);
// Excludes bot message, returns 3 user messages
expect(messages).toHaveLength(3);
// Includes bot message, returns all 4 messages
expect(messages).toHaveLength(4);
expect(newTimestamp).toBe('2024-01-01T00:00:04.000Z');
});
@@ -286,9 +287,10 @@ describe('getNewMessages', () => {
'2024-01-01T00:00:02.000Z',
'Andy',
);
// Only g1 msg2 (after ts, not bot)
expect(messages).toHaveLength(1);
expect(messages[0].content).toBe('g1 msg2');
// bot reply (00:00:03) + g1 msg2 (00:00:04) after ts
expect(messages).toHaveLength(2);
expect(messages[0].content).toBe('bot reply');
expect(messages[1].content).toBe('g1 msg2');
});
it('returns empty for no registered groups', () => {

View File

@@ -119,6 +119,15 @@ function createSchema(database: Database.Database): void {
/* column already exists */
}
// Add agent_type column if it doesn't exist (migration for Codex support)
try {
database.exec(
`ALTER TABLE registered_groups ADD COLUMN agent_type TEXT DEFAULT 'claude-code'`,
);
} catch {
/* column already exists */
}
// Add channel and is_group columns if they don't exist (migration for existing DBs)
try {
database.exec(`ALTER TABLE chats ADD COLUMN channel TEXT`);
@@ -316,10 +325,10 @@ export function getNewMessages(
// Subquery takes the N most recent, outer query re-sorts chronologically.
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
FROM messages
WHERE timestamp > ? AND chat_jid IN (${placeholders})
AND is_bot_message = 0 AND content NOT LIKE ?
AND content NOT LIKE ?
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC
LIMIT ?
@@ -349,10 +358,10 @@ export function getMessagesSince(
// Subquery takes the N most recent, outer query re-sorts chronologically.
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
FROM messages
WHERE chat_jid = ? AND timestamp > ?
AND is_bot_message = 0 AND content NOT LIKE ?
AND content NOT LIKE ?
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC
LIMIT ?
@@ -363,6 +372,18 @@ export function getMessagesSince(
.all(chatJid, sinceTimestamp, `${botPrefix}:%`, limit) as NewMessage[];
}
export function getLastHumanMessageTimestamp(chatJid: string): string | null {
const row = db
.prepare(
`SELECT timestamp FROM messages
WHERE chat_jid = ? AND is_bot_message = 0 AND is_from_me = 0
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC LIMIT 1`,
)
.get(chatJid) as { timestamp: string } | undefined;
return row?.timestamp ?? null;
}
export function createTask(
task: Omit<ScheduledTask, 'last_run' | 'last_result'>,
): void {

View File

@@ -59,11 +59,16 @@ export class GroupQueue {
this.processMessagesFn = fn;
}
enqueueMessageCheck(groupJid: string): void {
enqueueMessageCheck(groupJid: string, groupFolder?: string): void {
if (this.shuttingDown) return;
const state = this.getGroup(groupJid);
// Pre-set groupFolder so sendMessage can pipe IPC while container starts
if (groupFolder && !state.groupFolder) {
state.groupFolder = groupFolder;
}
if (state.active) {
state.pendingMessages = true;
logger.debug({ groupJid }, 'Container active, message queued');

View File

@@ -31,6 +31,7 @@ import {
getAllRegisteredGroups,
getAllSessions,
getAllTasks,
getLastHumanMessageTimestamp,
getMessagesSince,
getNewMessages,
getRegisteredGroup,
@@ -52,6 +53,7 @@ import {
loadSenderAllowlist,
shouldDropMessage,
} from './sender-allowlist.js';
import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed } from './session-commands.js';
import { startSchedulerLoop } from './task-scheduler.js';
import { Channel, NewMessage, RegisteredGroup } from './types.js';
import { logger } from './logger.js';
@@ -164,6 +166,33 @@ async function processGroupMessages(chatJid: string): Promise<boolean> {
if (missedMessages.length === 0) return true;
// --- Session command interception (before trigger check) ---
const cmdResult = await handleSessionCommand({
missedMessages,
isMainGroup,
groupName: group.name,
triggerPattern: TRIGGER_PATTERN,
timezone: TIMEZONE,
deps: {
sendMessage: (text) => channel.sendMessage(chatJid, text),
setTyping: (typing) => channel.setTyping?.(chatJid, typing) ?? Promise.resolve(),
runAgent: (prompt, onOutput) => runAgent(group, prompt, chatJid, onOutput),
closeStdin: () => queue.closeStdin(chatJid),
advanceCursor: (ts) => { lastAgentTimestamp[chatJid] = ts; saveState(); },
formatMessages,
canSenderInteract: (msg) => {
const hasTrigger = TRIGGER_PATTERN.test(msg.content.trim());
const reqTrigger = !isMainGroup && group.requiresTrigger !== false;
return isMainGroup || !reqTrigger || (hasTrigger && (
msg.is_from_me ||
isTriggerAllowed(chatJid, msg.sender, loadSenderAllowlist())
));
},
},
});
if (cmdResult.handled) return cmdResult.success;
// --- End session command interception ---
// For non-main groups, check if trigger is required and present
if (!isMainGroup && group.requiresTrigger !== false) {
const allowlistCfg = loadSenderAllowlist();
@@ -172,7 +201,9 @@ async function processGroupMessages(chatJid: string): Promise<boolean> {
TRIGGER_PATTERN.test(m.content.trim()) &&
(m.is_from_me || isTriggerAllowed(chatJid, m.sender, allowlistCfg)),
);
if (!hasTrigger) return true;
if (!hasTrigger) {
return true;
}
}
const prompt = formatMessages(missedMessages, TIMEZONE);
@@ -388,6 +419,51 @@ async function startMessageLoop(): Promise<void> {
}
const isMainGroup = group.isMain === true;
// --- Bot-collaboration timeout ---
// If all new messages are from bots, only process if a human
// sent a message within the last 12 hours.
const BOT_COLLAB_TIMEOUT_MS = 12 * 60 * 60 * 1000;
const allFromBots = groupMessages.every(
(m) => m.is_from_me || !!m.is_bot_message,
);
if (allFromBots) {
const lastHuman = getLastHumanMessageTimestamp(chatJid);
if (
!lastHuman ||
Date.now() - new Date(lastHuman).getTime() >
BOT_COLLAB_TIMEOUT_MS
) {
logger.info(
{ chatJid, lastHuman },
'Bot-collaboration timeout: no human message within 12h, skipping',
);
continue;
}
}
// --- End bot-collaboration timeout ---
// --- Session command interception (message loop) ---
// Scan ALL messages in the batch for a session command.
const loopCmdMsg = groupMessages.find(
(m) => extractSessionCommand(m.content, TRIGGER_PATTERN) !== null,
);
if (loopCmdMsg) {
// Only close active container if the sender is authorized — otherwise an
// untrusted user could kill in-flight work by sending /compact (DoS).
// closeStdin no-ops internally when no container is active.
if (isSessionCommandAllowed(isMainGroup, loopCmdMsg.is_from_me === true)) {
queue.closeStdin(chatJid);
}
// Enqueue so processGroupMessages handles auth + cursor advancement.
// Don't pipe via IPC — slash commands need a fresh container with
// string prompt (not MessageStream) for SDK recognition.
queue.enqueueMessageCheck(chatJid);
continue;
}
// --- End session command interception ---
const needsTrigger = !isMainGroup && group.requiresTrigger !== false;
// For non-main groups, only act on trigger messages.
@@ -431,7 +507,7 @@ async function startMessageLoop(): Promise<void> {
);
} else {
// No active container — enqueue for a new one
queue.enqueueMessageCheck(chatJid);
queue.enqueueMessageCheck(chatJid, group.folder);
}
}
}
@@ -455,7 +531,7 @@ function recoverPendingMessages(): void {
{ group: group.name, pendingCount: pending.length },
'Recovery: found unprocessed messages',
);
queue.enqueueMessageCheck(chatJid);
queue.enqueueMessageCheck(chatJid, group.folder);
}
}
}
@@ -472,7 +548,7 @@ async function main(): Promise<void> {
loadState();
// Start credential proxy (containers route API calls through this)
const proxyServer = await startCredentialProxy(
await startCredentialProxy(
CREDENTIAL_PROXY_PORT,
PROXY_BIND_HOST,
);
@@ -480,7 +556,6 @@ async function main(): Promise<void> {
// Graceful shutdown handlers
const shutdown = async (signal: string) => {
logger.info({ signal }, 'Shutdown signal received');
proxyServer.close();
await queue.shutdown(10000);
for (const ch of channels) await ch.disconnect();
process.exit(0);

View File

@@ -0,0 +1,214 @@
import { describe, it, expect, vi } from 'vitest';
import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed } from './session-commands.js';
import type { NewMessage } from './types.js';
import type { SessionCommandDeps } from './session-commands.js';
describe('extractSessionCommand', () => {
const trigger = /^@Andy\b/i;
it('detects bare /compact', () => {
expect(extractSessionCommand('/compact', trigger)).toBe('/compact');
});
it('detects /compact with trigger prefix', () => {
expect(extractSessionCommand('@Andy /compact', trigger)).toBe('/compact');
});
it('rejects /compact with extra text', () => {
expect(extractSessionCommand('/compact now please', trigger)).toBeNull();
});
it('rejects partial matches', () => {
expect(extractSessionCommand('/compaction', trigger)).toBeNull();
});
it('rejects regular messages', () => {
expect(extractSessionCommand('please compact the conversation', trigger)).toBeNull();
});
it('handles whitespace', () => {
expect(extractSessionCommand(' /compact ', trigger)).toBe('/compact');
});
it('is case-sensitive for the command', () => {
expect(extractSessionCommand('/Compact', trigger)).toBeNull();
});
});
describe('isSessionCommandAllowed', () => {
it('allows main group regardless of sender', () => {
expect(isSessionCommandAllowed(true, false)).toBe(true);
});
it('allows trusted/admin sender (is_from_me) in non-main group', () => {
expect(isSessionCommandAllowed(false, true)).toBe(true);
});
it('denies untrusted sender in non-main group', () => {
expect(isSessionCommandAllowed(false, false)).toBe(false);
});
it('allows trusted sender in main group', () => {
expect(isSessionCommandAllowed(true, true)).toBe(true);
});
});
function makeMsg(content: string, overrides: Partial<NewMessage> = {}): NewMessage {
return {
id: 'msg-1',
chat_jid: 'group@test',
sender: 'user@test',
sender_name: 'User',
content,
timestamp: '100',
...overrides,
};
}
function makeDeps(overrides: Partial<SessionCommandDeps> = {}): SessionCommandDeps {
return {
sendMessage: vi.fn().mockResolvedValue(undefined),
setTyping: vi.fn().mockResolvedValue(undefined),
runAgent: vi.fn().mockResolvedValue('success'),
closeStdin: vi.fn(),
advanceCursor: vi.fn(),
formatMessages: vi.fn().mockReturnValue('<formatted>'),
canSenderInteract: vi.fn().mockReturnValue(true),
...overrides,
};
}
const trigger = /^@Andy\b/i;
describe('handleSessionCommand', () => {
it('returns handled:false when no session command found', async () => {
const deps = makeDeps();
const result = await handleSessionCommand({
missedMessages: [makeMsg('hello')],
isMainGroup: true,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result.handled).toBe(false);
});
it('handles authorized /compact in main group', async () => {
const deps = makeDeps();
const result = await handleSessionCommand({
missedMessages: [makeMsg('/compact')],
isMainGroup: true,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.runAgent).toHaveBeenCalledWith('/compact', expect.any(Function));
expect(deps.advanceCursor).toHaveBeenCalledWith('100');
});
it('sends denial to interactable sender in non-main group', async () => {
const deps = makeDeps();
const result = await handleSessionCommand({
missedMessages: [makeMsg('/compact', { is_from_me: false })],
isMainGroup: false,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.sendMessage).toHaveBeenCalledWith('Session commands require admin access.');
expect(deps.runAgent).not.toHaveBeenCalled();
expect(deps.advanceCursor).toHaveBeenCalledWith('100');
});
it('silently consumes denied command when sender cannot interact', async () => {
const deps = makeDeps({ canSenderInteract: vi.fn().mockReturnValue(false) });
const result = await handleSessionCommand({
missedMessages: [makeMsg('/compact', { is_from_me: false })],
isMainGroup: false,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.sendMessage).not.toHaveBeenCalled();
expect(deps.advanceCursor).toHaveBeenCalledWith('100');
});
it('processes pre-compact messages before /compact', async () => {
const deps = makeDeps();
const msgs = [
makeMsg('summarize this', { timestamp: '99' }),
makeMsg('/compact', { timestamp: '100' }),
];
const result = await handleSessionCommand({
missedMessages: msgs,
isMainGroup: true,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.formatMessages).toHaveBeenCalledWith([msgs[0]], 'UTC');
// Two runAgent calls: pre-compact + /compact
expect(deps.runAgent).toHaveBeenCalledTimes(2);
expect(deps.runAgent).toHaveBeenCalledWith('<formatted>', expect.any(Function));
expect(deps.runAgent).toHaveBeenCalledWith('/compact', expect.any(Function));
});
it('allows is_from_me sender in non-main group', async () => {
const deps = makeDeps();
const result = await handleSessionCommand({
missedMessages: [makeMsg('/compact', { is_from_me: true })],
isMainGroup: false,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.runAgent).toHaveBeenCalledWith('/compact', expect.any(Function));
});
it('reports failure when command-stage runAgent returns error without streamed status', async () => {
// runAgent resolves 'error' but callback never gets status: 'error'
const deps = makeDeps({ runAgent: vi.fn().mockImplementation(async (prompt, onOutput) => {
await onOutput({ status: 'success', result: null });
return 'error';
})});
const result = await handleSessionCommand({
missedMessages: [makeMsg('/compact')],
isMainGroup: true,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.sendMessage).toHaveBeenCalledWith(expect.stringContaining('failed'));
});
it('returns success:false on pre-compact failure with no output', async () => {
const deps = makeDeps({ runAgent: vi.fn().mockResolvedValue('error') });
const msgs = [
makeMsg('summarize this', { timestamp: '99' }),
makeMsg('/compact', { timestamp: '100' }),
];
const result = await handleSessionCommand({
missedMessages: msgs,
isMainGroup: true,
groupName: 'test',
triggerPattern: trigger,
timezone: 'UTC',
deps,
});
expect(result).toEqual({ handled: true, success: false });
expect(deps.sendMessage).toHaveBeenCalledWith(expect.stringContaining('Failed to process'));
});
});

143
src/session-commands.ts Normal file
View File

@@ -0,0 +1,143 @@
import type { NewMessage } from './types.js';
import { logger } from './logger.js';
/**
* Extract a session slash command from a message, stripping the trigger prefix if present.
* Returns the slash command (e.g., '/compact') or null if not a session command.
*/
export function extractSessionCommand(content: string, triggerPattern: RegExp): string | null {
let text = content.trim();
text = text.replace(triggerPattern, '').trim();
if (text === '/compact') return '/compact';
return null;
}
/**
* Check if a session command sender is authorized.
* Allowed: main group (any sender), or trusted/admin sender (is_from_me) in any group.
*/
export function isSessionCommandAllowed(isMainGroup: boolean, isFromMe: boolean): boolean {
return isMainGroup || isFromMe;
}
/** Minimal agent result interface — matches the subset of ContainerOutput used here. */
export interface AgentResult {
status: 'success' | 'error';
result?: string | object | null;
}
/** Dependencies injected by the orchestrator. */
export interface SessionCommandDeps {
sendMessage: (text: string) => Promise<void>;
setTyping: (typing: boolean) => Promise<void>;
runAgent: (
prompt: string,
onOutput: (result: AgentResult) => Promise<void>,
) => Promise<'success' | 'error'>;
closeStdin: () => void;
advanceCursor: (timestamp: string) => void;
formatMessages: (msgs: NewMessage[], timezone: string) => string;
/** Whether the denied sender would normally be allowed to interact (for denial messages). */
canSenderInteract: (msg: NewMessage) => boolean;
}
function resultToText(result: string | object | null | undefined): string {
if (!result) return '';
const raw = typeof result === 'string' ? result : JSON.stringify(result);
return raw.replace(/<internal>[\s\S]*?<\/internal>/g, '').trim();
}
/**
* Handle session command interception in processGroupMessages.
* Scans messages for a session command, handles auth + execution.
* Returns { handled: true, success } if a command was found; { handled: false } otherwise.
* success=false means the caller should retry (cursor was not advanced).
*/
export async function handleSessionCommand(opts: {
missedMessages: NewMessage[];
isMainGroup: boolean;
groupName: string;
triggerPattern: RegExp;
timezone: string;
deps: SessionCommandDeps;
}): Promise<{ handled: false } | { handled: true; success: boolean }> {
const { missedMessages, isMainGroup, groupName, triggerPattern, timezone, deps } = opts;
const cmdMsg = missedMessages.find(
(m) => extractSessionCommand(m.content, triggerPattern) !== null,
);
const command = cmdMsg ? extractSessionCommand(cmdMsg.content, triggerPattern) : null;
if (!command || !cmdMsg) return { handled: false };
if (!isSessionCommandAllowed(isMainGroup, cmdMsg.is_from_me === true)) {
// DENIED: send denial if the sender would normally be allowed to interact,
// then silently consume the command by advancing the cursor past it.
// Trade-off: other messages in the same batch are also consumed (cursor is
// a high-water mark). Acceptable for this narrow edge case.
if (deps.canSenderInteract(cmdMsg)) {
await deps.sendMessage('Session commands require admin access.');
}
deps.advanceCursor(cmdMsg.timestamp);
return { handled: true, success: true };
}
// AUTHORIZED: process pre-compact messages first, then run the command
logger.info({ group: groupName, command }, 'Session command');
const cmdIndex = missedMessages.indexOf(cmdMsg);
const preCompactMsgs = missedMessages.slice(0, cmdIndex);
// Send pre-compact messages to the agent so they're in the session context.
if (preCompactMsgs.length > 0) {
const prePrompt = deps.formatMessages(preCompactMsgs, timezone);
let hadPreError = false;
let preOutputSent = false;
const preResult = await deps.runAgent(prePrompt, async (result) => {
if (result.status === 'error') hadPreError = true;
const text = resultToText(result.result);
if (text) {
await deps.sendMessage(text);
preOutputSent = true;
}
// Close stdin on session-update marker — emitted after query completes,
// so all results (including multi-result runs) are already written.
if (result.status === 'success' && result.result === null) {
deps.closeStdin();
}
});
if (preResult === 'error' || hadPreError) {
logger.warn({ group: groupName }, 'Pre-compact processing failed, aborting session command');
await deps.sendMessage(`Failed to process messages before ${command}. Try again.`);
if (preOutputSent) {
// Output was already sent — don't retry or it will duplicate.
// Advance cursor past pre-compact messages, leave command pending.
deps.advanceCursor(preCompactMsgs[preCompactMsgs.length - 1].timestamp);
return { handled: true, success: true };
}
return { handled: true, success: false };
}
}
// Forward the literal slash command as the prompt (no XML formatting)
await deps.setTyping(true);
let hadCmdError = false;
const cmdOutput = await deps.runAgent(command, async (result) => {
if (result.status === 'error') hadCmdError = true;
const text = resultToText(result.result);
if (text) await deps.sendMessage(text);
});
// Advance cursor to the command — messages AFTER it remain pending for next poll.
deps.advanceCursor(cmdMsg.timestamp);
await deps.setTyping(false);
if (cmdOutput === 'error' || hadCmdError) {
await deps.sendMessage(`${command} failed. The session is unchanged.`);
}
return { handled: true, success: true };
}

View File

@@ -32,6 +32,8 @@ export interface ContainerConfig {
timeout?: number; // Default: 300000 (5 minutes)
}
export type AgentType = 'claude-code' | 'codex';
export interface RegisteredGroup {
name: string;
folder: string;
@@ -40,6 +42,7 @@ export interface RegisteredGroup {
containerConfig?: ContainerConfig;
requiresTrigger?: boolean; // Default: true for groups, false for solo chats
isMain?: boolean; // True for the main control group (no trigger, elevated privileges)
agentType?: AgentType;
}
export interface NewMessage {